From cd42dd8ca7fdf466ef5fbf9f1a7e8cc3e4dc537e Mon Sep 17 00:00:00 2001 From: Hampus Date: Tue, 1 Sep 2026 00:06:33 +0200 Subject: [PATCH] fix(api): rebuild the deletion queue under its lock (#2292) --- fluxer_api/src/api/app/APILifecycle.ts | 9 +- .../src/api/app/DeletionQueueStartup.ts | 25 ++++++ .../app/tests/DeletionQueueStartup.test.ts | 86 +++++++++++++++++++ 3 files changed, 113 insertions(+), 7 deletions(-) create mode 100644 fluxer_api/src/api/app/DeletionQueueStartup.ts create mode 100644 fluxer_api/src/api/app/tests/DeletionQueueStartup.test.ts diff --git a/fluxer_api/src/api/app/APILifecycle.ts b/fluxer_api/src/api/app/APILifecycle.ts index 9097416d0..9baa77623 100644 --- a/fluxer_api/src/api/app/APILifecycle.ts +++ b/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(); - if (await kvDeletionQueue.needsRebuild()) { - logger.info('KV deletion queue needs rebuild, rebuilding...'); - await kvDeletionQueue.rebuildState(); - } else { - logger.info('KV deletion queue state is healthy'); - } + await ensureDeletionQueueState(getKVAccountDeletionQueue(), logger); } catch (error) { logger.error({error}, 'Failed to verify KV deletion queue state'); throw error; diff --git a/fluxer_api/src/api/app/DeletionQueueStartup.ts b/fluxer_api/src/api/app/DeletionQueueStartup.ts new file mode 100644 index 000000000..ca313207d --- /dev/null +++ b/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 { + 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); + } +} diff --git a/fluxer_api/src/api/app/tests/DeletionQueueStartup.test.ts b/fluxer_api/src/api/app/tests/DeletionQueueStartup.test.ts new file mode 100644 index 000000000..ed3339a15 --- /dev/null +++ b/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): 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 = []; + 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(); + }); +});