Skip to content

Commit 234412a

Browse files
authored
[core] Add Promise.race([sleep, hook]) prefix determinism tests (vercel#2125)
Translates the diagrammed pattern into 5 prefix-replay tests that run the same workflow against progressively longer prefixes of a 5-event log (hook_created, wait_created, hook_received A, wait_completed, hook_received B). Each test asserts the consumer takes the same deterministic path: suspending at the right intermediate point with the right invocationsQueue state, or completing with race winners [hookA, sleep]. The full-log test verifies that the trailing hook_received B is consumed by the dangling race-2 hook awaiter without producing an unconsumed-event error. Runs in both sync and async deserialization modes via the existing defineTests harness.
1 parent b0d0561 commit 234412a

2 files changed

Lines changed: 247 additions & 0 deletions

File tree

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,2 @@
1+
---
2+
---

packages/core/src/hook-sleep-interaction.test.ts

Lines changed: 245 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -914,6 +914,251 @@ function defineTests(mode: 'sync' | 'async') {
914914
);
915915
});
916916
});
917+
918+
// ─── Prefix replay determinism: Promise.race([sleep, hook]) twice ────
919+
//
920+
// Pattern from the diagram:
921+
//
922+
// const s = sleep('1d'); // wait_created
923+
// const r1 = Promise.race([s, hook]); // hook_received A → hookA wins
924+
// // "winner: hookA"
925+
// // wait_completed (s resolves)
926+
// const r2 = Promise.race([s, hook]); // hook_received B → sleep wins
927+
// // (s already resolved, beats new hook await)
928+
// // "winner: sleep"
929+
//
930+
// Full event log (in order):
931+
// evnt_0: hook_created
932+
// evnt_1: wait_created
933+
// evnt_2: hook_received A
934+
// evnt_3: wait_completed
935+
// evnt_4: hook_received B
936+
//
937+
// We assert that the consumer goes down the same deterministic path no
938+
// matter where replay stops: at every prefix, the workflow either suspends
939+
// cleanly at the right point with the right invocationsQueue state, or
940+
// completes with the right race winners. No prefix should ever produce an
941+
// unconsumed-event error.
942+
describe(`Promise.race([sleep, hook]) prefix determinism ${label}`, () => {
943+
type RaceResult = { kind: 'hook'; value: unknown } | { kind: 'sleep' };
944+
945+
// Build the canonical 5-event log used by every prefix test. Helper takes
946+
// pre-dehydrated payloads so each test can construct them once.
947+
function buildFullEventLog(payloadA: unknown, payloadB: unknown): Event[] {
948+
return [
949+
{
950+
eventId: 'evnt_0',
951+
runId: 'wrun_test',
952+
eventType: 'hook_created',
953+
correlationId: `hook_${CORR_IDS[0]}`,
954+
eventData: {
955+
token: 'test-token',
956+
isWebhook: false,
957+
},
958+
createdAt: new Date(),
959+
},
960+
{
961+
eventId: 'evnt_1',
962+
runId: 'wrun_test',
963+
eventType: 'wait_created',
964+
correlationId: `wait_${CORR_IDS[1]}`,
965+
eventData: { resumeAt: new Date('2099-01-01') },
966+
createdAt: new Date(),
967+
},
968+
{
969+
eventId: 'evnt_2',
970+
runId: 'wrun_test',
971+
eventType: 'hook_received',
972+
correlationId: `hook_${CORR_IDS[0]}`,
973+
eventData: {
974+
token: 'test-token',
975+
payload: payloadA,
976+
},
977+
createdAt: new Date(),
978+
},
979+
{
980+
eventId: 'evnt_3',
981+
runId: 'wrun_test',
982+
eventType: 'wait_completed',
983+
correlationId: `wait_${CORR_IDS[1]}`,
984+
eventData: { resumeAt: new Date('2099-01-01') },
985+
createdAt: new Date(),
986+
},
987+
{
988+
eventId: 'evnt_4',
989+
runId: 'wrun_test',
990+
eventType: 'hook_received',
991+
correlationId: `hook_${CORR_IDS[0]}`,
992+
eventData: {
993+
token: 'test-token',
994+
payload: payloadB,
995+
},
996+
createdAt: new Date(),
997+
},
998+
];
999+
}
1000+
1001+
// The workflow body is identical across every prefix test. Returned
1002+
// results are wrapped in discriminated unions so the test can tell hook
1003+
// and sleep winners apart.
1004+
function makeWorkflowFn(ctx: WorkflowOrchestratorContext) {
1005+
const createHook = createCreateHook(ctx);
1006+
const sleep = createSleep(ctx);
1007+
1008+
return async () => {
1009+
const hook = createHook({ token: 'test-token' });
1010+
const s = sleep('1d');
1011+
1012+
const r1: RaceResult = await Promise.race([
1013+
s.then(() => ({ kind: 'sleep' as const })),
1014+
(hook as Promise<unknown>).then((value) => ({
1015+
kind: 'hook' as const,
1016+
value,
1017+
})),
1018+
]);
1019+
1020+
const r2: RaceResult = await Promise.race([
1021+
s.then(() => ({ kind: 'sleep' as const })),
1022+
(hook as Promise<unknown>).then((value) => ({
1023+
kind: 'hook' as const,
1024+
value,
1025+
})),
1026+
]);
1027+
1028+
return [r1, r2];
1029+
};
1030+
}
1031+
1032+
async function buildPayloads() {
1033+
const ops: Promise<any>[] = [];
1034+
const [payloadA, payloadB] = await Promise.all([
1035+
dehydrateStepReturnValue('A', 'wrun_test', undefined, ops),
1036+
dehydrateStepReturnValue('B', 'wrun_test', undefined, ops),
1037+
]);
1038+
return { payloadA, payloadB };
1039+
}
1040+
1041+
it('prefix [hook_created]: registers hook, then suspends with wait+hook pending', async () => {
1042+
await setupHydrateMock();
1043+
const { payloadA, payloadB } = await buildPayloads();
1044+
const fullLog = buildFullEventLog(payloadA, payloadB);
1045+
const ctx = setupWorkflowContext(fullLog.slice(0, 1));
1046+
1047+
const { error } = await runWithDiscontinuation(ctx, makeWorkflowFn(ctx));
1048+
1049+
expect(error).toBeDefined();
1050+
expect(WorkflowSuspension.is(error)).toBe(true);
1051+
1052+
// The wait was created in user code (sleep('1d')) but never saw its
1053+
// wait_created event — it sits in invocationsQueue without
1054+
// hasCreatedEvent set. The hook is registered but isn't an
1055+
// invocationsQueue entry (hooks are only queued when an awaiter is
1056+
// pending in some implementations — here, no hook payload arrives so
1057+
// the queue snapshot just shows the wait).
1058+
const pendingWaits = [...ctx.invocationsQueue.values()].filter(
1059+
(i) => i.type === 'wait'
1060+
);
1061+
expect(pendingWaits).toHaveLength(1);
1062+
expect(
1063+
pendingWaits[0].type === 'wait' && pendingWaits[0].hasCreatedEvent
1064+
).toBeFalsy();
1065+
});
1066+
1067+
it('prefix [hook_created, wait_created]: registers wait too, then suspends with neither race resolved', async () => {
1068+
await setupHydrateMock();
1069+
const { payloadA, payloadB } = await buildPayloads();
1070+
const fullLog = buildFullEventLog(payloadA, payloadB);
1071+
const ctx = setupWorkflowContext(fullLog.slice(0, 2));
1072+
1073+
const { error } = await runWithDiscontinuation(ctx, makeWorkflowFn(ctx));
1074+
1075+
expect(error).toBeDefined();
1076+
expect(WorkflowSuspension.is(error)).toBe(true);
1077+
1078+
// The wait_created was consumed: the wait item should now be flagged.
1079+
const pendingWaits = [...ctx.invocationsQueue.values()].filter(
1080+
(i) => i.type === 'wait'
1081+
);
1082+
expect(pendingWaits).toHaveLength(1);
1083+
expect(
1084+
pendingWaits[0].type === 'wait' && pendingWaits[0].hasCreatedEvent
1085+
).toBe(true);
1086+
});
1087+
1088+
it('prefix [..., hook_received A]: race1 resolves with hookA, then suspends before race2 can resolve', async () => {
1089+
await setupHydrateMock();
1090+
const { payloadA, payloadB } = await buildPayloads();
1091+
const fullLog = buildFullEventLog(payloadA, payloadB);
1092+
const ctx = setupWorkflowContext(fullLog.slice(0, 3));
1093+
1094+
const { error } = await runWithDiscontinuation(ctx, makeWorkflowFn(ctx));
1095+
1096+
// Race 1 resolved with hookA, race 2 is now awaiting both s (not
1097+
// resolved — no wait_completed) and a fresh `await hook` (no more
1098+
// hook_received). With nothing left, the workflow must suspend.
1099+
expect(error).toBeDefined();
1100+
expect(WorkflowSuspension.is(error)).toBe(true);
1101+
1102+
// Wait should still be in the queue with hasCreatedEvent === true.
1103+
const pendingWaits = [...ctx.invocationsQueue.values()].filter(
1104+
(i) => i.type === 'wait'
1105+
);
1106+
expect(pendingWaits).toHaveLength(1);
1107+
expect(
1108+
pendingWaits[0].type === 'wait' && pendingWaits[0].hasCreatedEvent
1109+
).toBe(true);
1110+
});
1111+
1112+
it('prefix [..., wait_completed]: race1 = hookA, race2 = sleep, workflow returns cleanly', async () => {
1113+
await setupHydrateMock();
1114+
const { payloadA, payloadB } = await buildPayloads();
1115+
const fullLog = buildFullEventLog(payloadA, payloadB);
1116+
const ctx = setupWorkflowContext(fullLog.slice(0, 4));
1117+
1118+
const { result, error } = await runWithDiscontinuation(
1119+
ctx,
1120+
makeWorkflowFn(ctx)
1121+
);
1122+
1123+
// No error: race 1 resolves with hookA (hook_received A), then s
1124+
// resolves (wait_completed), and race 2's fresh `await hook` is beaten
1125+
// by the already-resolved s — so sleep wins race 2.
1126+
expect(error).toBeUndefined();
1127+
expect(result).toEqual([{ kind: 'hook', value: 'A' }, { kind: 'sleep' }]);
1128+
1129+
// After wait_completed, the wait is removed from invocationsQueue.
1130+
const pendingWaits = [...ctx.invocationsQueue.values()].filter(
1131+
(i) => i.type === 'wait'
1132+
);
1133+
expect(pendingWaits).toHaveLength(0);
1134+
});
1135+
1136+
it('full event log [..., hook_received B]: race1 = hookA, race2 = sleep; B is consumed by the still-subscribed hook awaiter without orphan error', async () => {
1137+
await setupHydrateMock();
1138+
const { payloadA, payloadB } = await buildPayloads();
1139+
const fullLog = buildFullEventLog(payloadA, payloadB);
1140+
const ctx = setupWorkflowContext(fullLog);
1141+
1142+
const { result, error } = await runWithDiscontinuation(
1143+
ctx,
1144+
makeWorkflowFn(ctx)
1145+
);
1146+
1147+
// The race outcome must match the 4-event prefix exactly — adding
1148+
// hook_received B at the end must not change the deterministic path
1149+
// the workflow takes.
1150+
expect(error).toBeUndefined();
1151+
expect(result).toEqual([{ kind: 'hook', value: 'A' }, { kind: 'sleep' }]);
1152+
1153+
// The dangling hook awaiter (the loser of race 2) is still subscribed
1154+
// when hook_received B arrives, so the event is consumed and no
1155+
// unconsumed-event error fires.
1156+
const pendingWaits = [...ctx.invocationsQueue.values()].filter(
1157+
(i) => i.type === 'wait'
1158+
);
1159+
expect(pendingWaits).toHaveLength(0);
1160+
});
1161+
});
9171162
}
9181163

9191164
// ─── Run tests in both modes ────────────────────────────

0 commit comments

Comments
 (0)