diff --git a/.changeset/cyan-impalas-drop.md b/.changeset/cyan-impalas-drop.md new file mode 100644 index 0000000000..a845151cc8 --- /dev/null +++ b/.changeset/cyan-impalas-drop.md @@ -0,0 +1,2 @@ +--- +--- diff --git a/packages/core/src/runtime.test.ts b/packages/core/src/runtime.test.ts index 8b79c9a138..728b6780a5 100644 --- a/packages/core/src/runtime.test.ts +++ b/packages/core/src/runtime.test.ts @@ -879,9 +879,14 @@ describe('workflowEntrypoint step-dispatch ack ordering', () => { // Wait until the dispatch send has started (the handler has replayed, // created step_created, and entered the blocked queue() send), then assert // the handler has NOT resolved while the send is still in flight. - await vi.waitFor(() => { - expect(order).toContain('queue_dispatch_start'); - }); + // The full VM replay leading up to the send can take well over + // vi.waitFor's default 1s timeout on slow CI runners (notably Windows). + await vi.waitFor( + () => { + expect(order).toContain('queue_dispatch_start'); + }, + { timeout: 15_000 } + ); // Flush microtasks so any (incorrect) early resolution would be observable. await new Promise((r) => setTimeout(r, 20)); diff --git a/packages/core/src/workflow.test.ts b/packages/core/src/workflow.test.ts index f5954a6377..eccb9fc2ef 100644 --- a/packages/core/src/workflow.test.ts +++ b/packages/core/src/workflow.test.ts @@ -400,107 +400,113 @@ describe('runWorkflow', () => { ]); }); - it('should repeatedly replay a recorded step branch without looking ahead through Date', async () => { - const ops: Promise[] = []; - const workflowRunId = 'wrun_123'; - const workflowRun: WorkflowRun = { - runId: workflowRunId, - workflowName: 'workflow', - status: 'running', - input: await dehydrateWorkflowArguments( - [], - workflowRunId, - noEncryptionKey, - ops - ), - createdAt: new Date('2024-01-01T00:00:00.000Z'), - updatedAt: new Date('2024-01-01T00:00:00.000Z'), - startedAt: new Date('2024-01-01T00:00:00.000Z'), - deploymentId: 'test-deployment', - }; - - const startStepId = 'step_01HK153X00SP082GGA0AAJC6PJ'; - const branchStepId = 'step_01HK153X00SP082GGA0AAJC6PK'; - const events: Event[] = [ - { - eventId: 'event-run-created', + // 200 batched VM replays consume several CPU-seconds even on fast hardware; + // on 2-vCPU CI runners (notably Windows) competing with parallel test + // workers, the default 60s timeout is regularly exceeded. + it( + 'should repeatedly replay a recorded step branch without looking ahead through Date', + { timeout: 180_000 }, + async () => { + const ops: Promise[] = []; + const workflowRunId = 'wrun_123'; + const workflowRun: WorkflowRun = { runId: workflowRunId, - eventType: 'run_created', + workflowName: 'workflow', + status: 'running', + input: await dehydrateWorkflowArguments( + [], + workflowRunId, + noEncryptionKey, + ops + ), createdAt: new Date('2024-01-01T00:00:00.000Z'), - }, - { - eventId: 'event-run-started', - runId: workflowRunId, - eventType: 'run_started', - createdAt: new Date('2024-01-01T00:00:00.500Z'), - }, - { - eventId: 'event-start-created', - runId: workflowRunId, - eventType: 'step_created', - correlationId: startStepId, - eventData: { stepName: 'startQueuedLogicalRunStep' }, - createdAt: new Date('2024-01-01T00:00:01.000Z'), - }, - { - eventId: 'event-start-started', - runId: workflowRunId, - eventType: 'step_started', - correlationId: startStepId, - eventData: { stepName: 'startQueuedLogicalRunStep' }, - createdAt: new Date('2024-01-01T00:00:01.500Z'), - }, - { - eventId: 'event-start-completed', - runId: workflowRunId, - eventType: 'step_completed', - correlationId: startStepId, - eventData: { - stepName: 'startQueuedLogicalRunStep', - result: await dehydrateStepReturnValue( - 'started', - workflowRunId, - noEncryptionKey, - ops - ), + updatedAt: new Date('2024-01-01T00:00:00.000Z'), + startedAt: new Date('2024-01-01T00:00:00.000Z'), + deploymentId: 'test-deployment', + }; + + const startStepId = 'step_01HK153X00SP082GGA0AAJC6PJ'; + const branchStepId = 'step_01HK153X00SP082GGA0AAJC6PK'; + const events: Event[] = [ + { + eventId: 'event-run-created', + runId: workflowRunId, + eventType: 'run_created', + createdAt: new Date('2024-01-01T00:00:00.000Z'), }, - createdAt: new Date('2024-01-01T00:00:02.000Z'), - }, - { - eventId: 'event-drain-created', - runId: workflowRunId, - eventType: 'step_created', - correlationId: branchStepId, - eventData: { stepName: 'drainLogicalRunRuntimeStep' }, - createdAt: new Date('2024-01-01T00:00:02.500Z'), - }, - { - eventId: 'event-drain-started', - runId: workflowRunId, - eventType: 'step_started', - correlationId: branchStepId, - eventData: { stepName: 'drainLogicalRunRuntimeStep' }, - createdAt: new Date('2024-01-01T00:00:03.000Z'), - }, - { - eventId: 'event-drain-completed', - runId: workflowRunId, - eventType: 'step_completed', - correlationId: branchStepId, - eventData: { - stepName: 'drainLogicalRunRuntimeStep', - result: await dehydrateStepReturnValue( - 'drained', - workflowRunId, - noEncryptionKey, - ops - ), + { + eventId: 'event-run-started', + runId: workflowRunId, + eventType: 'run_started', + createdAt: new Date('2024-01-01T00:00:00.500Z'), }, - createdAt: new Date('2024-01-01T00:00:03.500Z'), - }, - ]; + { + eventId: 'event-start-created', + runId: workflowRunId, + eventType: 'step_created', + correlationId: startStepId, + eventData: { stepName: 'startQueuedLogicalRunStep' }, + createdAt: new Date('2024-01-01T00:00:01.000Z'), + }, + { + eventId: 'event-start-started', + runId: workflowRunId, + eventType: 'step_started', + correlationId: startStepId, + eventData: { stepName: 'startQueuedLogicalRunStep' }, + createdAt: new Date('2024-01-01T00:00:01.500Z'), + }, + { + eventId: 'event-start-completed', + runId: workflowRunId, + eventType: 'step_completed', + correlationId: startStepId, + eventData: { + stepName: 'startQueuedLogicalRunStep', + result: await dehydrateStepReturnValue( + 'started', + workflowRunId, + noEncryptionKey, + ops + ), + }, + createdAt: new Date('2024-01-01T00:00:02.000Z'), + }, + { + eventId: 'event-drain-created', + runId: workflowRunId, + eventType: 'step_created', + correlationId: branchStepId, + eventData: { stepName: 'drainLogicalRunRuntimeStep' }, + createdAt: new Date('2024-01-01T00:00:02.500Z'), + }, + { + eventId: 'event-drain-started', + runId: workflowRunId, + eventType: 'step_started', + correlationId: branchStepId, + eventData: { stepName: 'drainLogicalRunRuntimeStep' }, + createdAt: new Date('2024-01-01T00:00:03.000Z'), + }, + { + eventId: 'event-drain-completed', + runId: workflowRunId, + eventType: 'step_completed', + correlationId: branchStepId, + eventData: { + stepName: 'drainLogicalRunRuntimeStep', + result: await dehydrateStepReturnValue( + 'drained', + workflowRunId, + noEncryptionKey, + ops + ), + }, + createdAt: new Date('2024-01-01T00:00:03.500Z'), + }, + ]; - const workflowCode = `const start = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("startQueuedLogicalRunStep"); + const workflowCode = `const start = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("startQueuedLogicalRunStep"); const drain = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("drainLogicalRunRuntimeStep"); const settle = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("settleQueueActionStep"); async function workflow() { @@ -511,57 +517,58 @@ describe('runWorkflow', () => { return await settle(); }${getWorkflowTransformCode('workflow')}`; - const replayCount = 200; - const maxConcurrentReplays = 8; - const replay = async () => { - try { - const result = await runWorkflow( - workflowCode, - workflowRun, - events, - noEncryptionKey + const replayCount = 200; + const maxConcurrentReplays = 8; + const replay = async () => { + try { + const result = await runWorkflow( + workflowCode, + workflowRun, + events, + noEncryptionKey + ); + const value = await hydrateWorkflowReturnValue( + result as any, + workflowRunId, + noEncryptionKey, + ops + ); + return value === 'drained' + ? 'drained' + : `unexpected replay result: ${String(value)}`; + } catch (error) { + return error instanceof Error ? error.message : String(error); + } + }; + const outcomes: string[] = []; + for ( + let replayIndex = 0; + replayIndex < replayCount; + replayIndex += maxConcurrentReplays + ) { + const batchSize = Math.min( + maxConcurrentReplays, + replayCount - replayIndex ); - const value = await hydrateWorkflowReturnValue( - result as any, - workflowRunId, - noEncryptionKey, - ops + outcomes.push( + ...(await Promise.all(Array.from({ length: batchSize }, replay))) ); - return value === 'drained' - ? 'drained' - : `unexpected replay result: ${String(value)}`; - } catch (error) { - return error instanceof Error ? error.message : String(error); } - }; - const outcomes: string[] = []; - for ( - let replayIndex = 0; - replayIndex < replayCount; - replayIndex += maxConcurrentReplays - ) { - const batchSize = Math.min( - maxConcurrentReplays, - replayCount - replayIndex - ); - outcomes.push( - ...(await Promise.all(Array.from({ length: batchSize }, replay))) - ); + const failures = outcomes.filter((outcome) => outcome !== 'drained'); + + expect({ + replayCount, + drainedCount: outcomes.length - failures.length, + failureCount: failures.length, + firstFailure: failures[0], + }).toEqual({ + replayCount, + drainedCount: replayCount, + failureCount: 0, + firstFailure: undefined, + }); } - const failures = outcomes.filter((outcome) => outcome !== 'drained'); - - expect({ - replayCount, - drainedCount: outcomes.length - failures.length, - failureCount: failures.length, - firstFailure: failures[0], - }).toEqual({ - replayCount, - drainedCount: replayCount, - failureCount: 0, - firstFailure: undefined, - }); - }); + ); // TODO: Date.now determinism is currently broken in the workflow!! it.fails('should maintain determinism of `Date` across executions', async () => {