From fe500380a148696732c92f9af472a387ca2cdc49 Mon Sep 17 00:00:00 2001 From: Roy Keene Date: Mon, 31 Aug 2026 17:22:49 -0500 Subject: [PATCH 1/2] Add some more tests --- src/lib/queue/index.test.ts | 198 ++++++++++++++++++++++++++++++++++++ 1 file changed, 198 insertions(+) diff --git a/src/lib/queue/index.test.ts b/src/lib/queue/index.test.ts index a59b3821..52d2eb13 100644 --- a/src/lib/queue/index.test.ts +++ b/src/lib/queue/index.test.ts @@ -1804,6 +1804,204 @@ suite.sequential('Driver Tests', async function() { expect(completedInB?.request).toEqual({ key: 'completed-in-b-2' }); }); + /* + * Retention cleanup must free idempotent keys for deleted + * completed entries, without freeing keys still held by + * surviving entries in the same partition. + */ + testRunner('Delete Expired Completed Cleans Idempotent Keys Without Over/Under Delete', async function() { + const retentionMs = 100; + await using queueInfo = await driverConfig.create('delete-expired-completed-idempotent'); + const localQueue = queueInfo.queue; + + await using runner = new KeetaAnchorQueueRunnerJSONConfigProc<{ key: string; }, null>({ + id: 'delete-expired-completed-idempotent-runner', + queue: localQueue, + completedRetentionDays: retentionMs / 86_400_000, + processor: async function() { + return({ status: 'completed', output: null }); + } + }); + + const expiredKey1 = generateRequestID(); + const expiredKey2 = generateRequestID(); + const keptKey = generateRequestID(); + + const expiredID = await localQueue.add( + { key: 'old-completed' }, + { idempotentKeys: new Set([expiredKey1, expiredKey2]) } + ); + await localQueue.setStatus(expiredID, 'completed'); + + const keptCompletedID = await localQueue.add( + { key: 'kept-completed' }, + { idempotentKeys: new Set([keptKey]) } + ); + + await asleep(retentionMs * 2); + + await localQueue.setStatus(keptCompletedID, 'completed'); + + const pendingKey = generateRequestID(); + const pendingID = await localQueue.add( + { key: 'pending' }, + { idempotentKeys: new Set([pendingKey]) } + ); + + await runner.maintain(); + + expect(await localQueue.get(expiredID)).toBeNull(); + expect(await localQueue.get(keptCompletedID)).not.toBeNull(); + expect(await localQueue.get(pendingID)).not.toBeNull(); + + /* + * Under-delete would leave expiredKey1/expiredKey2 + * reserved and cause IdempotentExistsError here. + */ + const reusedID = await localQueue.add( + { key: 'reused-expired-keys' }, + { idempotentKeys: new Set([expiredKey1, expiredKey2]) } + ); + expect(await localQueue.get(reusedID)).not.toBeNull(); + + /* + * Over-delete would free keptKey/pendingKey and let + * these adds succeed. + */ + try { + await localQueue.add( + { key: 'should-conflict-kept' }, + { idempotentKeys: new Set([keptKey]) } + ); + expect.unreachable('Expected IdempotentExistsError for kept completed key'); + } catch (error: unknown) { + expect(Errors.IdempotentExistsError.isInstance(error)).toBe(true); + if (!Errors.IdempotentExistsError.isInstance(error)) { + throw(new Error('internal error: Error is not IdempotentExistsError')); + } + expect(error.idempotentIDsFound).toEqual(new Set([keptKey])); + } + + try { + await localQueue.add( + { key: 'should-conflict-pending' }, + { idempotentKeys: new Set([pendingKey]) } + ); + expect.unreachable('Expected IdempotentExistsError for pending key'); + } catch (error: unknown) { + expect(Errors.IdempotentExistsError.isInstance(error)).toBe(true); + if (!Errors.IdempotentExistsError.isInstance(error)) { + throw(new Error('internal error: Error is not IdempotentExistsError')); + } + expect(error.idempotentIDsFound).toEqual(new Set([pendingKey])); + } + }); + + /* + * Status-guarded delete must not free idempotent keys when + * the entry status does not match the delete target. + */ + testRunner('Delete With Mismatched Status Does Not Remove Idempotent Keys', async function() { + await using queueInfo = await driverConfig.create('delete-mismatched-status-idempotent'); + const localQueue = queueInfo.queue; + + const idempotentKey = generateRequestID(); + const entryID = await localQueue.add( + { key: 'status-guarded' }, + { idempotentKeys: new Set([idempotentKey]) } + ); + await localQueue.setStatus(entryID, 'completed'); + + await localQueue.delete([{ id: entryID, status: 'pending' }]); + + expect(await localQueue.get(entryID)).not.toBeNull(); + expect((await localQueue.get(entryID))?.status).toBe('completed'); + + try { + await localQueue.add( + { key: 'should-still-conflict' }, + { idempotentKeys: new Set([idempotentKey]) } + ); + expect.unreachable('Expected IdempotentExistsError after mismatched-status delete'); + } catch (error: unknown) { + expect(Errors.IdempotentExistsError.isInstance(error)).toBe(true); + if (!Errors.IdempotentExistsError.isInstance(error)) { + throw(new Error('internal error: Error is not IdempotentExistsError')); + } + expect(error.idempotentIDsFound).toEqual(new Set([idempotentKey])); + } + + await localQueue.delete([{ id: entryID, status: 'completed' }]); + expect(await localQueue.get(entryID)).toBeNull(); + + const reusedID = await localQueue.add( + { key: 'reused-after-matching-delete' }, + { idempotentKeys: new Set([idempotentKey]) } + ); + expect(await localQueue.get(reusedID)).not.toBeNull(); + }); + + /* + * Cleaning a completed entry in one partition must free that + * partition's idempotent keys without touching the same keys + * held by another partition. + */ + testRunner('Delete Expired Completed Idempotent Keys Are Partition Scoped', async function() { + const retentionMs = 100; + await using queueInfo = await driverConfig.create('delete-expired-completed-idempotent-partition'); + const localQueue = queueInfo.queue; + await using partitionA = await localQueue.partition('partition-a'); + await using partitionB = await localQueue.partition('partition-b'); + + await using runnerA = new KeetaAnchorQueueRunnerJSONConfigProc<{ key: string; }, null>({ + id: 'delete-expired-completed-idempotent-partition-a-runner', + queue: partitionA, + completedRetentionDays: retentionMs / 86_400_000, + processor: async function() { + return({ status: 'completed', output: null }); + } + }); + + const sharedIdempotentKey = generateRequestID(); + + const expiredAID = await partitionA.add( + { key: 'completed-in-a' }, + { idempotentKeys: new Set([sharedIdempotentKey]) } + ); + await partitionA.setStatus(expiredAID, 'completed'); + + const pendingBID = await partitionB.add( + { key: 'pending-in-b' }, + { idempotentKeys: new Set([sharedIdempotentKey]) } + ); + + await asleep(retentionMs * 2); + await runnerA.maintain(); + + expect(await partitionA.get(expiredAID)).toBeNull(); + expect(await partitionB.get(pendingBID)).not.toBeNull(); + + const reusedAID = await partitionA.add( + { key: 'reused-in-a' }, + { idempotentKeys: new Set([sharedIdempotentKey]) } + ); + expect(await partitionA.get(reusedAID)).not.toBeNull(); + + try { + await partitionB.add( + { key: 'should-conflict-in-b' }, + { idempotentKeys: new Set([sharedIdempotentKey]) } + ); + expect.unreachable('Expected IdempotentExistsError in other partition'); + } catch (error: unknown) { + expect(Errors.IdempotentExistsError.isInstance(error)).toBe(true); + if (!Errors.IdempotentExistsError.isInstance(error)) { + throw(new Error('internal error: Error is not IdempotentExistsError')); + } + expect(error.idempotentIDsFound).toEqual(new Set([sharedIdempotentKey])); + } + }); + /* Test that mutating the entry results does not affect the stored entry */ testRunner('Entry Immutability', async function() { const id = await queue.add({ key: 'immutable', nested: { value: 42 }}); From 13a1c7c192944ea2bb6c6a0a4bd72927a857a637 Mon Sep 17 00:00:00 2001 From: Roy Keene Date: Wed, 2 Sep 2026 12:17:24 -0500 Subject: [PATCH 2/2] More comments to help understand why the test is constructed in the way that it is --- src/lib/queue/index.test.ts | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/src/lib/queue/index.test.ts b/src/lib/queue/index.test.ts index 52d2eb13..c611a777 100644 --- a/src/lib/queue/index.test.ts +++ b/src/lib/queue/index.test.ts @@ -1827,12 +1827,22 @@ suite.sequential('Driver Tests', async function() { const expiredKey2 = generateRequestID(); const keptKey = generateRequestID(); + /* + * Create a queue entry at T=0 which will be expired + * when maintain is called (after `asleep()`). + */ const expiredID = await localQueue.add( { key: 'old-completed' }, { idempotentKeys: new Set([expiredKey1, expiredKey2]) } ); await localQueue.setStatus(expiredID, 'completed'); + /* + * Create a queue entry at T=0 which will NOT be + * expired when maintain is called because the + * status isn't changed to `completed` until + * T=retentionMs * 2, which is after the `asleep()` call. + */ const keptCompletedID = await localQueue.add( { key: 'kept-completed' }, { idempotentKeys: new Set([keptKey]) } @@ -1850,8 +1860,21 @@ suite.sequential('Driver Tests', async function() { await runner.maintain(); + /* + * This is expired because it was set to completed before + * the `asleep()` call, and should have been deleted by the + * retention cleanup in `runner.maintain()`. + */ expect(await localQueue.get(expiredID)).toBeNull(); + /* + * This is not expired because it was set to completed after + * the `asleep()` call, so is more recent than the retention + * period + */ expect(await localQueue.get(keptCompletedID)).not.toBeNull(); + /* + * This is not expired because it's still pending + */ expect(await localQueue.get(pendingID)).not.toBeNull(); /*