diff --git a/packages/core/src/runtime/wait-continuation.test.ts b/packages/core/src/runtime/wait-continuation.test.ts index 3d79cd5b91..25ac6fe594 100644 --- a/packages/core/src/runtime/wait-continuation.test.ts +++ b/packages/core/src/runtime/wait-continuation.test.ts @@ -9,18 +9,21 @@ const CORR_ID = 'wait_01ABC'; const NOW = new Date('2026-05-19T12:00:20.500Z').getTime(); describe('getWaitContinuationDispatch', () => { - describe('mid-range waits (bare correlationId key)', () => { - it('uses the bare correlationId so re-observations dedupe', () => { - expect(getWaitContinuationDispatch(60, CORR_ID, NOW)).toEqual({ - delaySeconds: 60, - idempotencyKey: CORR_ID, - }); + describe('mid-range waits', () => { + it('keeps the delay and prefixes the key with the correlationId', () => { + const { delaySeconds, idempotencyKey } = getWaitContinuationDispatch( + 60, + CORR_ID, + NOW + ); + expect(delaySeconds).toBe(60); + expect(idempotencyKey.startsWith(`${CORR_ID}:`)).toBe(true); }); - it('is stable across suspension passes targeting the same deadline', () => { + it('PROBE: does NOT dedupe across suspension passes', () => { const pass1 = getWaitContinuationDispatch(60, CORR_ID, NOW); const pass2 = getWaitContinuationDispatch(45, CORR_ID, NOW + 15_000); - expect(pass2.idempotencyKey).toBe(pass1.idempotencyKey); + expect(pass2.idempotencyKey).not.toBe(pass1.idempotencyKey); }); it('covers the full band up to the max delay', () => { @@ -34,35 +37,28 @@ describe('getWaitContinuationDispatch', () => { CORR_ID, NOW ); - expect(low.idempotencyKey).toBe(CORR_ID); - expect(high).toEqual({ - delaySeconds: WAIT_CONTINUATION_MAX_DELAY_SECONDS, - idempotencyKey: CORR_ID, - }); + expect(low.idempotencyKey.startsWith(`${CORR_ID}:`)).toBe(true); + expect(high.delaySeconds).toBe(WAIT_CONTINUATION_MAX_DELAY_SECONDS); + expect(high.idempotencyKey.startsWith(`${CORR_ID}:`)).toBe(true); }); }); - describe('near-elapsed waits (second-bucketed key)', () => { - it('suffixes the key with the current epoch second', () => { - expect( - getWaitContinuationDispatch( - NEAR_ELAPSED_WAIT_THRESHOLD_SECONDS, - CORR_ID, - NOW - ) - ).toEqual({ - delaySeconds: NEAR_ELAPSED_WAIT_THRESHOLD_SECONDS, - idempotencyKey: `${CORR_ID}:${Math.floor(NOW / 1000)}`, - }); + describe('near-elapsed waits', () => { + it('keeps the full remaining time as the delay', () => { + const { delaySeconds, idempotencyKey } = getWaitContinuationDispatch( + NEAR_ELAPSED_WAIT_THRESHOLD_SECONDS, + CORR_ID, + NOW + ); + expect(delaySeconds).toBe(NEAR_ELAPSED_WAIT_THRESHOLD_SECONDS); + expect(idempotencyKey.startsWith(`${CORR_ID}:`)).toBe(true); }); - it('collapses same-second duplicates but frees the key for a later retry', () => { + it('PROBE: does NOT collapse same-second duplicates', () => { const first = getWaitContinuationDispatch(1, CORR_ID, NOW); const sameSecond = getWaitContinuationDispatch(1, CORR_ID, NOW + 400); - // A retry can only be enqueued after the >= 1s delay of the first - // message, which guarantees a later epoch-second bucket. const retry = getWaitContinuationDispatch(1, CORR_ID, NOW + 1000); - expect(sameSecond.idempotencyKey).toBe(first.idempotencyKey); + expect(sameSecond.idempotencyKey).not.toBe(first.idempotencyKey); expect(retry.idempotencyKey).not.toBe(first.idempotencyKey); }); }); @@ -70,21 +66,24 @@ describe('getWaitContinuationDispatch', () => { describe('waits beyond the max delay (chained hops)', () => { const SEVEN_DAYS = 7 * 24 * 3600; // 604800s > 7 * MAX_DELAY (579600s) - it('clamps the delay to the max and suffixes the key with the hop index', () => { - expect(getWaitContinuationDispatch(SEVEN_DAYS, CORR_ID, NOW)).toEqual({ - delaySeconds: WAIT_CONTINUATION_MAX_DELAY_SECONDS, - idempotencyKey: `${CORR_ID}:hop-8`, - }); + it('clamps the delay to the max and retains the hop index in the key', () => { + const { delaySeconds, idempotencyKey } = getWaitContinuationDispatch( + SEVEN_DAYS, + CORR_ID, + NOW + ); + expect(delaySeconds).toBe(WAIT_CONTINUATION_MAX_DELAY_SECONDS); + expect(idempotencyKey.startsWith(`${CORR_ID}:hop-8:`)).toBe(true); }); - it('keeps the key stable for re-observations within the same hop window', () => { + it('PROBE: does NOT dedupe re-observations within one hop window', () => { const pass1 = getWaitContinuationDispatch(SEVEN_DAYS, CORR_ID, NOW); const pass2 = getWaitContinuationDispatch( SEVEN_DAYS - 3600, CORR_ID, NOW + 3600_000 ); - expect(pass2.idempotencyKey).toBe(pass1.idempotencyKey); + expect(pass2.idempotencyKey).not.toBe(pass1.idempotencyKey); }); it('produces a fresh key at each hop delivery so the chain advances', () => { @@ -107,7 +106,7 @@ describe('getWaitContinuationDispatch', () => { expect(new Set(keys).size).toBe(keys.length); // 604800s chains as 7 max-delay hops + 1 remainder hop. expect(keys).toHaveLength(8); - expect(keys[keys.length - 1]).toBe(CORR_ID); + expect(keys[keys.length - 1]?.startsWith(`${CORR_ID}:`)).toBe(true); }); it('uses a fresh key when the final partial hop lands in the near-elapsed band', () => { @@ -118,7 +117,7 @@ describe('getWaitContinuationDispatch', () => { CORR_ID, NOW + SEVEN_DAYS * 1000 ); - expect(nearEnd.idempotencyKey).toMatch(new RegExp(`^${CORR_ID}:\\d+$`)); + expect(nearEnd.idempotencyKey.startsWith(`${CORR_ID}:`)).toBe(true); }); }); diff --git a/packages/core/src/runtime/wait-continuation.ts b/packages/core/src/runtime/wait-continuation.ts index 59b3765c06..02ae2e0036 100644 --- a/packages/core/src/runtime/wait-continuation.ts +++ b/packages/core/src/runtime/wait-continuation.ts @@ -1,6 +1,12 @@ /** * Wait-continuation dispatch: delay + idempotency-key selection. * + * !! PROBE BRANCH — NOT FOR MERGE !! + * Dedupe is disabled: every key gets a unique suffix, so no enqueue is ever + * collapsed. The rest of this comment describes the behaviour this branch + * deliberately removes; read it as the rationale under test, not as what the + * code now does. See `uniqueSuffix` below. + * * When V2 suspension processing observes a pending wait, it enqueues a * delayed "continuation" message that fires once the wait elapses and * drives the next replay (which completes the wait via the "complete @@ -101,6 +107,22 @@ export interface WaitContinuationDispatch { * (floored at 1s by the suspension handler); `waitCorrelationId` * identifies the wait so repeated suspension passes dedupe. */ +/** + * PROBE (not for merge): makes every continuation key unique, disabling + * dedupe. + * + * A key is still attached because its absence is not neutral — some worlds + * (world-postgres) serialize key-less workflow messages per run, which would + * park the continuation behind the handler's own inline step execution. This + * keeps a key and only removes its collapsing property. + * + * The counter covers two enqueues within the same millisecond in one process; + * `Math.random` covers concurrent processes. + */ +let probeCounter = 0; +const uniqueSuffix = (now: number): string => + `${now}-${(probeCounter++).toString(36)}-${Math.random().toString(36).slice(2, 10)}`; + export function getWaitContinuationDispatch( timeoutSeconds: number, waitCorrelationId: string, @@ -120,7 +142,7 @@ export function getWaitContinuationDispatch( if (timeoutSeconds <= nearElapsedThreshold) { return { delaySeconds: timeoutSeconds, - idempotencyKey: `${waitCorrelationId}:${Math.floor(now / 1000)}`, + idempotencyKey: `${waitCorrelationId}:${uniqueSuffix(now)}`, }; } @@ -128,6 +150,8 @@ export function getWaitContinuationDispatch( return { delaySeconds: Math.min(timeoutSeconds, maxDelaySeconds), idempotencyKey: - hop === 1 ? waitCorrelationId : `${waitCorrelationId}:hop-${hop}`, + hop === 1 + ? `${waitCorrelationId}:${uniqueSuffix(now)}` + : `${waitCorrelationId}:hop-${hop}:${uniqueSuffix(now)}`, }; }