skills/vercel__workflow
1
0
Fork 0
mirror of https://github.com/vercel/workflow.git synced 2026-09-14 19:59:43 +08:00
Code Issues Packages Projects Releases Wiki Activity

Make sleep() a core runtime function (#40)

Browse Source 
* Handle the sleep not completing at the end of the event log

* Handle multiple timers being created in the same suspension

* Better `buildWorkflowSuspensionMessage()`

* New sleep unit tests

---------

Co-authored-by: Nathan Rajlich <n@n8.io>
This commit is contained in:
Pranay Prakash
2025-10-30 15:31:25 -07:00
committed by GitHub
parent 20d51f0d7f
commit 70be894ad1
15 changed files with 712 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/spotty-eggs-sip.md
+7
View File
@@ -0,0 +1,7 @@
---
"workflow": patch
"@workflow/world": patch
"@workflow/core": patch
---
Implement `sleep()` natively into the workflow runtime
CLAUDE.md
+26 -2
View File
@@ -64,8 +64,32 @@ cd packages/core && pnpm test
# Test specific file
cd packages/core && pnpm vitest run src/[filename].test.ts
# Run E2E tests
cd packages/core && pnpm test:e2e
# Run E2E tests (requires environment variables and running dev server)
# Note: Use nextjs-turbopack for local e2e testing (not example app - it has no dev server)
# Step 1: Start the dev server in background
cd workbench/nextjs-turbopack && pnpm dev > /tmp/nextjs-dev.log 2>&1 &
# Step 2: Wait for server to be ready (usually 15-20 seconds)
sleep 15
# Step 3: Run the e2e tests from the project root
DEPLOYMENT_URL="http://localhost:3000" APP_NAME="nextjs-turbopack" pnpm vitest run packages/core/e2e/e2e.test.ts
# Step 4: Stop the dev server when done
pkill -f "pnpm dev"
# To run specific tests, use the -t flag:
DEPLOYMENT_URL="http://localhost:3000" APP_NAME="nextjs-turbopack" pnpm vitest run packages/core/e2e/e2e.test.ts -t "sleeping"
# For production testing against deployed Vercel app:
# See .github/workflows/tests.yml for required environment variables:
# - DEPLOYMENT_URL: URL of deployed app
# - APP_NAME: App name (example, nextjs-turbopack, nextjs-webpack, nitro)
# - WORKFLOW_VERCEL_ENV: Environment (production or preview)
# - WORKFLOW_VERCEL_AUTH_TOKEN: Vercel auth token
# - WORKFLOW_VERCEL_TEAM: Vercel team ID
# - WORKFLOW_VERCEL_PROJECT: Vercel project ID
```
### Example App Development
packages/core/src/global.ts
+20 -2
View File
@@ -14,7 +14,17 @@ export interface HookInvocationQueueItem {
metadata?: Serializable;
}
export type QueueItem = StepInvocationQueueItem | HookInvocationQueueItem;
export interface WaitInvocationQueueItem {
type: 'wait';
correlationId: string;
resumeAt: Date;
hasCreatedEvent?: boolean;
}
export type QueueItem =
| StepInvocationQueueItem
| HookInvocationQueueItem
| WaitInvocationQueueItem;
/**
* An error that is thrown when one or more operations (steps/hooks/etc.) are called but do
@@ -27,10 +37,12 @@ export class WorkflowSuspension extends Error {
globalThis: typeof globalThis;
stepCount: number;
hookCount: number;
waitCount: number;
constructor(steps: QueueItem[], global: typeof globalThis) {
const stepCount = steps.filter((s) => s.type === 'step').length;
const hookCount = steps.filter((s) => s.type === 'hook').length;
const waitCount = steps.filter((s) => s.type === 'wait').length;
// Build description parts
const parts: string[] = [];
@@ -40,15 +52,20 @@ export class WorkflowSuspension extends Error {
if (hookCount > 0) {
parts.push(`${hookCount} ${hookCount === 1 ? 'hook' : 'hooks'}`);
}
if (waitCount > 0) {
parts.push(`${waitCount} ${waitCount === 1 ? 'wait' : 'waits'}`);
}
// Determine verb (has/have) and action (run/created/received)
const totalCount = stepCount + hookCount;
const totalCount = stepCount + hookCount + waitCount;
const hasOrHave = totalCount === 1 ? 'has' : 'have';
let action: string;
if (stepCount > 0) {
action = 'run';
} else if (hookCount > 0) {
action = 'created';
} else if (waitCount > 0) {
action = 'created';
} else {
action = 'received';
}
@@ -63,6 +80,7 @@ export class WorkflowSuspension extends Error {
this.globalThis = global;
this.stepCount = stepCount;
this.hookCount = hookCount;
this.waitCount = waitCount;
}
static is(value: unknown): value is WorkflowSuspension {
packages/core/src/index.ts
+1
View File
@@ -33,4 +33,5 @@ export {
getWorkflowMetadata,
type WorkflowMetadata,
} from './step/get-workflow-metadata.js';
export { sleep } from './sleep.js';
export { getWritable } from './writable-stream.js';
packages/core/src/runtime.ts
+69 -1
View File
@@ -336,6 +336,29 @@ export function workflowEntrypoint(workflowCode: string) {
// Load all events into memory before running
const events = await getAllWorkflowRunEvents(workflowRun.runId);
// Check for any elapsed waits and create wait_completed events
const now = Date.now();
for (const event of events) {
if (event.eventType === 'wait_created') {
const resumeAt = event.eventData.resumeAt as Date;
const hasCompleted = events.some(
(e) =>
e.eventType === 'wait_completed' &&
e.correlationId === event.correlationId
);
// If wait has elapsed and hasn't been completed yet
if (!hasCompleted && now >= resumeAt.getTime()) {
const completedEvent = await world.events.create(runId, {
eventType: 'wait_completed',
correlationId: event.correlationId,
});
// Add the event to the events array so the workflow can see it
events.push(completedEvent);
}
}
}
const result = await runWorkflow(workflowCode, workflowRun, events);
// Update the workflow run with the result
@@ -353,13 +376,15 @@ export function workflowEntrypoint(workflowCode: string) {
const suspensionMessage = buildWorkflowSuspensionMessage(
runId,
err.stepCount,
err.hookCount
err.hookCount,
err.waitCount
);
if (suspensionMessage) {
// Note: suspensionMessage logged only in debug mode to avoid production noise
// console.debug(suspensionMessage);
}
// Process each operation in the queue (steps and hooks)
let minTimeoutSeconds: number | null = null;
for (const queueItem of err.steps) {
if (queueItem.type === 'step') {
// Handle step operations
@@ -441,12 +466,55 @@ export function workflowEntrypoint(workflowCode: string) {
}
throw err;
}
} else if (queueItem.type === 'wait') {
// Handle wait operations
try {
// Only create wait_created event if it hasn't been created yet
if (!queueItem.hasCreatedEvent) {
await world.events.create(runId, {
eventType: 'wait_created',
correlationId: queueItem.correlationId,
eventData: {
resumeAt: queueItem.resumeAt,
},
});
}
// Calculate how long to wait before resuming
const now = Date.now();
const resumeAtMs = queueItem.resumeAt.getTime();
const delayMs = Math.max(1000, resumeAtMs - now);
const timeoutSeconds = Math.ceil(delayMs / 1000);
// Track the minimum timeout across all waits
if (
minTimeoutSeconds === null ||
timeoutSeconds < minTimeoutSeconds
) {
minTimeoutSeconds = timeoutSeconds;
}
} catch (err) {
if (WorkflowAPIError.is(err) && err.status === 409) {
// Wait already exists, so we can skip it
console.warn(
`Wait with correlation ID "${queueItem.correlationId}" already exists, skipping: ${err.message}`
);
continue;
}
throw err;
}
}
}
span?.setAttributes({
...Attribute.WorkflowRunStatus('pending_steps'),
...Attribute.WorkflowStepsCreated(err.steps.length),
});
// If we encountered any waits, return the minimum timeout
if (minTimeoutSeconds !== null) {
return { timeoutSeconds: minTimeoutSeconds };
}
} else {
const errorName = getErrorName(err);
const errorStack = getErrorStack(err);
packages/core/src/sleep.ts
+34
View File
@@ -0,0 +1,34 @@
import type { StringValue } from 'ms';
import { WORKFLOW_SLEEP } from './symbols.js';
/**
* Sleep within a workflow for a given duration.
*
* This is a built-in runtime function that uses timer events in the event log.
*
* @param duration - The duration to sleep for, this is a string in the format
* of `"1000ms"`, `"1s"`, `"1m"`, `"1h"`, or `"1d"`.
* @overload
* @returns A promise that resolves when the sleep is complete.
*/
export async function sleep(duration: StringValue): Promise<void>;
/**
* Sleep within a workflow until a specific date.
*
* This is a built-in runtime function that uses timer events in the event log.
*
* @param date - The date to sleep until, this must be a future date.
* @overload
* @returns A promise that resolves when the sleep is complete.
*/
export async function sleep(date: Date): Promise<void>;
export async function sleep(param: StringValue | Date): Promise<void> {
// Inside the workflow VM, the sleep function is stored in the globalThis object behind a symbol
const sleepFn = (globalThis as any)[WORKFLOW_SLEEP];
if (!sleepFn) {
throw new Error('`sleep()` can only be called inside a workflow function');
}
return sleepFn(param);
}
packages/core/src/symbols.ts
+1
View File
@@ -1,5 +1,6 @@
export const WORKFLOW_USE_STEP = Symbol.for('WORKFLOW_USE_STEP');
export const WORKFLOW_CREATE_HOOK = Symbol.for('WORKFLOW_CREATE_HOOK');
export const WORKFLOW_SLEEP = Symbol.for('WORKFLOW_SLEEP');
export const WORKFLOW_CONTEXT = Symbol.for('WORKFLOW_CONTEXT');
export const WORKFLOW_GET_STREAM_ID = Symbol.for('WORKFLOW_GET_STREAM_ID');
export const STREAM_NAME_SYMBOL = Symbol.for('WORKFLOW_STREAM_NAME');
packages/core/src/util.test.ts
+61 -19
View File
@@ -5,70 +5,112 @@ describe('buildWorkflowSuspensionMessage', () => {
const runId = 'test-run-123';
it('should return null when both counts are zero', () => {
const result = buildWorkflowSuspensionMessage(runId, 0, 0);
const result = buildWorkflowSuspensionMessage(runId, 0, 0, 0);
expect(result).toBeNull();
});
it('should handle single step', () => {
const result = buildWorkflowSuspensionMessage(runId, 1, 0);
const result = buildWorkflowSuspensionMessage(runId, 1, 0, 0);
expect(result).toBe(
`[Workflows] "${runId}" - 1 step to be enqueued\n Workflow will suspend and resume when steps are created`
`[Workflows] "${runId}" - 1 step to be enqueued\n Workflow will suspend and resume when steps are completed`
);
});
it('should handle multiple steps', () => {
const result = buildWorkflowSuspensionMessage(runId, 3, 0);
const result = buildWorkflowSuspensionMessage(runId, 3, 0, 0);
expect(result).toBe(
`[Workflows] "${runId}" - 3 steps to be enqueued\n Workflow will suspend and resume when steps are created`
`[Workflows] "${runId}" - 3 steps to be enqueued\n Workflow will suspend and resume when steps are completed`
);
});
it('should handle single hook', () => {
const result = buildWorkflowSuspensionMessage(runId, 0, 1);
const result = buildWorkflowSuspensionMessage(runId, 0, 1, 0);
expect(result).toBe(
`[Workflows] "${runId}" - 1 hook to be enqueued\n Workflow will suspend and resume when steps are created and hooks are triggered`
`[Workflows] "${runId}" - 1 hook to be enqueued\n Workflow will suspend and resume when hooks are received`
);
});
it('should handle multiple hooks', () => {
const result = buildWorkflowSuspensionMessage(runId, 0, 2);
const result = buildWorkflowSuspensionMessage(runId, 0, 2, 0);
expect(result).toBe(
`[Workflows] "${runId}" - 2 hooks to be enqueued\n Workflow will suspend and resume when steps are created and hooks are triggered`
`[Workflows] "${runId}" - 2 hooks to be enqueued\n Workflow will suspend and resume when hooks are received`
);
});
it('should handle single step and single hook', () => {
const result = buildWorkflowSuspensionMessage(runId, 1, 1);
const result = buildWorkflowSuspensionMessage(runId, 1, 1, 0);
expect(result).toBe(
`[Workflows] "${runId}" - 1 step and 1 hook to be enqueued\n Workflow will suspend and resume when steps are created and hooks are triggered`
`[Workflows] "${runId}" - 1 step and 1 hook to be enqueued\n Workflow will suspend and resume when steps are completed and hooks are received`
);
});
it('should handle multiple steps and single hook', () => {
const result = buildWorkflowSuspensionMessage(runId, 5, 1);
const result = buildWorkflowSuspensionMessage(runId, 5, 1, 0);
expect(result).toBe(
`[Workflows] "${runId}" - 5 steps and 1 hook to be enqueued\n Workflow will suspend and resume when steps are created and hooks are triggered`
`[Workflows] "${runId}" - 5 steps and 1 hook to be enqueued\n Workflow will suspend and resume when steps are completed and hooks are received`
);
});
it('should handle single step and multiple hooks', () => {
const result = buildWorkflowSuspensionMessage(runId, 1, 3);
const result = buildWorkflowSuspensionMessage(runId, 1, 3, 0);
expect(result).toBe(
`[Workflows] "${runId}" - 1 step and 3 hooks to be enqueued\n Workflow will suspend and resume when steps are created and hooks are triggered`
`[Workflows] "${runId}" - 1 step and 3 hooks to be enqueued\n Workflow will suspend and resume when steps are completed and hooks are received`
);
});
it('should handle multiple steps and multiple hooks', () => {
const result = buildWorkflowSuspensionMessage(runId, 4, 2);
const result = buildWorkflowSuspensionMessage(runId, 4, 2, 0);
expect(result).toBe(
`[Workflows] "${runId}" - 4 steps and 2 hooks to be enqueued\n Workflow will suspend and resume when steps are created and hooks are triggered`
`[Workflows] "${runId}" - 4 steps and 2 hooks to be enqueued\n Workflow will suspend and resume when steps are completed and hooks are received`
);
});
it('should handle large numbers correctly', () => {
const result = buildWorkflowSuspensionMessage(runId, 100, 50);
const result = buildWorkflowSuspensionMessage(runId, 100, 50, 0);
expect(result).toBe(
`[Workflows] "${runId}" - 100 steps and 50 hooks to be enqueued\n Workflow will suspend and resume when steps are created and hooks are triggered`
`[Workflows] "${runId}" - 100 steps and 50 hooks to be enqueued\n Workflow will suspend and resume when steps are completed and hooks are received`
);
});
it('should handle single wait without steps or hooks', () => {
const result = buildWorkflowSuspensionMessage(runId, 0, 0, 1);
expect(result).toBe(
`[Workflows] "${runId}" - 1 timer to be enqueued\n Workflow will suspend and resume when timers have elapsed`
);
});
it('should handle multiple waits without steps or hooks', () => {
const result = buildWorkflowSuspensionMessage(runId, 0, 0, 2);
expect(result).toBe(
`[Workflows] "${runId}" - 2 timers to be enqueued\n Workflow will suspend and resume when timers have elapsed`
);
});
it('should handle hooks and waits without steps', () => {
const result = buildWorkflowSuspensionMessage(runId, 0, 1, 1);
expect(result).toBe(
`[Workflows] "${runId}" - 1 hook and 1 timer to be enqueued\n Workflow will suspend and resume when hooks are received and timers have elapsed`
);
});
it('should handle steps and waits without hooks', () => {
const result = buildWorkflowSuspensionMessage(runId, 1, 0, 1);
expect(result).toBe(
`[Workflows] "${runId}" - 1 step and 1 timer to be enqueued\n Workflow will suspend and resume when steps are completed and timers have elapsed`
);
});
it('should handle steps, hooks, and waits', () => {
const result = buildWorkflowSuspensionMessage(runId, 1, 1, 1);
expect(result).toBe(
`[Workflows] "${runId}" - 1 step and 1 hook and 1 timer to be enqueued\n Workflow will suspend and resume when steps are completed and hooks are received and timers have elapsed`
);
});
it('should handle multiple waits with steps and hooks', () => {
const result = buildWorkflowSuspensionMessage(runId, 2, 1, 3);
expect(result).toBe(
`[Workflows] "${runId}" - 2 steps and 1 hook and 3 timers to be enqueued\n Workflow will suspend and resume when steps are completed and hooks are received and timers have elapsed`
);
});
});
packages/core/src/util.ts
+20 -8
View File
@@ -41,18 +41,20 @@ export function once<T>(fn: () => T) {
}
/**
* Builds a workflow suspension log message based on the counts of steps and hooks.
* Builds a workflow suspension log message based on the counts of steps, hooks, and waits.
* @param runId - The workflow run ID
* @param stepCount - Number of steps to be enqueued
* @param hookCount - Number of hooks to be enqueued
* @returns The formatted log message or null if both counts are 0
* @param waitCount - Number of waits to be enqueued
* @returns The formatted log message or null if all counts are 0
*/
export function buildWorkflowSuspensionMessage(
runId: string,
stepCount: number,
hookCount: number
hookCount: number,
waitCount: number
): string | null {
if (stepCount === 0 && hookCount === 0) {
if (stepCount === 0 && hookCount === 0 && waitCount === 0) {
return null;
}
@@ -63,11 +65,21 @@ export function buildWorkflowSuspensionMessage(
if (hookCount > 0) {
parts.push(`${hookCount} ${hookCount === 1 ? 'hook' : 'hooks'}`);
}
if (waitCount > 0) {
parts.push(`${waitCount} ${waitCount === 1 ? 'timer' : 'timers'}`);
}
const resumeMsg =
hookCount > 0
? 'steps are created and hooks are triggered'
: 'steps are created';
const resumeMsgParts: string[] = [];
if (stepCount > 0) {
resumeMsgParts.push('steps are completed');
}
if (hookCount > 0) {
resumeMsgParts.push('hooks are received');
}
if (waitCount > 0) {
resumeMsgParts.push('timers have elapsed');
}
const resumeMsg = resumeMsgParts.join(' and ');
return `[Workflows] "${runId}" - ${parts.join(' and ')} to be enqueued\n Workflow will suspend and resume when ${resumeMsg}`;
}
packages/core/src/workflow.test.ts
+350 -15
View File
@@ -257,31 +257,34 @@ describe('runWorkflow', () => {
{
eventId: 'event-0',
runId: workflowRunId,
eventType: 'step_started',
correlationId: 'step_01HK153X008RT6YEW43G8QX6JX',
eventType: 'wait_created',
correlationId: 'wait_01HK153X008RT6YEW43G8QX6JX',
eventData: {
resumeAt: new Date('2024-01-01T00:00:01.000Z'),
},
createdAt: new Date('2024-01-01T00:00:01.000Z'),
},
{
eventId: 'event-1',
runId: workflowRunId,
eventType: 'step_started',
correlationId: 'step_01HK153X008RT6YEW43G8QX6JY',
eventType: 'wait_created',
correlationId: 'wait_01HK153X008RT6YEW43G8QX6JY',
eventData: {
resumeAt: new Date('2024-01-01T00:00:02.000Z'),
},
createdAt: new Date('2024-01-01T00:00:01.000Z'),
},
{
eventId: 'event-2',
runId: workflowRunId,
eventType: 'step_completed',
correlationId: 'step_01HK153X008RT6YEW43G8QX6JX',
eventData: {
result: dehydrateStepReturnValue(undefined, ops),
},
eventType: 'wait_completed',
correlationId: 'wait_01HK153X008RT6YEW43G8QX6JX',
createdAt: new Date('2024-01-01T00:00:03.000Z'),
},
];
const workflowCode = `
const sleep = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("sleep");
const sleep = globalThis[Symbol.for("WORKFLOW_SLEEP")];
async function workflow() {
await Promise.race([sleep(1), sleep(2)]);
return Date.now();
@@ -296,11 +299,8 @@ describe('runWorkflow', () => {
{
eventId: 'event-3',
runId: workflowRunId,
eventType: 'step_completed',
correlationId: 'step_01HK153X008RT6YEW43G8QX6JY',
eventData: {
result: dehydrateStepReturnValue(undefined, ops),
},
eventType: 'wait_completed',
correlationId: 'wait_01HK153X008RT6YEW43G8QX6JY',
createdAt: new Date('2024-01-01T00:00:04.000Z'),
},
]);
@@ -1891,4 +1891,339 @@ describe('runWorkflow', () => {
expect(result_obj.redirect).toEqual('follow'); // default, since not set in req1
});
});
describe('sleep', () => {
it('should suspend and resume a basic single sleep', async () => {
const ops: Promise<any>[] = [];
const workflowRunId = 'test-run-123';
const workflowRun: WorkflowRun = {
runId: workflowRunId,
workflowName: 'workflow',
status: 'running',
input: dehydrateWorkflowArguments([], 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 resumeAt = new Date('2024-01-01T00:00:05.000Z');
const events: Event[] = [
{
eventId: 'event-0',
runId: workflowRunId,
eventType: 'wait_created',
correlationId: 'wait_01HK153X008RT6YEW43G8QX6JX',
eventData: {
resumeAt,
},
createdAt: new Date('2024-01-01T00:00:00.000Z'),
},
{
eventId: 'event-1',
runId: workflowRunId,
eventType: 'wait_completed',
correlationId: 'wait_01HK153X008RT6YEW43G8QX6JX',
createdAt: new Date('2024-01-01T00:00:05.000Z'),
},
];
const result = await runWorkflow(
`const sleep = globalThis[Symbol.for("WORKFLOW_SLEEP")];
async function workflow() {
await sleep('5s');
return 'sleep completed';
}${getWorkflowTransformCode('workflow')}`,
workflowRun,
events
);
expect(hydrateWorkflowReturnValue(result as any, ops)).toEqual(
'sleep completed'
);
});
it('should throw `WorkflowSuspension` when sleep has no wait_completed event', async () => {
let error: Error | undefined;
try {
const ops: Promise<any>[] = [];
const workflowRunId = 'test-run-123';
const workflowRun: WorkflowRun = {
runId: workflowRunId,
workflowName: 'workflow',
status: 'running',
input: dehydrateWorkflowArguments([], 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 events: Event[] = [];
await runWorkflow(
`const sleep = globalThis[Symbol.for("WORKFLOW_SLEEP")];
async function workflow() {
await sleep('5s');
return 'done';
}${getWorkflowTransformCode('workflow')}`,
workflowRun,
events
);
} catch (err) {
error = err as Error;
}
assert(error);
expect(error.name).toEqual('WorkflowSuspension');
expect(error.message).toEqual('1 wait has not been created yet');
expect((error as WorkflowSuspension).steps).toHaveLength(1);
expect((error as WorkflowSuspension).steps[0].type).toEqual('wait');
});
it('should handle multiple simultaneous sleeps with Promise.all()', async () => {
const ops: Promise<any>[] = [];
const workflowRunId = 'test-run-123';
const workflowRun: WorkflowRun = {
runId: workflowRunId,
workflowName: 'workflow',
status: 'running',
input: dehydrateWorkflowArguments([], 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 events: Event[] = [
{
eventId: 'event-0',
runId: workflowRunId,
eventType: 'wait_created',
correlationId: 'wait_01HK153X008RT6YEW43G8QX6JX',
eventData: {
resumeAt: new Date('2024-01-01T00:00:02.000Z'),
},
createdAt: new Date('2024-01-01T00:00:00.000Z'),
},
{
eventId: 'event-1',
runId: workflowRunId,
eventType: 'wait_created',
correlationId: 'wait_01HK153X008RT6YEW43G8QX6JY',
eventData: {
resumeAt: new Date('2024-01-01T00:00:05.000Z'),
},
createdAt: new Date('2024-01-01T00:00:00.000Z'),
},
{
eventId: 'event-2',
runId: workflowRunId,
eventType: 'wait_completed',
correlationId: 'wait_01HK153X008RT6YEW43G8QX6JX',
createdAt: new Date('2024-01-01T00:00:02.000Z'),
},
{
eventId: 'event-3',
runId: workflowRunId,
eventType: 'wait_completed',
correlationId: 'wait_01HK153X008RT6YEW43G8QX6JY',
createdAt: new Date('2024-01-01T00:00:05.000Z'),
},
];
const result = await runWorkflow(
`const sleep = globalThis[Symbol.for("WORKFLOW_SLEEP")];
async function workflow() {
const results = await Promise.all([sleep('2s'), sleep('5s')]);
return 'all sleeps completed';
}${getWorkflowTransformCode('workflow')}`,
workflowRun,
events
);
expect(hydrateWorkflowReturnValue(result as any, ops)).toEqual(
'all sleeps completed'
);
});
it('should suspend with multiple sleeps but only one wait_completed event (partial completion)', async () => {
let error: Error | undefined;
try {
const ops: Promise<any>[] = [];
const workflowRunId = 'test-run-123';
const workflowRun: WorkflowRun = {
runId: workflowRunId,
workflowName: 'workflow',
status: 'running',
input: dehydrateWorkflowArguments([], 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 events: Event[] = [
{
eventId: 'event-0',
runId: workflowRunId,
eventType: 'wait_created',
correlationId: 'wait_01HK153X008RT6YEW43G8QX6JX',
eventData: {
resumeAt: new Date('2024-01-01T00:00:02.000Z'),
},
createdAt: new Date('2024-01-01T00:00:00.000Z'),
},
{
eventId: 'event-1',
runId: workflowRunId,
eventType: 'wait_created',
correlationId: 'wait_01HK153X008RT6YEW43G8QX6JY',
eventData: {
resumeAt: new Date('2024-01-01T00:00:05.000Z'),
},
createdAt: new Date('2024-01-01T00:00:00.000Z'),
},
{
eventId: 'event-2',
runId: workflowRunId,
eventType: 'wait_completed',
correlationId: 'wait_01HK153X008RT6YEW43G8QX6JX',
createdAt: new Date('2024-01-01T00:00:02.000Z'),
},
];
await runWorkflow(
`const sleep = globalThis[Symbol.for("WORKFLOW_SLEEP")];
async function workflow() {
const results = await Promise.all([sleep('2s'), sleep('5s')]);
return 'all sleeps completed';
}${getWorkflowTransformCode('workflow')}`,
workflowRun,
events
);
} catch (err) {
error = err as Error;
}
assert(error);
expect(error.name).toEqual('WorkflowSuspension');
expect((error as WorkflowSuspension).steps).toHaveLength(1);
expect((error as WorkflowSuspension).steps[0].type).toEqual('wait');
});
it('should handle sleep combined with steps', async () => {
const ops: Promise<any>[] = [];
const workflowRunId = 'test-run-123';
const workflowRun: WorkflowRun = {
runId: workflowRunId,
workflowName: 'workflow',
status: 'running',
input: dehydrateWorkflowArguments([], 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 events: Event[] = [
{
eventId: 'event-0',
runId: workflowRunId,
eventType: 'step_started',
correlationId: 'step_01HK153X008RT6YEW43G8QX6JX',
createdAt: new Date('2024-01-01T00:00:00.000Z'),
},
{
eventId: 'event-1',
runId: workflowRunId,
eventType: 'step_completed',
correlationId: 'step_01HK153X008RT6YEW43G8QX6JX',
eventData: {
result: dehydrateStepReturnValue(42, ops),
},
createdAt: new Date('2024-01-01T00:00:01.000Z'),
},
{
eventId: 'event-2',
runId: workflowRunId,
eventType: 'wait_created',
correlationId: 'wait_01HK153X008RT6YEW43G8QX6JY',
eventData: {
resumeAt: new Date('2024-01-01T00:00:03.000Z'),
},
createdAt: new Date('2024-01-01T00:00:01.000Z'),
},
{
eventId: 'event-3',
runId: workflowRunId,
eventType: 'wait_completed',
correlationId: 'wait_01HK153X008RT6YEW43G8QX6JY',
createdAt: new Date('2024-01-01T00:00:03.000Z'),
},
];
const result = await runWorkflow(
`const add = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("add");
const sleep = globalThis[Symbol.for("WORKFLOW_SLEEP")];
async function workflow() {
const stepResult = await add(1, 2);
await sleep('2s');
return { step: stepResult, slept: true };
}${getWorkflowTransformCode('workflow')}`,
workflowRun,
events
);
expect(hydrateWorkflowReturnValue(result as any, ops)).toEqual({
step: 42,
slept: true,
});
});
it('should handle sleep with Date parameter', async () => {
const ops: Promise<any>[] = [];
const workflowRunId = 'test-run-123';
const resumeAt = new Date('2024-01-01T00:00:05.000Z');
const workflowRun: WorkflowRun = {
runId: workflowRunId,
workflowName: 'workflow',
status: 'running',
input: dehydrateWorkflowArguments([], 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 events: Event[] = [
{
eventId: 'event-0',
runId: workflowRunId,
eventType: 'wait_created',
correlationId: 'wait_01HK153X008RT6YEW43G8QX6JX',
eventData: {
resumeAt,
},
createdAt: new Date('2024-01-01T00:00:00.000Z'),
},
{
eventId: 'event-1',
runId: workflowRunId,
eventType: 'wait_completed',
correlationId: 'wait_01HK153X008RT6YEW43G8QX6JX',
createdAt: resumeAt,
},
];
const result = await runWorkflow(
`const sleep = globalThis[Symbol.for("WORKFLOW_SLEEP")];
async function workflow() {
const resumeDate = new Date('2024-01-01T00:00:05.000Z');
await sleep(resumeDate);
return 'sleep with date completed';
}${getWorkflowTransformCode('workflow')}`,
workflowRun,
events
);
expect(hydrateWorkflowReturnValue(result as any, ops)).toEqual(
'sleep with date completed'
);
});
});
});
packages/core/src/workflow.ts
+5
View File
@@ -15,6 +15,7 @@ import {
BODY_INIT_SYMBOL,
WORKFLOW_CREATE_HOOK,
WORKFLOW_GET_STREAM_ID,
WORKFLOW_SLEEP,
WORKFLOW_USE_STEP,
} from './symbols.js';
import * as Attribute from './telemetry/semantic-conventions.js';
@@ -24,6 +25,7 @@ import { createContext } from './vm/index.js';
import type { WorkflowMetadata } from './workflow/get-workflow-metadata.js';
import { WORKFLOW_CONTEXT_SYMBOL } from './workflow/get-workflow-metadata.js';
import { createCreateHook } from './workflow/hook.js';
import { createSleep } from './workflow/sleep.js';
export async function runWorkflow(
workflowCode: string,
@@ -82,12 +84,15 @@ export async function runWorkflow(
const useStep = createUseStep(workflowContext);
const createHook = createCreateHook(workflowContext);
const sleep = createSleep(workflowContext);
// @ts-expect-error - `@types/node` says symbol is not valid, but it does work
vmGlobalThis[WORKFLOW_USE_STEP] = useStep;
// @ts-expect-error - `@types/node` says symbol is not valid, but it does work
vmGlobalThis[WORKFLOW_CREATE_HOOK] = createHook;
// @ts-expect-error - `@types/node` says symbol is not valid, but it does work
vmGlobalThis[WORKFLOW_SLEEP] = sleep;
// @ts-expect-error - `@types/node` says symbol is not valid, but it does work
vmGlobalThis[WORKFLOW_GET_STREAM_ID] = (namespace?: string) =>
getWorkflowRunStreamId(workflowRun.runId, namespace);
packages/core/src/workflow/index.ts
+1
View File
@@ -9,6 +9,7 @@ export type { Hook, HookOptions } from '../create-hook.js';
export { createHook, createWebhook } from './create-hook.js';
export { defineHook } from './define-hook.js';
export { getWorkflowMetadata } from './get-workflow-metadata.js';
export { sleep } from '../sleep.js';
export { getWritable } from './writable-stream.js';
// workflows can't use these functions, but we still need to provide
packages/core/src/workflow/sleep.ts
+99
View File
@@ -0,0 +1,99 @@
import type { StringValue } from 'ms';
import ms from 'ms';
import { EventConsumerResult } from '../events-consumer.js';
import { type WaitInvocationQueueItem, WorkflowSuspension } from '../global.js';
import type { WorkflowOrchestratorContext } from '../private.js';
import { withResolvers } from '../util.js';
export function createSleep(ctx: WorkflowOrchestratorContext) {
return async function sleepImpl(param: StringValue | Date): Promise<void> {
const { promise, resolve } = withResolvers<void>();
const correlationId = `wait_${ctx.generateUlid()}`;
// Calculate the resume time
let resumeAt: Date;
if (typeof param === 'string') {
const durationMs = ms(param);
if (typeof durationMs !== 'number' || durationMs < 0) {
throw new Error(
`Invalid sleep duration: "${param}". Expected a valid duration string like "1s", "1m", "1h", etc.`
);
}
resumeAt = new Date(Date.now() + durationMs);
} else if (
param instanceof Date ||
(param &&
typeof param === 'object' &&
typeof (param as any).getTime === 'function')
) {
// Handle both Date instances and date-like objects (from deserialization)
const dateParam =
param instanceof Date ? param : new Date((param as any).getTime());
resumeAt = dateParam;
} else {
throw new Error(
`Invalid sleep parameter. Expected a duration string or Date object.`
);
}
// Add wait to invocations queue
ctx.invocationsQueue.push({
type: 'wait',
correlationId,
resumeAt,
});
ctx.eventsConsumer.subscribe((event) => {
// If there are no events and we're waiting for wait_completed,
// suspend the workflow until the wait fires
if (!event) {
setTimeout(() => {
ctx.onWorkflowError(
new WorkflowSuspension(ctx.invocationsQueue, ctx.globalThis)
);
}, 0);
return EventConsumerResult.NotConsumed;
}
// Check for wait_created event to mark this wait as having the event created
if (
event?.eventType === 'wait_created' &&
event.correlationId === correlationId
) {
// Mark this wait as having the created event, but keep it in the queue
const waitItem = ctx.invocationsQueue.find(
(item) => item.type === 'wait' && item.correlationId === correlationId
) as WaitInvocationQueueItem | undefined;
if (waitItem) {
waitItem.hasCreatedEvent = true;
waitItem.resumeAt = event.eventData.resumeAt;
}
return EventConsumerResult.Consumed;
}
// Check for wait_completed event
if (
event?.eventType === 'wait_completed' &&
event.correlationId === correlationId
) {
// Remove this wait from the invocations queue
const index = ctx.invocationsQueue.findIndex(
(item) => item.type === 'wait' && item.correlationId === correlationId
);
if (index !== -1) {
ctx.invocationsQueue.splice(index, 1);
}
// Wait has elapsed, resolve the sleep
setTimeout(() => {
resolve();
}, 0);
return EventConsumerResult.Finished;
}
return EventConsumerResult.NotConsumed;
});
return promise;
};
}
packages/workflow/src/stdlib.ts
+1 -60
View File
@@ -1,69 +1,10 @@
/**
* This is the "standard library" of steps that we make available to all workflow users.
* The can be imported like so: `import { sleep, fetch } from 'workflow'`. and used in workflow.
* The can be imported like so: `import { fetch } from 'workflow'`. and used in workflow.
* The need to be exported directly in this package and cannot live in `core` to prevent
* circular dependencies post-compilation.
*/
import { RetryableError } from '@workflow/errors';
import ms, { type StringValue } from 'ms';
import { getStepMetadata } from './index.js';
// vqs has a max message visibility lifespan, the workflow sleep function
// will retry repeatedly until the user requested duration is reached.
// (Eventually make this configurable based on the queue backend adapter)
const MAX_SLEEP_DURATION_SECONDS = ms('23h') / 1000;
/**
* Sleep within a workflow for a given duration.
*
* @param duration - The duration to sleep for, this is a string in the format
* of `"1000ms"`, `"1s"`, `"1m"`, `"1h"`, or `"1d"`.
* @overload
* @returns A promise that resolves when the sleep is complete.
*/
export async function sleep(duration: StringValue): Promise<void>;
/**
* Sleep within a workflow until a specific date.
*
* @param date - The date to sleep until, this must be a future date.
* @overload
* @returns A promise that resolves when the sleep is complete.
*/
export async function sleep(date: Date): Promise<void>;
export async function sleep(param: StringValue | Date): Promise<void> {
'use step';
const { stepStartedAt } = getStepMetadata();
const durationMs =
typeof param === 'string'
? ms(param)
: param.getTime() - Number(stepStartedAt);
if (typeof durationMs !== 'number' || durationMs < 0) {
const message =
param instanceof Date
? `Invalid sleep date: "${param}". Expected a future date.`
: `Invalid sleep duration: "${param}". Expected a valid duration string like "1s", "1m", "1h", etc.`;
throw new Error(message);
}
const endAt = +stepStartedAt + durationMs;
const now = Date.now();
if (now < endAt) {
const remainingSeconds = (endAt - now) / 1000;
const retryAfter = Math.min(remainingSeconds, MAX_SLEEP_DURATION_SECONDS);
throw new RetryableError(
`Sleeping for ${ms(retryAfter * 1000, { long: true })}`,
{
retryAfter,
}
);
}
}
sleep.maxRetries = Infinity;
/**
* A hoisted `fetch()` function that is executed as a "step" function,
* for use within workflow functions.
packages/world/src/events.ts
+17
View File
@@ -10,6 +10,8 @@ export const EventTypeSchema = z.enum([
'hook_created',
'hook_received',
'hook_disposed',
'wait_created',
'wait_completed',
'workflow_completed',
'workflow_failed',
'workflow_started',
@@ -73,6 +75,19 @@ const HookDisposedEventSchema = BaseEventSchema.extend({
correlationId: z.string(),
});
const WaitCreatedEventSchema = BaseEventSchema.extend({
eventType: z.literal('wait_created'),
correlationId: z.string(),
eventData: z.object({
resumeAt: z.coerce.date(),
}),
});
const WaitCompletedEventSchema = BaseEventSchema.extend({
eventType: z.literal('wait_completed'),
correlationId: z.string(),
});
// TODO: not used yet
const WorkflowCompletedEventSchema = BaseEventSchema.extend({
eventType: z.literal('workflow_completed'),
@@ -100,6 +115,8 @@ export const CreateEventSchema = z.discriminatedUnion('eventType', [
HookCreatedEventSchema,
HookReceivedEventSchema,
HookDisposedEventSchema,
WaitCreatedEventSchema,
WaitCompletedEventSchema,
WorkflowCompletedEventSchema,
WorkflowFailedEventSchema,
WorkflowStartedEventSchema,