fix(api): rebuild the deletion queue under its lock (#2292)

fluxer_api/src/api/app/APILifecycle.ts

@@ -34,6 +34,7 @@ import {VisionarySlotInitializer} from '../stripe/VisionarySlotInitializer';
 import {VoiceDataInitializer} from '../voice/VoiceDataInitializer';
 import {JetStreamWorkerQueue} from '../worker/JetStreamWorkerQueue';
 import {WorkerService} from '../worker/WorkerService';
+import {ensureDeletionQueueState} from './DeletionQueueStartup';
 
 let jsConnectionManager: JetStreamConnectionManager | null = null;
 
@@ -134,13 +135,7 @@ export function createInitializer(config: APIConfig, logger: ILogger): () => Pro
 				logger.info('JetStream worker service initialized');
 			}
 			try {
-				const kvDeletionQueue = getKVAccountDeletionQueue();
+				await ensureDeletionQueueState(getKVAccountDeletionQueue(), logger);
-				if (await kvDeletionQueue.needsRebuild()) {
-					logger.info('KV deletion queue needs rebuild, rebuilding...');
-					await kvDeletionQueue.rebuildState();
-				} else {
-					logger.info('KV deletion queue state is healthy');
-				}
 			} catch (error) {
 				logger.error({error}, 'Failed to verify KV deletion queue state');
 				throw error;

fluxer_api/src/api/app/DeletionQueueStartup.ts

@@ -0,0 +1,25 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import type {ILogger} from '../ILogger';
+import type {KVAccountDeletionQueueService} from '../infrastructure/KVAccountDeletionQueueService';
+
+export async function ensureDeletionQueueState(
+	deletionQueue: KVAccountDeletionQueueService,
+	logger: ILogger,
+): Promise<void> {
+	if (!(await deletionQueue.needsRebuild())) {
+		logger.info('KV deletion queue state is healthy');
+		return;
+	}
+	const lockToken = await deletionQueue.acquireRebuildLock();
+	if (!lockToken) {
+		logger.info('Another instance is rebuilding the KV deletion queue, skipping');
+		return;
+	}
+	logger.info('KV deletion queue needs rebuild, rebuilding...');
+	try {
+		await deletionQueue.rebuildState(lockToken);
+	} finally {
+		await deletionQueue.releaseRebuildLock(lockToken);
+	}
+}

fluxer_api/src/api/app/tests/DeletionQueueStartup.test.ts

@@ -0,0 +1,86 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import {describe, expect, it} from 'vitest';
+import {createUserID} from '../../BrandedTypes';
+import {KVAccountDeletionQueueService} from '../../infrastructure/KVAccountDeletionQueueService';
+import type {User} from '../../models/User';
+import {MockKVProvider} from '../../test/mocks/MockKVProvider';
+import {NoopLogger} from '../../test/mocks/NoopLogger';
+import type {UserRepository} from '../../user/repositories/UserRepository';
+import {ensureDeletionQueueState} from '../DeletionQueueStartup';
+
+const WORKER_PAGE_COUNT = 2;
+
+function createPendingUser(index: number): User {
+	return {
+		id: createUserID(BigInt(9100 + index)),
+		pendingDeletionAt: new Date('2026-06-01T00:00:00.000Z'),
+		deletionReasonCode: 0,
+	} as unknown as User;
+}
+
+function createWorkerRepository(onPageStart: (page: number) => Promise<void>): UserRepository {
+	let page = 0;
+	return {
+		async scanAllUsersPage() {
+			page += 1;
+			await onPageStart(page);
+			return {
+				users: [createPendingUser(page)],
+				pageState: page < WORKER_PAGE_COUNT ? `page-${page}` : null,
+			};
+		},
+	} as unknown as UserRepository;
+}
+
+describe('ensureDeletionQueueState', () => {
+	it('leaves a rebuild owned by another instance alone', async () => {
+		const kvClient = new MockKVProvider();
+		let apiScans = 0;
+		const apiRepository = {
+			async scanAllUsersPage() {
+				apiScans += 1;
+				throw new Error('api pod scan failed');
+			},
+		} as unknown as UserRepository;
+		const apiQueue = new KVAccountDeletionQueueService(kvClient, apiRepository);
+		const apiFailures: Array<unknown> = [];
+		const workerQueue = new KVAccountDeletionQueueService(
+			kvClient,
+			createWorkerRepository(async (page) => {
+				if (page !== WORKER_PAGE_COUNT) {
+					return;
+				}
+				try {
+					await ensureDeletionQueueState(apiQueue, new NoopLogger());
+				} catch (error) {
+					apiFailures.push(error);
+				}
+			}),
+		);
+
+		const workerToken = await workerQueue.acquireRebuildLock();
+		expect(workerToken).not.toBeNull();
+		await workerQueue.rebuildState(workerToken);
+		expect(await workerQueue.releaseRebuildLock(workerToken!)).toBe(true);
+
+		const queued = await workerQueue.getReadyDeletions(Date.parse('2026-06-02T00:00:00.000Z'), 10);
+		expect(queued.map((deletion) => deletion.userId).sort()).toEqual([9101n, 9102n]);
+		expect(apiScans).toBe(0);
+		expect(apiFailures).toEqual([]);
+	});
+
+	it('rebuilds under the lock when no other instance holds it', async () => {
+		const kvClient = new MockKVProvider();
+		const queue = new KVAccountDeletionQueueService(
+			kvClient,
+			createWorkerRepository(async () => {}),
+		);
+
+		await ensureDeletionQueueState(queue, new NoopLogger());
+
+		const queued = await queue.getReadyDeletions(Date.parse('2026-06-02T00:00:00.000Z'), 10);
+		expect(queued.map((deletion) => deletion.userId).sort()).toEqual([9101n, 9102n]);
+		expect(await queue.acquireRebuildLock()).not.toBeNull();
+	});
+});