diff --git a/.changeset/quiet-signals-wait.md b/.changeset/quiet-signals-wait.md new file mode 100644 index 0000000..e69c4a2 --- /dev/null +++ b/.changeset/quiet-signals-wait.md @@ -0,0 +1,7 @@ +--- +'@tanstack/workflow-runtime': patch +--- + +Reject signal, approval, and timer deliveries for a run whose workflow version +is not registered, returning `version-mismatch` and leaving the run paused +instead of consuming the delivery and stranding the run. diff --git a/docs/concepts/replay-and-resume.md b/docs/concepts/replay-and-resume.md index 1dac4ef..bcb74cf 100644 --- a/docs/concepts/replay-and-resume.md +++ b/docs/concepts/replay-and-resume.md @@ -106,6 +106,10 @@ On resume the engine reads `RunState.workflowVersion` and routes to the matching Mismatched version with nothing in `previousVersions` → `RUN_ERRORED { code: 'workflow_version_mismatch' }`. +When the run is driven through `@tanstack/workflow-runtime`, signal, approval, and timer deliveries to such a run are rejected before they are consumed: the result kind is `version-mismatch` and the run stays `paused`, so restoring the missing version lets the same delivery resume it. + +A rejected signal or approval payload is not stored by the runtime. Keep it in a durable inbox or return a retryable HTTP response, such as 503, to a sender that retries. Retry with the same signal ID or approval ID after restoring the missing workflow version. The Cloudflare D1 payment example returns 503 for this case. + ## Attach (read-only subscribe) A second subscriber (browser refresh, mobile reconnect) reads current state without driving the run forward: diff --git a/examples/deployment-pocs/cloudflare-d1/src/worker.ts b/examples/deployment-pocs/cloudflare-d1/src/worker.ts index f4a07d8..e66e18f 100644 --- a/examples/deployment-pocs/cloudflare-d1/src/worker.ts +++ b/examples/deployment-pocs/cloudflare-d1/src/worker.ts @@ -73,12 +73,11 @@ export default { } if (url.pathname === '/signals/payment' && request.method === 'POST') { - return json( - await deliverPayment( - runtime, - await readJson(request), - ), + const result = await deliverPayment( + runtime, + await readJson(request), ) + return json(result, result.kind === 'version-mismatch' ? 503 : 200) } if (url.pathname === '/sweep' && request.method === 'POST') { diff --git a/packages/workflow-runtime/src/runtime-driver.ts b/packages/workflow-runtime/src/runtime-driver.ts index 904895c..b0445c2 100644 --- a/packages/workflow-runtime/src/runtime-driver.ts +++ b/packages/workflow-runtime/src/runtime-driver.ts @@ -212,6 +212,14 @@ async function deliverSignal< args.minYieldRemainingMs, ) const leaseMs = resolveLeaseMs(config, args.leaseMs) + const prepared = await prepareRunDelivery(config, telemetry, args.runId) + if (prepared.rejected) { + span.setAttribute( + 'tanstack.workflow.result_kind', + prepared.rejected.kind, + ) + return prepared.rejected + } const delivery = { signalId: args.signalId, stepId: args.stepId, @@ -236,7 +244,9 @@ async function deliverSignal< return result } - const workflow = await loadWorkflow(config, delivered.run.workflowId) + const workflow = + prepared.workflow ?? + (await loadWorkflow(config, delivered.run.workflowId)) const result = await driveClaimedRun(config, telemetry, { workflow, workflowId: delivered.run.workflowId, @@ -279,6 +289,8 @@ async function deliverSignalFromRuntime< args.minYieldRemainingMs, ) const leaseMs = resolveLeaseMs(config, args.leaseMs) + const prepared = await prepareRunDelivery(config, telemetry, args.runId) + if (prepared.rejected) return prepared.rejected const delivery = { signalId: args.signalId, stepId: args.stepId, @@ -301,7 +313,8 @@ async function deliverSignalFromRuntime< return resultFromSignalDelivery(args.runId, delivered) } - const workflow = await loadWorkflow(config, delivered.run.workflowId) + const workflow = + prepared.workflow ?? (await loadWorkflow(config, delivered.run.workflowId)) return driveClaimedRun(config, telemetry, { workflow, workflowId: delivered.run.workflowId, @@ -338,6 +351,14 @@ async function deliverApproval< args.minYieldRemainingMs, ) const leaseMs = resolveLeaseMs(config, args.leaseMs) + const prepared = await prepareRunDelivery(config, telemetry, args.runId) + if (prepared.rejected) { + span.setAttribute( + 'tanstack.workflow.result_kind', + prepared.rejected.kind, + ) + return prepared.rejected + } const delivered = await traceStoreOperation( telemetry, 'store.deliver_approval', @@ -355,7 +376,9 @@ async function deliverApproval< return result } - const workflow = await loadWorkflow(config, delivered.run.workflowId) + const workflow = + prepared.workflow ?? + (await loadWorkflow(config, delivered.run.workflowId)) const result = await driveClaimedRun(config, telemetry, { workflow, workflowId: delivered.run.workflowId, @@ -419,6 +442,8 @@ async function sweep>( const recovered: Array = [] const scheduled: Array = [] const timers: Array = [] + const seenTimers = new Set() + let rejectedTimer = false let deadlineReached = false while (recovered.length < maxRecoveredRuns) { @@ -554,7 +579,6 @@ async function sweep>( deadlineReached = true break } - const dueTimers = await traceStoreOperation( telemetry, 'store.claim_due_timers', @@ -562,16 +586,23 @@ async function sweep>( () => config.store.claimDueTimers({ now, - limit: 1, + limit: maxTimers, leaseOwner, leaseMs, }), ) - const timer = dueTimers[0] - if (!timer) break - - timers.push( - await deliverTimer(config, telemetry, { + let advanced = false + for (const timer of dueTimers) { + const timerKey = JSON.stringify([timer.runId, timer.signalId]) + if (seenTimers.has(timerKey)) continue + if (timers.length >= maxTimers) break + if (shouldStopForDeadline(deadline, minYieldRemainingMs)) { + deadlineReached = true + break + } + seenTimers.add(timerKey) + advanced = true + const result = await deliverTimer(config, telemetry, { timer, now, leaseOwner, @@ -582,8 +613,11 @@ async function sweep>( minYieldRemainingMs, yieldResumeAt: now + 1, publish: args.publish, - }), - ) + }) + if (result.kind === 'version-mismatch') rejectedTimer = true + timers.push(result) + } + if (!advanced || deadlineReached) break } const result = { @@ -593,6 +627,7 @@ async function sweep>( summary: summarizeSweep(recovered, scheduled, timers), deadlineReached, remainingMayExist: + rejectedTimer || deadlineReached || recovered.length >= maxRecoveredRuns || scheduled.length >= maxScheduledRuns || @@ -853,6 +888,66 @@ async function syncTimerFromRunState< ) } +// Deliveries are checked before the store consumes them: once consumed, +// a run whose version this runtime can't replay would be left `running` +// with no lease and nothing awaited, and redelivery would be refused. +async function prepareRunDelivery< + TWorkflows extends Record, +>( + config: WorkflowRuntimeConfig, + telemetry: WorkflowTelemetry, + runId: string, +): Promise<{ + workflow?: AnyWorkflowDefinition + rejected?: WorkflowRuntimeRunResult +}> { + const run = await traceStoreOperation( + telemetry, + 'store.load_run', + { runId }, + () => config.store.loadRun(runId), + ) + if (!run) { + return { + rejected: { kind: 'not-found', runId, events: [], eventCount: 0 }, + } + } + if ( + run.status === 'finished' || + run.status === 'errored' || + run.status === 'aborted' + ) { + return {} + } + + const state = await traceStoreOperation( + telemetry, + 'store.load_run_state', + { runId }, + () => config.store.loadRunState(runId), + ) + const version = state?.workflowVersion ?? run.workflowVersion + const workflow = await loadWorkflow(config, run.workflowId) + if ( + !version || + workflow.version === version || + workflow.previousVersions?.some((previous) => previous.version === version) + ) { + return { workflow } + } + + return { + rejected: { + kind: 'version-mismatch', + runId, + workflowId: run.workflowId, + run, + events: [], + eventCount: 0, + }, + } +} + async function loadWorkflow< TWorkflows extends Record, >( diff --git a/packages/workflow-runtime/src/types.ts b/packages/workflow-runtime/src/types.ts index d8c3d08..7867c9c 100644 --- a/packages/workflow-runtime/src/types.ts +++ b/packages/workflow-runtime/src/types.ts @@ -445,6 +445,7 @@ export type WorkflowRuntimeRunResultKind = | 'not-claimable' | 'not-waiting' | 'duplicate' + | 'version-mismatch' export interface WorkflowRuntimeRunResult { kind: WorkflowRuntimeRunResultKind diff --git a/packages/workflow-runtime/tests/runtime-driver.test.ts b/packages/workflow-runtime/tests/runtime-driver.test.ts index a618d05..3538252 100644 --- a/packages/workflow-runtime/tests/runtime-driver.test.ts +++ b/packages/workflow-runtime/tests/runtime-driver.test.ts @@ -731,6 +731,301 @@ describe('workflow runtime driver', () => { expect(due.timers[0]?.kind).toBe('completed') }) + it('rejects a signal for an unregistered version without consuming it', async () => { + const store = inMemoryWorkflowExecutionStore() + const define = (version: string) => + createWorkflow({ id: 'versioned', version }).handler(async (ctx) => { + return ctx.waitForEvent<{ ok: boolean }>('go', { id: 'go' }) + }) + const v1 = defineWorkflowRuntime({ + store, + workflows: { versioned: { load: async () => define('v1') } }, + }) + const v2 = defineWorkflowRuntime({ + store, + workflows: { versioned: { load: async () => define('v2') } }, + }) + + const start = await v1.startRun({ + workflowId: 'versioned', + runId: 'versioned:1', + input: {}, + now: 0, + }) + const rejected = await v2.deliverSignal({ + runId: 'versioned:1', + stepId: 'go', + signalId: 'signal-1', + name: 'go', + payload: { ok: true }, + now: 10, + }) + const afterReject = await store.loadRun('versioned:1') + const resumed = await v1.deliverSignal({ + runId: 'versioned:1', + stepId: 'go', + signalId: 'signal-1', + name: 'go', + payload: { ok: true }, + now: 20, + }) + + expect(start.kind).toBe('paused') + expect(rejected.kind).toBe('version-mismatch') + expect(rejected.run?.status).toBe('paused') + expect(afterReject?.status).toBe('paused') + expect(afterReject?.lease).toBeUndefined() + expect(resumed.kind).toBe('completed') + }) + + it('checks replay state when the execution summary has no version', async () => { + const store = inMemoryWorkflowExecutionStore() + const define = (version: string) => + createWorkflow({ id: 'legacy-summary', version }).handler(async (ctx) => + ctx.waitForEvent('go', { id: 'go' }), + ) + const old = defineWorkflowRuntime({ + store, + workflows: { 'legacy-summary': { load: async () => define('v1') } }, + }) + await old.startRun({ + workflowId: 'legacy-summary', + runId: 'legacy', + input: {}, + }) + const loadRun = store.loadRun.bind(store) + vi.spyOn(store, 'loadRun').mockImplementation(async (runId) => { + const run = await loadRun(runId) + return run ? { ...run, workflowVersion: undefined } : undefined + }) + const current = defineWorkflowRuntime({ + store, + workflows: { 'legacy-summary': { load: async () => define('v2') } }, + }) + const delivery = { + runId: 'legacy', + signalId: 'go-1', + name: 'go', + stepId: 'go', + payload: true, + } + expect((await current.deliverSignal(delivery)).kind).toBe( + 'version-mismatch', + ) + expect((await store.loadRun('legacy'))?.status).toBe('paused') + expect((await old.deliverSignal(delivery)).kind).toBe('completed') + }) + + it('preserves duplicate delivery results after a workflow is unregistered', async () => { + const store = inMemoryWorkflowExecutionStore() + const workflow = createWorkflow({ id: 'retired', version: 'v1' }).handler( + async (ctx) => ctx.waitForEvent('go', { id: 'go' }), + ) + const old = defineWorkflowRuntime({ + store, + workflows: { retired: { load: async () => workflow } }, + }) + await old.startRun({ workflowId: 'retired', runId: 'retired', input: {} }) + const delivery = { + runId: 'retired', + signalId: 'go-1', + name: 'go', + stepId: 'go', + payload: true, + } + expect((await old.deliverSignal(delivery)).kind).toBe('completed') + const current = defineWorkflowRuntime({ store, workflows: {} }) + expect((await current.deliverSignal(delivery)).kind).toBe('duplicate') + }) + + it('rejects an approval for an unregistered version without consuming it', async () => { + const store = inMemoryWorkflowExecutionStore() + const define = (version: string) => + createWorkflow({ id: 'approval', version }).handler(async (ctx) => { + return ctx.approve({ title: 'Ship it?' }) + }) + const v1 = defineWorkflowRuntime({ + store, + workflows: { approval: { load: async () => define('v1') } }, + }) + const v2 = defineWorkflowRuntime({ + store, + workflows: { approval: { load: async () => define('v2') } }, + }) + + await v1.startRun({ + workflowId: 'approval', + runId: 'approval:1', + input: {}, + now: 0, + }) + const approvalId = (await store.loadRun('approval:1'))?.pendingApproval + ?.approvalId + const approval = { approvalId: approvalId!, approved: true } + const rejected = await v2.deliverApproval({ + runId: 'approval:1', + approval, + now: 10, + }) + const afterReject = await store.loadRun('approval:1') + const resumed = await v1.deliverApproval({ + runId: 'approval:1', + approval, + now: 20, + }) + + expect(approvalId).toBeDefined() + expect(rejected.kind).toBe('version-mismatch') + expect(afterReject?.status).toBe('paused') + expect(resumed.kind).toBe('completed') + }) + + it('leaves a timer pending when its run version is unregistered', async () => { + const store = inMemoryWorkflowExecutionStore() + const define = (version: string) => + createWorkflow({ id: 'sleeper', version }).handler(async (ctx) => { + await ctx.sleepUntil(100) + return { awoke: true } + }) + const v1 = defineWorkflowRuntime({ + store, + workflows: { sleeper: { load: async () => define('v1') } }, + }) + const v2 = defineWorkflowRuntime({ + store, + workflows: { sleeper: { load: async () => define('v2') } }, + }) + + await v1.startRun({ + workflowId: 'sleeper', + runId: 'sleeper:1', + input: {}, + now: 0, + }) + const rejected = await v2.sweep({ now: 100, leaseMs: 50 }) + const afterReject = await store.loadRun('sleeper:1') + const resumed = await v1.sweep({ now: 200 }) + + expect(rejected.timers.map((timer) => timer.kind)).toEqual([ + 'version-mismatch', + ]) + expect(afterReject?.status).toBe('paused') + expect(resumed.timers.map((timer) => timer.kind)).toEqual(['completed']) + }) + + it('accepts a signal for a version registered via previousVersions', async () => { + const store = inMemoryWorkflowExecutionStore() + const define = (version: string) => + createWorkflow({ id: 'migrated', version }).handler(async (ctx) => { + await ctx.waitForEvent<{ ok: boolean }>('go', { id: 'go' }) + return { version } + }) + const v1 = defineWorkflowRuntime({ + store, + workflows: { migrated: { load: async () => define('v1') } }, + }) + const v2 = defineWorkflowRuntime({ + store, + workflows: { + migrated: { + load: async () => define('v2'), + previousVersions: { v1: async () => define('v1') }, + }, + }, + }) + + await v1.startRun({ + workflowId: 'migrated', + runId: 'migrated:1', + input: {}, + now: 0, + }) + const resumed = await v2.deliverSignal({ + runId: 'migrated:1', + stepId: 'go', + signalId: 'signal-1', + name: 'go', + payload: { ok: true }, + now: 10, + }) + + expect(resumed.kind).toBe('completed') + expect(resumed.run?.output).toEqual({ version: 'v1' }) + }) + + it('executes the same workflow definition that passed the delivery check', async () => { + const store = inMemoryWorkflowExecutionStore() + const define = (version: string) => + createWorkflow({ id: 'loaded-once', version }).handler(async (ctx) => + ctx.waitForEvent('go', { id: 'go' }), + ) + const first = define('v1') + const load = vi + .fn() + .mockResolvedValueOnce(first) + .mockResolvedValueOnce(first) + .mockResolvedValue(define('v2')) + const runtime = defineWorkflowRuntime({ + store, + workflows: { 'loaded-once': { load } }, + }) + await runtime.startRun({ + workflowId: 'loaded-once', + runId: 'once', + input: {}, + }) + const result = await runtime.deliverSignal({ + runId: 'once', + name: 'go', + stepId: 'go', + signalId: 'go-1', + payload: true, + }) + expect(result.kind).toBe('completed') + expect(load).toHaveBeenCalledTimes(2) + }) + + it('continues to other due timers when one workflow version is missing', async () => { + const store = inMemoryWorkflowExecutionStore() + const define = (id: string, version: string) => + createWorkflow({ id, version }).handler(async (ctx) => { + await ctx.sleepUntil(100) + return true + }) + const old = defineWorkflowRuntime({ + store, + workflows: { + missing: { load: async () => define('missing', 'v1') }, + available: { load: async () => define('available', 'v1') }, + }, + }) + await old.startRun({ + workflowId: 'missing', + runId: 'missing', + input: {}, + now: 0, + }) + await old.startRun({ + workflowId: 'available', + runId: 'available', + input: {}, + now: 0, + }) + const current = defineWorkflowRuntime({ + store, + workflows: { + missing: { load: async () => define('missing', 'v2') }, + available: { load: async () => define('available', 'v1') }, + }, + }) + const result = await current.sweep({ now: 100, maxTimers: 10 }) + expect(result.timers.map((timer) => timer.kind)).toEqual([ + 'version-mismatch', + 'completed', + ]) + expect(result.remainingMayExist).toBe(true) + }) + it('sweeps due schedule buckets into deterministic runs', async () => { const store = inMemoryWorkflowExecutionStore() const workflow = createWorkflow({ id: 'scheduled' }).handler(