Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions .changeset/world-runtime-deadline.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,7 @@
---
'@workflow/world': patch
'@workflow/world-vercel': patch
'@workflow/core': patch
---

Derive inline-execution duration limits from `World.getRuntimeDeadline()` (implemented on Vercel World via `@vercel/functions`).
48 changes: 44 additions & 4 deletions packages/core/src/runtime.ts
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ import {
type WorkflowRun,
type World,
} from '@workflow/world';
import ms from 'ms';
import { decodeTime } from 'ulid';
import {
classifyRunError,
Expand Down Expand Up @@ -525,6 +526,44 @@ function canRetainWorkflowSession(
return !openHook && !openWait;
}

/**
* Maximum inline-execution duration for a single handler invocation.
*
* Order of precedence:
* 1. `WORKFLOW_V2_TIMEOUT_MS` env var
* 2. Tiered budget derived from `world.getRuntimeDeadline()`
* 3. Default of 2 minutes
*/
async function getMaxInlineDurationMs(
world: World,
invocationStartTime: number
): Promise<number> {
const rawEnvOverride = process.env.WORKFLOW_V2_TIMEOUT_MS;
if (rawEnvOverride !== undefined && rawEnvOverride !== '') {
const parsed = Number(rawEnvOverride);
if (Number.isFinite(parsed) && parsed > 0) {
return parsed;
}
}

const runtimeDeadline = await world.getRuntimeDeadline?.();
if (runtimeDeadline !== undefined) {
const deadlineMs = runtimeDeadline.getTime();
if (!Number.isNaN(deadlineMs)) {
const maxDurationMs = Math.floor(deadlineMs - invocationStartTime);
if (maxDurationMs >= ms('25m')) {
return ms('10m');
}

if (maxDurationMs >= ms('10m')) {
return ms('5m');
}
}
}

return ms('2m');
}

/**
* Creates a single route which handles workflow execution requests,
* executing steps inline when possible to reduce function invocations
Expand All @@ -546,9 +585,6 @@ export function workflowEntrypoint(
): (req: Request) => Promise<Response> {
setWorkflowBasePath(options?.basePath);

const NO_INLINE_REPLAY_AFTER_MS =
Number(process.env.WORKFLOW_V2_TIMEOUT_MS) || 120_000;

const namespace = resolveQueueNamespace(options?.namespace);
const workflowPrefix = getQueueTopicPrefix('workflow', namespace);

Expand Down Expand Up @@ -748,6 +784,10 @@ export function workflowEntrypoint(
});

const invocationStartTime = Date.now();
const noInlineReplayAfterMs = await getMaxInlineDurationMs(
world,
invocationStartTime
);
let loopIteration = 0;
const replayRecoveryReporter = replayDivergence
? new ReplayRecoveryReporter(replayDivergence.count)
Expand Down Expand Up @@ -1864,7 +1904,7 @@ export function workflowEntrypoint(
// Check timeout before replay
if (
Date.now() - invocationStartTime >=
NO_INLINE_REPLAY_AFTER_MS
noInlineReplayAfterMs
) {
runtimeLogger.info(
'V2 timeout reached, re-scheduling workflow',
Expand Down
1 change: 1 addition & 0 deletions packages/world-vercel/package.json
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
"typecheck": "genversion --es6 src/version.ts && tsc --noEmit"
},
"dependencies": {
"@vercel/functions": "catalog:",
"@vercel/oidc": "catalog:",
"@vercel/queue": "catalog:",
"@workflow/errors": "workspace:*",
Expand Down
9 changes: 9 additions & 0 deletions packages/world-vercel/src/get-deadline.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,9 @@
/**
* Lazily loads `getDeadline` from `@vercel/functions` so importing
* `@workflow/world-vercel` does not evaluate that package at module load.
*/

export async function getDeadline() {
const { getDeadline } = await import('@vercel/functions');
return getDeadline();
}
2 changes: 2 additions & 0 deletions packages/world-vercel/src/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ import { SPEC_VERSION_SUPPORTS_COMPRESSION } from '@workflow/world';
import { createAnalytics } from './analytics.js';
import { createRunId, describeRun } from './create-run-id.js';
import { createGetEncryptionKeyForRun } from './encryption.js';
import { getDeadline } from './get-deadline.js';
import { instrumentObject } from './instrumentObject.js';
import { createQueue } from './queue.js';
import { createResolveLatestDeploymentId } from './resolve-latest-deployment.js';
Expand Down Expand Up @@ -56,6 +57,7 @@ export function createWorld(config?: APIConfig): World {
// `process.exit(1)` is an acceptable response to an exhausted replay
// budget.
processExitTriggersQueueRedelivery: true,
getRuntimeDeadline: getDeadline,
...createQueue(config),
...createStorage(config),
// Analytics list reads are served from an eventually-ingested store.
Expand Down
6 changes: 6 additions & 0 deletions packages/world/src/interfaces.ts
Original file line number Diff line number Diff line change
Expand Up @@ -440,6 +440,12 @@ export interface World extends Queue, Streamer, Storage {
*/
processExitTriggersQueueRedelivery?: boolean;

/**
* Absolute wall-clock time when the current function invocation will be
* terminated by the hosting platform, if known. Used to optimize runtime behavior.
*/
getRuntimeDeadline?(): Promise<Date | undefined>;

/**
* A function that will be called to start any background tasks needed by the World implementation.
* For example, in the case of a queue backed World, this would start the queue processing.
Expand Down
194 changes: 121 additions & 73 deletions pnpm-lock.yaml

Large diffs are not rendered by default.

2 changes: 1 addition & 1 deletion pnpm-workspace.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ catalog:
"@swc/core": 1.15.3
"@types/json-schema": ^7.0.15
"@types/node": 22.19.0
"@vercel/functions": ^3.4.3
"@vercel/functions": ^3.8.0
"@vercel/oidc": 3.2.0
"@vercel/queue": 0.4.0
"@vitest/coverage-v8": ^4.1.10
Expand Down
Loading