diff --git a/.changeset/calm-clocks-yield.md b/.changeset/calm-clocks-yield.md new file mode 100644 index 0000000..6d4466b --- /dev/null +++ b/.changeset/calm-clocks-yield.md @@ -0,0 +1,8 @@ +--- +'@tanstack/workflow-core': patch +'@tanstack/workflow-runtime': patch +--- + +Add runtime deadlines, automatic cooperative yielding at durable boundaries, +and deadline helpers under `ctx.runtime`. Timer wake identities now include the +durable operation ID so sequential waits at the same timestamp resume safely. diff --git a/.changeset/metal-shoes-observe.md b/.changeset/metal-shoes-observe.md new file mode 100644 index 0000000..8de6e66 --- /dev/null +++ b/.changeset/metal-shoes-observe.md @@ -0,0 +1,6 @@ +--- +'@tanstack/workflow-core': patch +'@tanstack/workflow-runtime': patch +--- + +Add first-class OpenTelemetry tracing for workflow runtime operations, durable store calls, and fresh step execution. diff --git a/docs/api/runtime.md b/docs/api/runtime.md index 65c90ec..dc5726b 100644 --- a/docs/api/runtime.md +++ b/docs/api/runtime.md @@ -18,6 +18,9 @@ import { defineWorkflowRuntime } from '@tanstack/workflow-runtime' const runtime = defineWorkflowRuntime({ store, defaultLeaseMs: 30_000, + telemetry: { + spanNamePrefix: 'tanstack.workflow', + }, workflows: { checkout: { load: async () => checkoutWorkflow, @@ -31,6 +34,43 @@ const runtime = defineWorkflowRuntime({ }) ``` +### Telemetry + +Workflow emits OpenTelemetry traces through `@opentelemetry/api`. Applications +own SDK/exporter setup. If no SDK is configured, spans are no-ops. + +```ts +const runtime = defineWorkflowRuntime({ + store, + workflows, + telemetry: { + attributes: ({ workflowId, runId }) => ({ + 'app.workflow': workflowId ?? 'unknown', + 'app.run': runId ?? 'unknown', + }), + mapStepMeta: (meta) => ({ + 'app.step_name': String(meta.name), + }), + }, +}) +``` + +Set `telemetry: false` or `telemetry: { enabled: false }` to disable tracing. + +`WorkflowTelemetryOptions`: + +| Option | Purpose | +| --- | --- | +| `enabled` | Set false to disable tracing. | +| `tracer` | Custom OpenTelemetry `Tracer`. Defaults to the global tracer provider. | +| `spanNamePrefix` | Prefix for Workflow span names. Defaults to `tanstack.workflow`. | +| `attributes` | Safe shared attributes or a function that returns them per span. | +| `recordExceptions` | Whether to call `span.recordException`. Defaults to true. | +| `mapStepMeta` | Optional mapper for safe `ctx.step(..., { meta })` attributes. | + +Workflow never records workflow input, output, signal payloads, step results, or +raw step metadata as attributes by default. + ### Workflow registration | Field | Purpose | @@ -40,6 +80,24 @@ const runtime = defineWorkflowRuntime({ | `previousVersions` | Loaders for old versions that still have paused runs. | | `schedules` | Registered recurring schedules for this workflow. | +## Runtime deadlines + +Every runtime method that can drive workflow code accepts the same budget +options: + +| Option | Purpose | +| --- | --- | +| `deadline` | Absolute wall-clock deadline in UTC milliseconds. | +| `maxDurationMs` | Maximum wall-clock duration for this runtime call. | +| `minYieldRemainingMs` | Budget reserved before fresh durable work. Defaults to 1000ms. | + +When both deadline forms are present, the earlier deadline wins. The runtime +checks the budget before fresh `step`, `now`, and `uuid` work. A low budget +pauses the run on a durable timer, releases its lease, and lets a later sweep +resume it. Existing `sleep`, `waitForEvent`, and `approve` calls already pause. + +Step timeouts remain per-attempt limits and do not use the runtime deadline. + ## `runtime.startRun` Starts or attaches to a run ID through the runtime store. @@ -63,6 +121,9 @@ Options: | `runId` | Stable run identifier. | | `input` | Workflow input for new runs. | | `now` | Override current time for tests or host events. | +| `deadline` | Absolute wall-clock deadline in UTC milliseconds. | +| `maxDurationMs` | Maximum wall-clock duration for this call. | +| `minYieldRemainingMs` | Budget reserved before fresh durable work. | | `leaseOwner` | Worker identity used while executing. | | `leaseMs` | Lease duration. | | `threadId` | Optional core engine thread ID. | @@ -124,7 +185,9 @@ Options: | `limit` | Backward-compatible shared limit for schedules and timers. | | `maxScheduledRuns` | Maximum due schedule buckets to start. | | `maxTimers` | Maximum due timers to resume. | +| `deadline` | Absolute wall-clock deadline in UTC milliseconds. | | `maxDurationMs` | Wall-clock budget for this sweep. | +| `minYieldRemainingMs` | Stops claims and fresh durable work inside this margin. | | `leaseOwner` | Worker identity for claimed work. | | `leaseMs` | Lease duration. | | `includeEvents` | Whether to retain emitted events. Sweeps usually set false. | diff --git a/docs/concepts/primitives.md b/docs/concepts/primitives.md index b24e0c4..9330e2e 100644 --- a/docs/concepts/primitives.md +++ b/docs/concepts/primitives.md @@ -124,6 +124,32 @@ ctx.emit('progress', { step: 3, of: 10 }) Run-level `AbortSignal`. Already-aborted state propagates to `step` fns via `stepCtx.signal`. +## `ctx.runtime` + +Runtime deadline metadata and cooperative yielding: + +```ts +ctx.runtime.deadline +ctx.runtime.timeRemaining() +ctx.runtime.shouldYield({ minRemainingMs: 2_000 }) +await ctx.runtime.yield({ reason: 'host deadline' }) +``` + +The runtime automatically yields before fresh durable work when its budget is +low, so queue loops do not need arbitrary batch sizes: + +```ts +for (let index = 0; ; index++) { + const item = await ctx.step(`take-${index}`, takeNextItem) + if (!item) break + await ctx.step(`process-${item.id}`, () => processItem(item)) +} +``` + +`stepCtx.runtime` exposes the deadline, remaining time, and `shouldYield` for +long-running step logic. A step cannot yield midway because partial step work is +not a durable checkpoint. Step timeouts remain independent. + ```ts await ctx.step('long-fetch', (stepCtx) => fetch(url, { signal: stepCtx.signal }), diff --git a/docs/config.json b/docs/config.json index bd03d51..598fb00 100644 --- a/docs/config.json +++ b/docs/config.json @@ -42,6 +42,10 @@ "label": "Persistence", "to": "guide/persistence" }, + { + "label": "Observability", + "to": "guide/observability" + }, { "label": "Deployment", "to": "guide/deployment" diff --git a/docs/cookbook/index.md b/docs/cookbook/index.md index 8800bf1..00e6bb5 100644 --- a/docs/cookbook/index.md +++ b/docs/cookbook/index.md @@ -95,6 +95,39 @@ export const workflowRuntime = defineWorkflowRuntime({ }) ``` +## Add OpenTelemetry tracing + +Configure your app's OpenTelemetry SDK in the host-specific instrumentation +entrypoint, then create the workflow runtime normally. Workflow emits spans +through `@opentelemetry/api` and does not configure exporters. + +```ts +export const workflowRuntime = defineWorkflowRuntime({ + store, + workflows, + telemetry: { + attributes: ({ workflowId }) => ({ + 'app.workflow': workflowId ?? 'unknown', + }), + }, +}) +``` + +For Railway or long-running Node services, initialize the SDK before importing +the runtime. For Vercel, Netlify, and Cloudflare, use the provider's supported +instrumentation/bootstrap entrypoint. Host adapters inherit tracing because they +call `runtime.sweep()`. + +Disable tracing for tests or sensitive environments: + +```ts +export const workflowRuntime = defineWorkflowRuntime({ + store, + workflows, + telemetry: false, +}) +``` + ## Apply workflow store migrations Workflow owns its durable store schema. Apply the package-owned SQL migration diff --git a/docs/guide/index.md b/docs/guide/index.md index 9ecdb83..cddaec2 100644 --- a/docs/guide/index.md +++ b/docs/guide/index.md @@ -20,8 +20,9 @@ This guide walks through the production shape: 1. Define a workflow with `@tanstack/workflow-core`. 2. Put it behind a runtime with `@tanstack/workflow-runtime`. 3. Persist executions in a durable store. -4. Wake due timers and schedules with a host cron, scheduled function, or worker. -5. Deploy the same workflow code on Cloudflare, Railway, Netlify, Node, Vercel, +4. Observe runtime, store, and step execution with OpenTelemetry. +5. Wake due timers and schedules with a host cron, scheduled function, or worker. +6. Deploy the same workflow code on Cloudflare, Railway, Netlify, Node, Vercel, or your own infrastructure. ## Package map @@ -162,6 +163,28 @@ export const workflowRuntime = defineWorkflowRuntime({ Use the in-memory store for tests and demos only. Production deployments need a store that can persist executions across invocations. +## Add OpenTelemetry + +TanStack Workflow emits OpenTelemetry traces through `@opentelemetry/api`. +Applications own SDK and exporter setup; Workflow emits no-op spans until an SDK +is configured. + +```ts +export const workflowRuntime = defineWorkflowRuntime({ + store, + workflows, + telemetry: { + attributes: ({ workflowId }) => ({ + 'app.workflow': workflowId ?? 'unknown', + }), + }, +}) +``` + +Workflow traces runtime operations, store calls, sweeps, and fresh `ctx.step` +executions. It does not record workflow input, output, signal payloads, step +results, or raw metadata as attributes by default. + ## Use Postgres for durability The first production-style store is Drizzle/Postgres: diff --git a/docs/guide/observability.md b/docs/guide/observability.md new file mode 100644 index 0000000..ec17142 --- /dev/null +++ b/docs/guide/observability.md @@ -0,0 +1,110 @@ +--- +id: observability +title: Observability +--- + +# Observability + +TanStack Workflow emits OpenTelemetry traces from the runtime and core engine. +The packages depend only on `@opentelemetry/api`; your application owns the OTel +SDK, exporter, collector, and provider setup. + +If no OpenTelemetry SDK is configured, tracing is a no-op. + +## What is traced + +The runtime creates spans for: + +- `tanstack.workflow.start_run` +- `tanstack.workflow.deliver_signal` +- `tanstack.workflow.deliver_approval` +- `tanstack.workflow.sweep` +- `tanstack.workflow.drive_run` + +Store calls made by the runtime are traced as child spans, such as +`tanstack.workflow.store.claim_run`, +`tanstack.workflow.store.append_events`, and +`tanstack.workflow.store.claim_due_timers`. + +Fresh `ctx.step` executions create `tanstack.workflow.step` spans. Replayed +steps do not create fresh step execution spans because no user step code ran. + +## Privacy defaults + +Workflow tracing intentionally does not record workflow input, workflow output, +signal payloads, step results, or arbitrary step metadata as attributes. + +Built-in attributes are stable identifiers and counts: + +- `tanstack.workflow.workflow_id` +- `tanstack.workflow.workflow_version` +- `tanstack.workflow.run_id` +- `tanstack.workflow.operation` +- `tanstack.workflow.result_kind` +- `tanstack.workflow.step_id` +- `tanstack.workflow.signal_name` +- `tanstack.workflow.schedule_id` +- `tanstack.workflow.bucket_id` +- `tanstack.workflow.lease_owner` +- `tanstack.workflow.event_count` +- `tanstack.workflow.events_truncated` + +If you want to add safe application attributes, pass `telemetry.attributes` or +`telemetry.mapStepMeta`. + +## Configure tracing + +Initialize OpenTelemetry in your app before Workflow code runs. Then create the +runtime normally: + +```ts +import { defineWorkflowRuntime } from '@tanstack/workflow-runtime' + +export const workflowRuntime = defineWorkflowRuntime({ + store, + workflows, +}) +``` + +Workflow uses the global OpenTelemetry tracer provider by default. + +To customize Workflow tracing: + +```ts +export const workflowRuntime = defineWorkflowRuntime({ + store, + workflows, + telemetry: { + spanNamePrefix: 'tanstack.workflow', + attributes: ({ workflowId }) => ({ + 'app.workflow_id': workflowId ?? 'unknown', + }), + mapStepMeta: (meta) => ({ + 'app.step_name': String(meta.name), + }), + }, +}) +``` + +Only use `mapStepMeta` for metadata you know is safe to export. Workflow does +not export step metadata by default. + +Disable tracing for a runtime: + +```ts +export const workflowRuntime = defineWorkflowRuntime({ + store, + workflows, + telemetry: false, +}) +``` + +## Deployment notes + +Host adapters do not need separate tracing configuration. Netlify, Vercel, +Cloudflare, and Railway helpers call `runtime.sweep()`, so they inherit the +runtime spans automatically. + +For serverless platforms, initialize the OpenTelemetry SDK in the platform's +recommended instrumentation entrypoint. Workflow only emits spans through the +OpenTelemetry API; exporter flushing and process lifecycle are app concerns. diff --git a/packages/workflow-cloudflare/package.json b/packages/workflow-cloudflare/package.json index 8fd0568..d74b1d3 100644 --- a/packages/workflow-cloudflare/package.json +++ b/packages/workflow-cloudflare/package.json @@ -1,6 +1,6 @@ { "name": "@tanstack/workflow-cloudflare", - "version": "0.0.1", + "version": "0.0.2", "description": "Cloudflare host adapter for TanStack Workflow runtimes.", "author": "Tanner Linsley", "license": "MIT", diff --git a/packages/workflow-core/package.json b/packages/workflow-core/package.json index ae1929f..3017f0f 100644 --- a/packages/workflow-core/package.json +++ b/packages/workflow-core/package.json @@ -58,6 +58,10 @@ "@standard-schema/spec": "^1.1.0" }, "devDependencies": { + "@opentelemetry/api": "^1.9.0", "zod": "^4.2.0" + }, + "peerDependencies": { + "@opentelemetry/api": "^1.9.0" } } diff --git a/packages/workflow-core/src/engine/run-workflow.ts b/packages/workflow-core/src/engine/run-workflow.ts index a2ae65e..0020a91 100644 --- a/packages/workflow-core/src/engine/run-workflow.ts +++ b/packages/workflow-core/src/engine/run-workflow.ts @@ -1,4 +1,5 @@ import { LogConflictError, StepTimeoutError, WorkflowPaused } from '../types' +import { createWorkflowTelemetry } from '../telemetry' import { diffState, snapshotState } from './state-diff' import type { AnyMiddleware, @@ -11,6 +12,7 @@ import type { RunState, RunStore, SerializedError, + ShouldYieldOptions, SignalDelivery, SleepOptions, StepContext, @@ -18,12 +20,17 @@ import type { StepRetryOptions, WaitForEventOptions, WorkflowEvent, + YieldOptions, } from '../types' +import type { WorkflowTelemetry, WorkflowTelemetryOptions } from '../telemetry' // ============================================================ // Public API // ============================================================ +const DEFAULT_MIN_YIELD_REMAINING_MS = 1_000 +const AUTOMATIC_YIELD_STEP_ID_PREFIX = '__runtime-yield-' + export interface RunWorkflowOptions { workflow: AnyWorkflowDefinition runStore: RunStore @@ -46,6 +53,18 @@ export interface RunWorkflowOptions { /** Called with the workflow's final output before the run record is * cleaned up. */ outputSink?: (output: unknown) => void + /** OpenTelemetry tracing configuration. Uses the global tracer provider + * by default and records no payloads/results as attributes. */ + telemetry?: false | WorkflowTelemetryOptions + /** Absolute UTC ms deadline for this runtime drive. Workflow automatically + * yields before fresh durable work when the remaining budget is too low. */ + deadline?: number + /** Minimum budget that must remain before starting fresh durable work. + * Defaults to 1000ms when `deadline` is set. */ + minYieldRemainingMs?: number + /** UTC ms wake time used for cooperative yields. Runtime sets this past the + * current sweep timestamp so yielded runs continue on a later sweep. */ + yieldResumeAt?: number } /** @@ -128,11 +147,17 @@ export async function* runWorkflow( interface DriveOptions extends RunWorkflowOptions { emit: (event: WorkflowEvent) => void + telemetryImpl?: WorkflowTelemetry } type DeliveryLostCode = 'signal_lost' | 'approval_lost' async function drive(options: DriveOptions): Promise { + options.telemetryImpl ??= createWorkflowTelemetry( + options.telemetry, + '@tanstack/workflow-core', + ) + if (options.runId && options.attach) { await attachRun(options) return @@ -420,6 +445,7 @@ async function driveHandler(args: DriveHandlerArgs): Promise { runStore, emit, abortController, + telemetry: options.telemetryImpl!, history: [...history], nextLogIndex: history.length, consumed: new Set(), @@ -429,7 +455,12 @@ async function driveHandler(args: DriveHandlerArgs): Promise { approve: 0, now: 0, uuid: 0, + yield: 0, }, + deadline: options.deadline, + minYieldRemainingMs: + options.minYieldRemainingMs ?? DEFAULT_MIN_YIELD_REMAINING_MS, + yieldResumeAt: options.yieldResumeAt, prevStateSnapshot: snapshotState(state), state, paused: false, @@ -440,6 +471,12 @@ async function driveHandler(args: DriveHandlerArgs): Promise { input: args.input, state, signal: abortController.signal, + runtime: { + deadline: options.deadline, + timeRemaining: () => timeRemaining(engine), + shouldYield: (opts) => shouldYield(engine, opts), + yield: (opts) => engineYield(engine, opts), + }, step: (id, fn, opts) => engineStep(engine, id, fn, opts), sleep: (ms, opts) => engineSleep(engine, ms, opts), @@ -472,6 +509,9 @@ async function driveHandler(args: DriveHandlerArgs): Promise { ctx, workflow.handler, ) + // A handler can accidentally catch the internal pause sentinel. The pause + // was already persisted, so it must still win over a normal handler return. + if (engine.paused) return output = validateWorkflowOutput(workflow, output) // Flush any final state delta. flushStateDelta(engine) @@ -553,6 +593,7 @@ interface EngineRuntime { runStore: RunStore emit: (event: WorkflowEvent) => void abortController: AbortController + telemetry: WorkflowTelemetry /** Pre-loaded log from prior invocations, used for replay short- * circuit. */ history: ReadonlyArray @@ -572,7 +613,11 @@ interface EngineRuntime { approve: number now: number uuid: number + yield: number } + deadline?: number + minYieldRemainingMs: number + yieldResumeAt?: number prevStateSnapshot: Record state: Record /** Set to `true` by the primitive that paused the run, so the @@ -580,6 +625,40 @@ interface EngineRuntime { paused: boolean } +function timeRemaining(engine: EngineRuntime): number { + if (engine.deadline === undefined) return Number.POSITIVE_INFINITY + return Math.max(0, engine.deadline - Date.now()) +} + +function shouldYield( + engine: EngineRuntime, + options?: ShouldYieldOptions, +): boolean { + if (engine.deadline === undefined) return false + const minRemainingMs = Math.max( + 0, + options?.minRemainingMs ?? engine.minYieldRemainingMs, + ) + return timeRemaining(engine) <= minRemainingMs +} + +async function yieldForDeadlineIfNeeded(engine: EngineRuntime): Promise { + if (!shouldYield(engine)) return + + const previousYields = engine.history.filter( + (event) => + event.type === 'SIGNAL_AWAITED' && + event.name === '__timer' && + event.stepId.startsWith(AUTOMATIC_YIELD_STEP_ID_PREFIX), + ).length + + await engineWaitForEvent(engine, '__timer', { + id: `${AUTOMATIC_YIELD_STEP_ID_PREFIX}${previousYields}`, + deadline: engine.yieldResumeAt ?? Date.now() + 1, + meta: { reason: 'deadline' }, + }) +} + // ============================================================ // Primitives — replay-aware durable steps // ============================================================ @@ -602,6 +681,12 @@ async function engineStep( e.stepId === stepId, ) if (cached) { + engine.telemetry.addActiveSpanEvent('workflow.step.replayed', { + 'tanstack.workflow.workflow_id': engine.workflow.id, + 'tanstack.workflow.workflow_version': engine.workflow.version, + 'tanstack.workflow.run_id': engine.runId, + 'tanstack.workflow.step_id': stepId, + }) if (cached.event.type === 'STEP_FAILED') { throw rehydrateError(cached.event.error) } @@ -614,6 +699,8 @@ async function engineStep( return event.result as T } + await yieldForDeadlineIfNeeded(engine) + // Fresh execution. engine.emit({ type: 'STEP_STARTED', @@ -622,142 +709,172 @@ async function engineStep( meta: options?.meta, }) - const startedAt = Date.now() - const retryPolicy = options?.retry ?? engine.workflow.defaultStepRetry - const maxAttempts = Math.max(1, retryPolicy?.maxAttempts ?? 1) - const attempts: Array<{ - startedAt: number - finishedAt: number - result?: unknown - error?: SerializedError - }> = [] - let lastError: unknown - let result: unknown - let succeeded = false - - for (let attempt = 1; attempt <= maxAttempts; attempt++) { - const attemptStart = Date.now() - const attemptController = new AbortController() - // Eager propagation: addEventListener doesn't fire for already- - // aborted signals, so check + abort upfront. - if (engine.abortController.signal.aborted) attemptController.abort() - const onParentAbort = () => attemptController.abort() - engine.abortController.signal.addEventListener('abort', onParentAbort, { - once: true, - }) - let timeoutHandle: ReturnType | null = null - let timedOut = false - if (options?.timeout && options.timeout > 0) { - timeoutHandle = setTimeout(() => { - timedOut = true - attemptController.abort() - }, options.timeout) - } + return await engine.telemetry.startActiveSpan( + 'step', + { + workflowId: engine.workflow.id, + workflowVersion: engine.workflow.version, + runId: engine.runId, + stepId, + }, + async (span) => { + const retryPolicy = options?.retry ?? engine.workflow.defaultStepRetry + const maxAttempts = Math.max(1, retryPolicy?.maxAttempts ?? 1) + const attempts: Array<{ + startedAt: number + finishedAt: number + result?: unknown + error?: SerializedError + }> = [] + let lastError: unknown + let result: unknown + let succeeded = false + + for (let attempt = 1; attempt <= maxAttempts; attempt++) { + const attemptStart = Date.now() + span.addEvent('workflow.step.attempt.started', { attempt }) + const attemptController = new AbortController() + // Eager propagation: addEventListener doesn't fire for already- + // aborted signals, so check + abort upfront. + if (engine.abortController.signal.aborted) attemptController.abort() + const onParentAbort = () => attemptController.abort() + engine.abortController.signal.addEventListener('abort', onParentAbort, { + once: true, + }) + let timeoutHandle: ReturnType | null = null + let timedOut = false + if (options?.timeout && options.timeout > 0) { + timeoutHandle = setTimeout(() => { + timedOut = true + span.addEvent('workflow.step.attempt.timeout', { attempt }) + attemptController.abort() + }, options.timeout) + } - try { - const fnPromise = Promise.resolve( - fn({ - id: `${engine.runId}:${stepId}`, - attempt, - signal: attemptController.signal, - }), - ) - result = options?.timeout - ? await Promise.race([ - fnPromise, - new Promise((_, reject) => { - attemptController.signal.addEventListener( + try { + const fnPromise = Promise.resolve( + fn({ + id: `${engine.runId}:${stepId}`, + attempt, + runtime: { + deadline: engine.deadline, + timeRemaining: () => timeRemaining(engine), + shouldYield: (opts) => shouldYield(engine, opts), + }, + signal: attemptController.signal, + }), + ) + result = options?.timeout + ? await Promise.race([ + fnPromise, + new Promise((_, reject) => { + attemptController.signal.addEventListener( + 'abort', + () => { + if (timedOut) { + reject(new StepTimeoutError(stepId, options.timeout!)) + } else if (engine.abortController.signal.aborted) { + reject(new Error('Workflow aborted')) + } else { + reject(new StepTimeoutError(stepId, options.timeout!)) + } + }, + { once: true }, + ) + }), + ]) + : await fnPromise + attempts.push({ + startedAt: attemptStart, + finishedAt: Date.now(), + result, + }) + span.addEvent('workflow.step.attempt.finished', { attempt }) + succeeded = true + if (timeoutHandle) clearTimeout(timeoutHandle) + engine.abortController.signal.removeEventListener( + 'abort', + onParentAbort, + ) + break + } catch (err) { + if (timeoutHandle) clearTimeout(timeoutHandle) + engine.abortController.signal.removeEventListener( + 'abort', + onParentAbort, + ) + lastError = err + attempts.push({ + startedAt: attemptStart, + finishedAt: Date.now(), + error: serializeError(err), + }) + span.addEvent('workflow.step.attempt.failed', { attempt }) + const shouldRetry = + attempt < maxAttempts && + (retryPolicy?.shouldRetry?.(err, attempt) ?? true) + if (!shouldRetry) break + const delayMs = computeBackoffMs(retryPolicy, attempt) + if (delayMs > 0) { + await new Promise((resolve) => { + const t = setTimeout(resolve, delayMs) + engine.abortController.signal.addEventListener( 'abort', () => { - if (timedOut) { - reject(new StepTimeoutError(stepId, options.timeout!)) - } else if (engine.abortController.signal.aborted) { - reject(new Error('Workflow aborted')) - } else { - reject(new StepTimeoutError(stepId, options.timeout!)) - } + clearTimeout(t) + resolve() }, { once: true }, ) - }), - ]) - : await fnPromise - attempts.push({ - startedAt: attemptStart, - finishedAt: Date.now(), - result, - }) - succeeded = true - if (timeoutHandle) clearTimeout(timeoutHandle) - engine.abortController.signal.removeEventListener('abort', onParentAbort) - break - } catch (err) { - if (timeoutHandle) clearTimeout(timeoutHandle) - engine.abortController.signal.removeEventListener('abort', onParentAbort) - lastError = err - attempts.push({ - startedAt: attemptStart, - finishedAt: Date.now(), - error: serializeError(err), - }) - const shouldRetry = - attempt < maxAttempts && - (retryPolicy?.shouldRetry?.(err, attempt) ?? true) - if (!shouldRetry) break - const delayMs = computeBackoffMs(retryPolicy, attempt) - if (delayMs > 0) { - await new Promise((resolve) => { - const t = setTimeout(resolve, delayMs) - engine.abortController.signal.addEventListener( - 'abort', - () => { - clearTimeout(t) - resolve() - }, - { once: true }, - ) - }) - if (engine.abortController.signal.aborted) break + }) + if (engine.abortController.signal.aborted) break + } + } } - } - } - if (!succeeded) { - const failedEvent: WorkflowEvent = { - type: 'STEP_FAILED', - ts: Date.now(), - stepId, - error: serializeError(lastError), - attempts: attempts.length > 1 ? attempts : undefined, - meta: options?.meta, - } - await emitAndAppend( - engine.runStore, - engine.runId, - engine.nextLogIndex++, - engine.emit, - failedEvent, - ) - throw rehydrateError(serializeError(lastError)) - } + if (!succeeded) { + const failedEvent: WorkflowEvent = { + type: 'STEP_FAILED', + ts: Date.now(), + stepId, + error: serializeError(lastError), + attempts: attempts.length > 1 ? attempts : undefined, + meta: options?.meta, + } + await emitAndAppend( + engine.runStore, + engine.runId, + engine.nextLogIndex++, + engine.emit, + failedEvent, + ) + throw rehydrateError(serializeError(lastError)) + } - void startedAt - const finishedEvent: WorkflowEvent = { - type: 'STEP_FINISHED', - ts: Date.now(), - stepId, - result, - attempts: attempts.length > 1 ? attempts : undefined, - meta: options?.meta, - } - await emitAndAppend( - engine.runStore, - engine.runId, - engine.nextLogIndex++, - engine.emit, - finishedEvent, + const finishedEvent: WorkflowEvent = { + type: 'STEP_FINISHED', + ts: Date.now(), + stepId, + result, + attempts: attempts.length > 1 ? attempts : undefined, + meta: options?.meta, + } + await emitAndAppend( + engine.runStore, + engine.runId, + engine.nextLogIndex++, + engine.emit, + finishedEvent, + ) + return result as T + }, + engine.telemetry.mapStepMeta(options?.meta, { + workflowId: engine.workflow.id, + workflowVersion: engine.workflow.version, + runId: engine.runId, + stepId, + }), ) - return result as T } async function engineWaitForEvent( @@ -866,6 +983,22 @@ function engineSleep( return engineSleepUntil(engine, Date.now() + ms, options) } +function engineYield( + engine: EngineRuntime, + options: YieldOptions = {}, +): Promise { + const meta = + options.reason === undefined + ? options.meta + : { ...options.meta, reason: options.reason } + + return engineWaitForEvent(engine, '__timer', { + id: options.id ?? `__yield-${engine.counters.yield++}`, + deadline: engine.yieldResumeAt ?? Date.now() + 1, + meta, + }) +} + async function engineApprove( engine: EngineRuntime, approveOptions: ApproveOptions, @@ -958,6 +1091,7 @@ async function engineNow( return (cached.event as Extract) .value } + await yieldForDeadlineIfNeeded(engine) const value = Date.now() await emitAndAppend( engine.runStore, @@ -986,6 +1120,7 @@ async function engineUuid( return (cached.event as Extract) .value } + await yieldForDeadlineIfNeeded(engine) const value = globalThis.crypto.randomUUID() await emitAndAppend( engine.runStore, @@ -1012,6 +1147,7 @@ const reservedCtxFields = new Set([ 'input', 'state', 'signal', + 'runtime', 'step', 'sleep', 'sleepUntil', diff --git a/packages/workflow-core/src/index.ts b/packages/workflow-core/src/index.ts index 1d4eaeb..1936e0c 100644 --- a/packages/workflow-core/src/index.ts +++ b/packages/workflow-core/src/index.ts @@ -16,6 +16,13 @@ export { fail, succeed } from './result' // ===== Engine ===== export { runWorkflow } from './engine/run-workflow' export type { RunWorkflowOptions } from './engine/run-workflow' +export { createWorkflowTelemetry } from './telemetry' +export type { + WorkflowTelemetry, + WorkflowTelemetryOptions, + WorkflowTelemetrySpanContext, + WorkflowTelemetryStepMetaContext, +} from './telemetry' export { handleWorkflowWebhook } from './engine/handle-webhook' export type { HandleWebhookOptions, @@ -67,11 +74,13 @@ export type { RunStore, SchemaInput, SerializedError, + ShouldYieldOptions, SignalDelivery, SleepOptions, StepAttempt, StepContext, StepOptions, + StepRuntimeContext, StepRetryOptions, WaitForEventOptions, WorkflowCtx, @@ -80,5 +89,7 @@ export type { WorkflowInput, WorkflowMetadata, WorkflowOutput, + WorkflowRuntimeContext, WorkflowState, + YieldOptions, } from './types' diff --git a/packages/workflow-core/src/telemetry.ts b/packages/workflow-core/src/telemetry.ts new file mode 100644 index 0000000..b52f811 --- /dev/null +++ b/packages/workflow-core/src/telemetry.ts @@ -0,0 +1,182 @@ +import { INVALID_SPAN_CONTEXT, SpanStatusCode, trace } from '@opentelemetry/api' +import type { Attributes, Span, Tracer } from '@opentelemetry/api' +import type { WorkflowMetadata } from './types' + +const DEFAULT_SPAN_NAME_PREFIX = 'tanstack.workflow' + +export interface WorkflowTelemetrySpanContext { + workflowId?: string + workflowVersion?: string + runId?: string + operation?: string + resultKind?: string + stepId?: string + signalName?: string + scheduleId?: string + bucketId?: string + leaseOwner?: string + eventCount?: number + eventsTruncated?: boolean +} + +export interface WorkflowTelemetryStepMetaContext { + workflowId?: string + workflowVersion?: string + runId: string + stepId: string +} + +export interface WorkflowTelemetryOptions { + enabled?: boolean + tracer?: Tracer + spanNamePrefix?: string + attributes?: + | Attributes + | ((ctx: WorkflowTelemetrySpanContext) => Attributes | undefined) + recordExceptions?: boolean + mapStepMeta?: ( + meta: WorkflowMetadata, + ctx: WorkflowTelemetryStepMetaContext, + ) => Attributes | undefined +} + +export interface WorkflowTelemetry { + enabled: boolean + spanNamePrefix: string + startActiveSpan: ( + operation: string, + spanContext: WorkflowTelemetrySpanContext, + fn: (span: Span) => T | Promise, + attributes?: Attributes, + ) => Promise + addActiveSpanEvent: (name: string, attributes?: Attributes) => void + mapStepMeta: ( + meta: WorkflowMetadata | undefined, + ctx: WorkflowTelemetryStepMetaContext, + ) => Attributes | undefined +} + +export function createWorkflowTelemetry( + options: false | WorkflowTelemetryOptions | undefined, + tracerName: string, +): WorkflowTelemetry { + if (options === false || options?.enabled === false) { + return disabledWorkflowTelemetry + } + + const tracer = options?.tracer ?? trace.getTracer(tracerName) + const spanNamePrefix = options?.spanNamePrefix ?? DEFAULT_SPAN_NAME_PREFIX + const recordExceptions = options?.recordExceptions ?? true + + return { + enabled: true, + spanNamePrefix, + async startActiveSpan(operation, spanContext, fn, attributes) { + return await tracer.startActiveSpan( + `${spanNamePrefix}.${operation}`, + { + attributes: { + ...resolveSharedAttributes(options?.attributes, spanContext), + ...safeSpanAttributes(spanContext), + ...attributes, + 'tanstack.workflow.operation': operation, + }, + }, + async (span) => { + try { + const result = await fn(span) + span.setStatus({ code: SpanStatusCode.OK }) + return result + } catch (error) { + markSpanError(span, error, recordExceptions) + throw error + } finally { + span.end() + } + }, + ) + }, + addActiveSpanEvent(name, attributes) { + trace.getActiveSpan()?.addEvent(name, attributes) + }, + mapStepMeta(meta, ctx) { + if (!meta || !options?.mapStepMeta) return undefined + return options.mapStepMeta(meta, ctx) + }, + } +} + +function markSpanError(span: Span, error: unknown, recordException: boolean) { + const serialized = serializeTelemetryError(error) + span.setStatus({ + code: SpanStatusCode.ERROR, + message: serialized.message, + }) + span.setAttribute('error.type', serialized.name) + span.setAttribute('exception.type', serialized.name) + span.setAttribute('exception.message', serialized.message) + if (recordException) { + if (error instanceof Error) { + span.recordException(error) + } else { + span.recordException(serialized) + } + } +} + +function resolveSharedAttributes( + attributes: WorkflowTelemetryOptions['attributes'], + spanContext: WorkflowTelemetrySpanContext, +) { + if (!attributes) return undefined + if (typeof attributes === 'function') return attributes(spanContext) + return attributes +} + +function safeSpanAttributes(ctx: WorkflowTelemetrySpanContext): Attributes { + return compactAttributes({ + 'tanstack.workflow.workflow_id': ctx.workflowId, + 'tanstack.workflow.workflow_version': ctx.workflowVersion, + 'tanstack.workflow.run_id': ctx.runId, + 'tanstack.workflow.result_kind': ctx.resultKind, + 'tanstack.workflow.step_id': ctx.stepId, + 'tanstack.workflow.signal_name': ctx.signalName, + 'tanstack.workflow.schedule_id': ctx.scheduleId, + 'tanstack.workflow.bucket_id': ctx.bucketId, + 'tanstack.workflow.lease_owner': ctx.leaseOwner, + 'tanstack.workflow.event_count': ctx.eventCount, + 'tanstack.workflow.events_truncated': ctx.eventsTruncated, + }) +} + +function compactAttributes(attributes: Attributes): Attributes { + return Object.fromEntries( + Object.entries(attributes).filter(([, value]) => value !== undefined), + ) +} + +function serializeTelemetryError(error: unknown) { + if (error instanceof Error) { + return { + name: error.name, + message: error.message, + } + } + + return { + name: typeof error, + message: String(error), + } +} + +const disabledWorkflowTelemetry: WorkflowTelemetry = { + enabled: false, + spanNamePrefix: DEFAULT_SPAN_NAME_PREFIX, + async startActiveSpan(_operation, _spanContext, fn) { + return await fn(trace.wrapSpanContext(INVALID_SPAN_CONTEXT)) + }, + addActiveSpanEvent() {}, + mapStepMeta() { + return undefined + }, +} diff --git a/packages/workflow-core/src/types.ts b/packages/workflow-core/src/types.ts index 4366627..9522471 100644 --- a/packages/workflow-core/src/types.ts +++ b/packages/workflow-core/src/types.ts @@ -226,6 +226,8 @@ export interface StepContext { id: string /** Current attempt number (1-indexed). */ attempt: number + /** Current runtime-drive metadata and budget helpers. */ + runtime: StepRuntimeContext /** Per-attempt AbortSignal. Fires on: * - step timeout firing * - run-level abort (Ctrl+C / external cancellation) */ @@ -280,6 +282,28 @@ export interface SleepOptions extends DurableOperationOptions {} export interface DeterministicValueOptions extends DurableOperationOptions {} +export interface ShouldYieldOptions { + /** Minimum budget that must remain before starting fresh durable work. */ + minRemainingMs?: number +} + +export interface YieldOptions extends DurableOperationOptions { + reason?: string +} + +export interface StepRuntimeContext { + /** Absolute UTC ms runtime deadline, when the runtime supplied one. */ + deadline?: number + /** Remaining runtime budget in ms. Returns Infinity when unbounded. */ + timeRemaining: () => number + /** True when the runtime is too close to its deadline for fresh work. */ + shouldYield: (options?: ShouldYieldOptions) => boolean +} + +export interface WorkflowRuntimeContext extends StepRuntimeContext { + yield: (options?: YieldOptions) => Promise +} + export interface ApproveOptions extends DurableOperationOptions { title: string description?: string @@ -304,6 +328,8 @@ export interface BaseCtx { state: TState /** AbortSignal for the run as a whole. */ signal: AbortSignal + /** Current runtime-drive metadata and budget helpers. */ + runtime: WorkflowRuntimeContext // ── Durable primitives (replay-aware) ──────────────────────── step: ( @@ -333,6 +359,7 @@ export type ReservedCtxFields = | 'input' | 'state' | 'signal' + | 'runtime' | 'step' | 'sleep' | 'sleepUntil' diff --git a/packages/workflow-core/tests/telemetry.test.ts b/packages/workflow-core/tests/telemetry.test.ts new file mode 100644 index 0000000..6f16ae2 --- /dev/null +++ b/packages/workflow-core/tests/telemetry.test.ts @@ -0,0 +1,222 @@ +import { describe, expect, it, vi } from 'vitest' +import { trace } from '@opentelemetry/api' +import { + createWorkflow, + createWorkflowTelemetry, + inMemoryRunStore, + runWorkflow, +} from '../src' +import { collect } from './test-utils' +import type { + Attributes, + Context, + Link, + Span, + SpanContext, + SpanOptions, + SpanStatus, + TimeInput, + Tracer, +} from '@opentelemetry/api' + +describe('workflow OpenTelemetry tracing', () => { + it('does not expose an ambient span when telemetry is disabled', async () => { + const ambientSpan = new TestSpan('ambient', undefined) + const activeSpanSpy = vi + .spyOn(trace, 'getActiveSpan') + .mockReturnValue(ambientSpan) + + try { + const telemetry = createWorkflowTelemetry(false, 'test') + const callbackSpan = await telemetry.startActiveSpan( + 'disabled', + {}, + async (span) => span, + ) + + expect(callbackSpan).not.toBe(ambientSpan) + expect(callbackSpan.isRecording()).toBe(false) + } finally { + activeSpanSpy.mockRestore() + } + }) + + it('records fresh step spans without payload or result attributes', async () => { + const tracer = new TestTracer() + const workflow = createWorkflow({ id: 'otel-core', version: 'v1' }).handler( + async (ctx) => { + const value = await ctx.step( + 'safe-step', + () => ({ secret: 'do-not-record' }), + { meta: { publicName: 'safe' } }, + ) + return value + }, + ) + + await collect( + runWorkflow({ + workflow, + input: { secret: 'input-secret' }, + runStore: inMemoryRunStore(), + telemetry: { + tracer, + attributes: { 'app.safe': 'yes' }, + mapStepMeta: (meta) => ({ 'app.step_name': String(meta.publicName) }), + }, + }), + ) + + const stepSpan = tracer.spans.find( + (span) => span.name === 'tanstack.workflow.step', + ) + expect(stepSpan).toBeDefined() + expect(stepSpan?.attributes).toMatchObject({ + 'app.safe': 'yes', + 'app.step_name': 'safe', + 'tanstack.workflow.workflow_id': 'otel-core', + 'tanstack.workflow.workflow_version': 'v1', + 'tanstack.workflow.step_id': 'safe-step', + }) + expect(JSON.stringify(stepSpan?.attributes)).not.toContain('secret') + expect(stepSpan?.events.map((event) => event.name)).toEqual([ + 'workflow.step.attempt.started', + 'workflow.step.attempt.finished', + ]) + }) + + it('records retry attempt events and failed step exceptions', async () => { + const tracer = new TestTracer() + const workflow = createWorkflow({ id: 'otel-retry' }).handler( + async (ctx) => { + try { + await ctx.step( + 'flaky', + () => { + throw new Error('boom') + }, + { retry: { maxAttempts: 2, backoff: 'fixed', baseMs: 1 } }, + ) + } catch { + return { recovered: true } + } + return { recovered: false } + }, + ) + + await collect( + runWorkflow({ + workflow, + input: {}, + runStore: inMemoryRunStore(), + telemetry: { tracer }, + }), + ) + + const stepSpan = tracer.spans.find( + (span) => span.name === 'tanstack.workflow.step', + ) + expect(stepSpan?.events.map((event) => event.name)).toEqual([ + 'workflow.step.attempt.started', + 'workflow.step.attempt.failed', + 'workflow.step.attempt.started', + 'workflow.step.attempt.failed', + ]) + expect(stepSpan?.status.code).toBe(2) + expect(stepSpan?.exceptions).toHaveLength(1) + }) +}) + +class TestTracer implements Tracer { + spans: Array = [] + + startSpan(name: string, options?: SpanOptions): Span { + const span = new TestSpan(name, options?.attributes) + this.spans.push(span) + return span + } + + startActiveSpan unknown>( + name: string, + optionsOrFn: SpanOptions | F, + contextOrFn?: Context | F, + maybeFn?: F, + ): ReturnType { + const options = typeof optionsOrFn === 'function' ? undefined : optionsOrFn + const fn = + typeof optionsOrFn === 'function' + ? optionsOrFn + : typeof contextOrFn === 'function' + ? contextOrFn + : maybeFn + if (!fn) throw new Error('Missing span callback') + return fn(this.startSpan(name, options)) as ReturnType + } +} + +class TestSpan implements Span { + attributes: Attributes = {} + events: Array<{ name: string; attributes?: Attributes }> = [] + exceptions: Array = [] + status: SpanStatus = { code: 0 } + ended = false + + constructor( + readonly name: string, + attributes: Attributes | undefined, + ) { + if (attributes) this.attributes = { ...attributes } + } + + spanContext(): SpanContext { + return { + traceId: '00000000000000000000000000000001', + spanId: '0000000000000001', + traceFlags: 1, + } + } + + setAttribute(key: string, value: Attributes[string]): this { + this.attributes[key] = value + return this + } + + setAttributes(attributes: Attributes): this { + Object.assign(this.attributes, attributes) + return this + } + + addEvent(name: string, attributes?: Attributes): this { + this.events.push({ name, attributes }) + return this + } + + addLink(_link: Link): this { + return this + } + + addLinks(_links: Array): this { + return this + } + + setStatus(status: SpanStatus): this { + this.status = status + return this + } + + updateName(_name: string): this { + return this + } + + end(_endTime?: TimeInput): void { + this.ended = true + } + + isRecording(): boolean { + return true + } + + recordException(exception: unknown): void { + this.exceptions.push(exception) + } +} diff --git a/packages/workflow-netlify/package.json b/packages/workflow-netlify/package.json index e133af0..7573168 100644 --- a/packages/workflow-netlify/package.json +++ b/packages/workflow-netlify/package.json @@ -1,6 +1,6 @@ { "name": "@tanstack/workflow-netlify", - "version": "0.0.2", + "version": "0.0.3", "description": "Netlify host adapter for TanStack Workflow runtimes.", "author": "Tanner Linsley", "license": "MIT", diff --git a/packages/workflow-railway/package.json b/packages/workflow-railway/package.json index 56837e1..6c4db8c 100644 --- a/packages/workflow-railway/package.json +++ b/packages/workflow-railway/package.json @@ -1,6 +1,6 @@ { "name": "@tanstack/workflow-railway", - "version": "0.0.1", + "version": "0.0.2", "description": "Railway host adapter for TanStack Workflow runtimes.", "author": "Tanner Linsley", "license": "MIT", diff --git a/packages/workflow-runtime/package.json b/packages/workflow-runtime/package.json index dd4f334..e0cc5cd 100644 --- a/packages/workflow-runtime/package.json +++ b/packages/workflow-runtime/package.json @@ -1,6 +1,6 @@ { "name": "@tanstack/workflow-runtime", - "version": "0.0.1", + "version": "0.0.2", "description": "Runtime contracts for scheduling, durable execution stores, leases, timers, signals, and host adapters for TanStack Workflow.", "author": "Tanner Linsley", "license": "MIT", @@ -56,5 +56,11 @@ ], "dependencies": { "@tanstack/workflow-core": "workspace:*" + }, + "devDependencies": { + "@opentelemetry/api": "^1.9.0" + }, + "peerDependencies": { + "@opentelemetry/api": "^1.9.0" } } diff --git a/packages/workflow-runtime/src/index.ts b/packages/workflow-runtime/src/index.ts index 0333fd8..cf19c95 100644 --- a/packages/workflow-runtime/src/index.ts +++ b/packages/workflow-runtime/src/index.ts @@ -11,6 +11,12 @@ export type { MaterializeWorkflowSchedulesOptions, } from './schedule-materializer' +export type { + WorkflowTelemetryOptions, + WorkflowTelemetrySpanContext, + WorkflowTelemetryStepMetaContext, +} from '@tanstack/workflow-core' + export type { AppendEventsArgs, AppendEventsResult, diff --git a/packages/workflow-runtime/src/run-store-adapter.ts b/packages/workflow-runtime/src/run-store-adapter.ts index 4c7e301..57183bb 100644 --- a/packages/workflow-runtime/src/run-store-adapter.ts +++ b/packages/workflow-runtime/src/run-store-adapter.ts @@ -2,6 +2,7 @@ import type { DeleteReason, RunState, WorkflowEvent, + WorkflowTelemetry, } from '@tanstack/workflow-core' import type { WorkflowRunStoreAdapter, @@ -10,18 +11,25 @@ import type { export function createRunStoreAdapter( store: WorkflowRunStoreAdapterStore, + telemetry?: WorkflowTelemetry, ): WorkflowRunStoreAdapter { return { getRunState(runId) { - return store.loadRunState(runId) + return traceStoreOperation(telemetry, 'store.load_run_state', () => + store.loadRunState(runId), + ) }, setRunState(_runId: string, state: RunState) { - return store.saveRunState({ state }) + return traceStoreOperation(telemetry, 'store.save_run_state', () => + store.saveRunState({ state }), + ) }, deleteRun(runId: string, reason: DeleteReason) { - return store.deleteRun(runId, reason) + return traceStoreOperation(telemetry, 'store.delete_run', () => + store.deleteRun(runId, reason), + ) }, async appendEvent( @@ -29,15 +37,21 @@ export function createRunStoreAdapter( expectedNextIndex: number, event: WorkflowEvent, ) { - await store.appendEvents({ - runId, - expectedNextIndex, - events: [event], - }) + await traceStoreOperation(telemetry, 'store.append_events', () => + store.appendEvents({ + runId, + expectedNextIndex, + events: [event], + }), + ) }, async getEvents(runId: string) { - const events = await store.readEvents({ runId }) + const events = await traceStoreOperation( + telemetry, + 'store.read_events', + () => store.readEvents({ runId }), + ) return events.map((event) => event.event) }, @@ -47,3 +61,12 @@ export function createRunStoreAdapter( : undefined, } } + +async function traceStoreOperation( + telemetry: WorkflowTelemetry | undefined, + operation: string, + fn: () => Promise, +) { + if (!telemetry) return await fn() + return await telemetry.startActiveSpan(operation, {}, fn) +} diff --git a/packages/workflow-runtime/src/runtime-driver.ts b/packages/workflow-runtime/src/runtime-driver.ts index b4e88d4..e1eafcb 100644 --- a/packages/workflow-runtime/src/runtime-driver.ts +++ b/packages/workflow-runtime/src/runtime-driver.ts @@ -1,8 +1,9 @@ -import { runWorkflow } from '@tanstack/workflow-core' +import { createWorkflowTelemetry, runWorkflow } from '@tanstack/workflow-core' import { createRunStoreAdapter } from './run-store-adapter' import type { AnyWorkflowDefinition, WorkflowEvent, + WorkflowTelemetry, } from '@tanstack/workflow-core' import type { DeliverApprovalResult, @@ -22,24 +23,30 @@ import type { const DEFAULT_LEASE_MS = 30_000 const DEFAULT_SWEEP_LIMIT = 25 +const DEFAULT_MIN_YIELD_REMAINING_MS = 1_000 export function createRuntimeDriver< TWorkflows extends Record, >(config: WorkflowRuntimeConfig) { + const telemetry = createWorkflowTelemetry( + config.telemetry, + '@tanstack/workflow-runtime', + ) + return { startRun(args: WorkflowRuntimeStartRunArgs) { - return startRun(config, args) + return startRun(config, telemetry, args) }, deliverSignal( args: WorkflowRuntimeDeliverSignalArgs, ) { - return deliverSignal(config, args) + return deliverSignal(config, telemetry, args) }, deliverApproval(args: WorkflowRuntimeDeliverApprovalArgs) { - return deliverApproval(config, args) + return deliverApproval(config, telemetry, args) }, sweep(args: WorkflowRuntimeSweepArgs = {}) { - return sweep(config, args) + return sweep(config, telemetry, args) }, } } @@ -48,20 +55,119 @@ async function startRun< TWorkflows extends Record, >( config: WorkflowRuntimeConfig, + telemetry: WorkflowTelemetry, args: WorkflowRuntimeStartRunArgs, ): Promise { - const now = args.now ?? Date.now() + return await telemetry.startActiveSpan( + 'start_run', + { + workflowId: args.workflowId, + runId: args.runId, + leaseOwner: args.leaseOwner, + }, + async (span) => { + const startedAt = Date.now() + const now = args.now ?? startedAt + const deadline = resolveRuntimeDeadline(args, startedAt) + const minYieldRemainingMs = normalizeMinYieldRemainingMs( + args.minYieldRemainingMs, + ) + const workflow = await loadWorkflow(config, args.workflowId) + const workflowVersion = workflow.version + span.setAttribute( + 'tanstack.workflow.workflow_version', + workflowVersion ?? '', + ) + const created = await traceStoreOperation( + telemetry, + 'store.create_run', + { + workflowId: args.workflowId, + workflowVersion, + runId: args.runId, + }, + () => + config.store.createRun({ + runId: args.runId, + workflowId: args.workflowId, + workflowVersion, + input: args.input, + now, + }), + ) + + if (created.kind === 'existing' && created.run.status !== 'queued') { + const result = resultFromExistingRun(created.run) + span.setAttribute('tanstack.workflow.result_kind', result.kind) + return result + } + + const result = await driveClaimedRun(config, telemetry, { + workflow, + workflowId: args.workflowId, + runId: args.runId, + input: args.input, + now, + leaseOwner: args.leaseOwner, + leaseMs: args.leaseMs, + threadId: args.threadId, + includeEvents: args.includeEvents, + maxEvents: args.maxEvents, + deadline, + minYieldRemainingMs, + yieldResumeAt: now + 1, + }) + span.setAttribute('tanstack.workflow.result_kind', result.kind) + span.setAttribute('tanstack.workflow.event_count', result.eventCount) + if (result.eventsTruncated !== undefined) { + span.setAttribute( + 'tanstack.workflow.events_truncated', + result.eventsTruncated, + ) + } + return result + }, + ) +} + +async function startRunFromSweep< + TWorkflows extends Record, +>( + config: WorkflowRuntimeConfig, + telemetry: WorkflowTelemetry, + args: WorkflowRuntimeStartRunArgs & { yieldResumeAt?: number }, +): Promise { + const startedAt = Date.now() + const now = args.now ?? startedAt + const deadline = resolveRuntimeDeadline(args, startedAt) + const minYieldRemainingMs = normalizeMinYieldRemainingMs( + args.minYieldRemainingMs, + ) const workflow = await loadWorkflow(config, args.workflowId) const workflowVersion = workflow.version - await config.store.createRun({ - runId: args.runId, - workflowId: args.workflowId, - workflowVersion, - input: args.input, - now, - }) + const created = await traceStoreOperation( + telemetry, + 'store.create_run', + { + workflowId: args.workflowId, + workflowVersion, + runId: args.runId, + }, + () => + config.store.createRun({ + runId: args.runId, + workflowId: args.workflowId, + workflowVersion, + input: args.input, + now, + }), + ) - return driveClaimedRun(config, { + if (created.kind === 'existing' && created.run.status !== 'queued') { + return resultFromRunSnapshot(created.run) + } + + return driveClaimedRun(config, telemetry, { workflow, workflowId: args.workflowId, runId: args.runId, @@ -72,6 +178,9 @@ async function startRun< threadId: args.threadId, includeEvents: args.includeEvents, maxEvents: args.maxEvents, + deadline, + minYieldRemainingMs, + yieldResumeAt: now + 1, }) } @@ -80,9 +189,88 @@ async function deliverSignal< TPayload, >( config: WorkflowRuntimeConfig, + telemetry: WorkflowTelemetry, args: WorkflowRuntimeDeliverSignalArgs, ): Promise { - const now = args.now ?? Date.now() + return await telemetry.startActiveSpan( + 'deliver_signal', + { + runId: args.runId, + signalName: args.name, + leaseOwner: args.leaseOwner, + }, + async (span) => { + const startedAt = Date.now() + const now = args.now ?? startedAt + const deadline = resolveRuntimeDeadline(args, startedAt) + const minYieldRemainingMs = normalizeMinYieldRemainingMs( + args.minYieldRemainingMs, + ) + const delivery = { + signalId: args.signalId, + stepId: args.stepId, + name: args.name, + payload: args.payload, + meta: args.meta, + } + const delivered = await traceStoreOperation( + telemetry, + 'store.deliver_signal', + { runId: args.runId, signalName: args.name }, + () => + config.store.deliverSignal({ + runId: args.runId, + delivery, + now, + }), + ) + if (delivered.kind !== 'delivered') { + const result = resultFromSignalDelivery(args.runId, delivered) + span.setAttribute('tanstack.workflow.result_kind', result.kind) + return result + } + + const workflow = await loadWorkflow(config, delivered.run.workflowId) + const result = await driveClaimedRun(config, telemetry, { + workflow, + workflowId: delivered.run.workflowId, + runId: args.runId, + signalDelivery: delivery, + now, + leaseOwner: args.leaseOwner, + leaseMs: args.leaseMs, + threadId: args.threadId, + includeEvents: args.includeEvents, + maxEvents: args.maxEvents, + deadline, + minYieldRemainingMs, + yieldResumeAt: now + 1, + }) + span.setAttribute( + 'tanstack.workflow.workflow_id', + result.workflowId ?? '', + ) + span.setAttribute('tanstack.workflow.result_kind', result.kind) + span.setAttribute('tanstack.workflow.event_count', result.eventCount) + return result + }, + ) +} + +async function deliverSignalFromRuntime< + TWorkflows extends Record, + TPayload, +>( + config: WorkflowRuntimeConfig, + telemetry: WorkflowTelemetry, + args: WorkflowRuntimeDeliverSignalArgs & { yieldResumeAt?: number }, +): Promise { + const startedAt = Date.now() + const now = args.now ?? startedAt + const deadline = resolveRuntimeDeadline(args, startedAt) + const minYieldRemainingMs = normalizeMinYieldRemainingMs( + args.minYieldRemainingMs, + ) const delivery = { signalId: args.signalId, stepId: args.stepId, @@ -90,17 +278,23 @@ async function deliverSignal< payload: args.payload, meta: args.meta, } - const delivered = await config.store.deliverSignal({ - runId: args.runId, - delivery, - now, - }) + const delivered = await traceStoreOperation( + telemetry, + 'store.deliver_signal', + { runId: args.runId, signalName: args.name }, + () => + config.store.deliverSignal({ + runId: args.runId, + delivery, + now, + }), + ) if (delivered.kind !== 'delivered') { return resultFromSignalDelivery(args.runId, delivered) } const workflow = await loadWorkflow(config, delivered.run.workflowId) - return driveClaimedRun(config, { + return driveClaimedRun(config, telemetry, { workflow, workflowId: delivered.run.workflowId, runId: args.runId, @@ -111,6 +305,9 @@ async function deliverSignal< threadId: args.threadId, includeEvents: args.includeEvents, maxEvents: args.maxEvents, + deadline, + minYieldRemainingMs, + yieldResumeAt: now + 1, }) } @@ -118,134 +315,210 @@ async function deliverApproval< TWorkflows extends Record, >( config: WorkflowRuntimeConfig, + telemetry: WorkflowTelemetry, args: WorkflowRuntimeDeliverApprovalArgs, ): Promise { - const now = args.now ?? Date.now() - const delivered = await config.store.deliverApproval({ - runId: args.runId, - approval: args.approval, - now, - }) - if (delivered.kind !== 'delivered') { - return resultFromApprovalDelivery(args.runId, delivered) - } - - const workflow = await loadWorkflow(config, delivered.run.workflowId) - return driveClaimedRun(config, { - workflow, - workflowId: delivered.run.workflowId, - runId: args.runId, - approval: args.approval, - now, - leaseOwner: args.leaseOwner, - leaseMs: args.leaseMs, - threadId: args.threadId, - includeEvents: args.includeEvents, - maxEvents: args.maxEvents, - }) + return await telemetry.startActiveSpan( + 'deliver_approval', + { runId: args.runId, leaseOwner: args.leaseOwner }, + async (span) => { + const startedAt = Date.now() + const now = args.now ?? startedAt + const deadline = resolveRuntimeDeadline(args, startedAt) + const minYieldRemainingMs = normalizeMinYieldRemainingMs( + args.minYieldRemainingMs, + ) + const delivered = await traceStoreOperation( + telemetry, + 'store.deliver_approval', + { runId: args.runId }, + () => + config.store.deliverApproval({ + runId: args.runId, + approval: args.approval, + now, + }), + ) + if (delivered.kind !== 'delivered') { + const result = resultFromApprovalDelivery(args.runId, delivered) + span.setAttribute('tanstack.workflow.result_kind', result.kind) + return result + } + + const workflow = await loadWorkflow(config, delivered.run.workflowId) + const result = await driveClaimedRun(config, telemetry, { + workflow, + workflowId: delivered.run.workflowId, + runId: args.runId, + approval: args.approval, + now, + leaseOwner: args.leaseOwner, + leaseMs: args.leaseMs, + threadId: args.threadId, + includeEvents: args.includeEvents, + maxEvents: args.maxEvents, + deadline, + minYieldRemainingMs, + yieldResumeAt: now + 1, + }) + span.setAttribute( + 'tanstack.workflow.workflow_id', + result.workflowId ?? '', + ) + span.setAttribute('tanstack.workflow.result_kind', result.kind) + span.setAttribute('tanstack.workflow.event_count', result.eventCount) + return result + }, + ) } async function sweep>( config: WorkflowRuntimeConfig, + telemetry: WorkflowTelemetry, args: WorkflowRuntimeSweepArgs, ): Promise { - const now = args.now ?? Date.now() - const startedAt = Date.now() - const maxScheduledRuns = normalizeSweepLimit( - args.maxScheduledRuns ?? args.limit, - DEFAULT_SWEEP_LIMIT, - 'maxScheduledRuns', - ) - const maxTimers = normalizeSweepLimit( - args.maxTimers ?? args.limit, - DEFAULT_SWEEP_LIMIT, - 'maxTimers', + return await telemetry.startActiveSpan( + 'sweep', + { leaseOwner: args.leaseOwner }, + async (span) => { + const startedAt = Date.now() + const now = args.now ?? startedAt + const deadline = resolveRuntimeDeadline(args, startedAt) + const minYieldRemainingMs = normalizeMinYieldRemainingMs( + args.minYieldRemainingMs, + ) + const maxScheduledRuns = normalizeSweepLimit( + args.maxScheduledRuns ?? args.limit, + DEFAULT_SWEEP_LIMIT, + 'maxScheduledRuns', + ) + const maxTimers = normalizeSweepLimit( + args.maxTimers ?? args.limit, + DEFAULT_SWEEP_LIMIT, + 'maxTimers', + ) + const leaseOwner = args.leaseOwner ?? `sweep:${now}` + span.setAttribute('tanstack.workflow.lease_owner', leaseOwner) + const leaseMs = args.leaseMs ?? config.defaultLeaseMs ?? DEFAULT_LEASE_MS + const scheduled: Array = [] + const timers: Array = [] + let deadlineReached = false + + while (scheduled.length < maxScheduledRuns) { + if (shouldStopForDeadline(deadline, minYieldRemainingMs)) { + deadlineReached = true + break + } + + const buckets = await traceStoreOperation( + telemetry, + 'store.claim_due_schedule_buckets', + { leaseOwner }, + () => + config.store.claimDueScheduleBuckets({ + now, + limit: 1, + leaseOwner, + leaseMs, + }), + ) + const bucket = buckets[0] + if (!bucket) break + + const result = await startRunFromSweep(config, telemetry, { + workflowId: bucket.workflowId, + runId: bucket.runId, + input: bucket.input, + now, + leaseOwner, + leaseMs, + includeEvents: args.includeEvents, + maxEvents: args.maxEvents, + deadline, + minYieldRemainingMs, + yieldResumeAt: now + 1, + }) + if (result.kind !== 'not-claimable' && result.kind !== 'not-found') { + await traceStoreOperation( + telemetry, + 'store.mark_schedule_bucket_started', + { + scheduleId: bucket.scheduleId, + bucketId: bucket.bucketId, + runId: bucket.runId, + }, + () => + config.store.markScheduleBucketStarted({ + scheduleId: bucket.scheduleId, + bucketId: bucket.bucketId, + runId: bucket.runId, + now, + }), + ) + } + scheduled.push(result) + } + + while (timers.length < maxTimers) { + if (shouldStopForDeadline(deadline, minYieldRemainingMs)) { + deadlineReached = true + break + } + + const dueTimers = await traceStoreOperation( + telemetry, + 'store.claim_due_timers', + { leaseOwner }, + () => + config.store.claimDueTimers({ + now, + limit: 1, + leaseOwner, + leaseMs, + }), + ) + const timer = dueTimers[0] + if (!timer) break + + timers.push( + await deliverTimer(config, telemetry, { + timer, + now, + leaseOwner, + leaseMs, + includeEvents: args.includeEvents, + maxEvents: args.maxEvents, + deadline, + minYieldRemainingMs, + yieldResumeAt: now + 1, + }), + ) + } + + const result = { + scheduled, + timers, + summary: summarizeSweep(scheduled, timers), + deadlineReached, + remainingMayExist: + deadlineReached || + scheduled.length >= maxScheduledRuns || + timers.length >= maxTimers, + } + span.setAttribute( + 'tanstack.workflow.event_count', + result.summary.eventCount, + ) + return result + }, ) - const leaseOwner = args.leaseOwner ?? `sweep:${now}` - const leaseMs = args.leaseMs ?? config.defaultLeaseMs ?? DEFAULT_LEASE_MS - const scheduled: Array = [] - const timers: Array = [] - let deadlineReached = false - - while (scheduled.length < maxScheduledRuns) { - if (isPastSweepDeadline(startedAt, args.maxDurationMs)) { - deadlineReached = true - break - } - - const buckets = await config.store.claimDueScheduleBuckets({ - now, - limit: 1, - leaseOwner, - leaseMs, - }) - const bucket = buckets[0] - if (!bucket) break - - const result = await startRun(config, { - workflowId: bucket.workflowId, - runId: bucket.runId, - input: bucket.input, - now, - leaseOwner, - leaseMs, - includeEvents: args.includeEvents, - maxEvents: args.maxEvents, - }) - if (result.kind !== 'not-claimable' && result.kind !== 'not-found') { - await config.store.markScheduleBucketStarted({ - scheduleId: bucket.scheduleId, - bucketId: bucket.bucketId, - runId: bucket.runId, - now, - }) - } - scheduled.push(result) - } - - while (timers.length < maxTimers) { - if (isPastSweepDeadline(startedAt, args.maxDurationMs)) { - deadlineReached = true - break - } - - const dueTimers = await config.store.claimDueTimers({ - now, - limit: 1, - leaseOwner, - leaseMs, - }) - const timer = dueTimers[0] - if (!timer) break - - timers.push( - await deliverTimer(config, { - timer, - now, - leaseOwner, - leaseMs, - includeEvents: args.includeEvents, - maxEvents: args.maxEvents, - }), - ) - } - - return { - scheduled, - timers, - summary: summarizeSweep(scheduled, timers), - deadlineReached, - remainingMayExist: - deadlineReached || - scheduled.length >= maxScheduledRuns || - timers.length >= maxTimers, - } } async function deliverTimer< TWorkflows extends Record, >( config: WorkflowRuntimeConfig, + telemetry: WorkflowTelemetry, args: { timer: TimerWakeup now: number @@ -253,9 +526,12 @@ async function deliverTimer< leaseMs: number includeEvents?: boolean maxEvents?: number + deadline?: number + minYieldRemainingMs?: number + yieldResumeAt?: number }, ) { - return deliverSignal(config, { + return deliverSignalFromRuntime(config, telemetry, { runId: args.timer.runId, signalId: args.timer.signalId, name: '__timer', @@ -265,6 +541,9 @@ async function deliverTimer< leaseMs: args.leaseMs, includeEvents: args.includeEvents, maxEvents: args.maxEvents, + deadline: args.deadline, + minYieldRemainingMs: args.minYieldRemainingMs, + yieldResumeAt: args.yieldResumeAt, }) } @@ -272,6 +551,7 @@ async function driveClaimedRun< TWorkflows extends Record, >( config: WorkflowRuntimeConfig, + telemetry: WorkflowTelemetry, args: { workflow: AnyWorkflowDefinition workflowId: string @@ -285,91 +565,181 @@ async function driveClaimedRun< threadId?: string includeEvents?: boolean maxEvents?: number + deadline?: number + minYieldRemainingMs?: number + yieldResumeAt?: number }, ): Promise { - const leaseOwner = args.leaseOwner ?? `runtime:${args.runId}` - const leaseMs = args.leaseMs ?? config.defaultLeaseMs ?? DEFAULT_LEASE_MS - const claim = await config.store.claimRun({ - runId: args.runId, - leaseOwner, - leaseMs, - now: args.now, - }) - - if (claim.kind === 'not-found') { - return { - kind: 'not-found', - runId: args.runId, + return await telemetry.startActiveSpan( + 'drive_run', + { workflowId: args.workflowId, - eventCount: 0, - events: [], - } - } - if (claim.kind === 'not-claimable') { - return { - kind: 'not-claimable', + workflowVersion: args.workflow.version, runId: args.runId, - workflowId: args.workflowId, - run: claim.run, - eventCount: 0, - events: [], - } - } - - const runStore = createRunStoreAdapter(config.store) - const collected = await collectWorkflowEvents( - runWorkflow({ - workflow: args.workflow, - runStore, - runId: args.runId, - input: args.input, - signalDelivery: args.signalDelivery, - approval: args.approval, - threadId: args.threadId, - }), - { - includeEvents: args.includeEvents ?? true, - maxEvents: args.maxEvents, + leaseOwner: args.leaseOwner, + }, + async (span) => { + const leaseOwner = args.leaseOwner ?? `runtime:${args.runId}` + const leaseMs = args.leaseMs ?? config.defaultLeaseMs ?? DEFAULT_LEASE_MS + span.setAttribute('tanstack.workflow.lease_owner', leaseOwner) + const claim = await traceStoreOperation( + telemetry, + 'store.claim_run', + { + workflowId: args.workflowId, + workflowVersion: args.workflow.version, + runId: args.runId, + leaseOwner, + }, + () => + config.store.claimRun({ + runId: args.runId, + leaseOwner, + leaseMs, + now: args.now, + }), + ) + + if (claim.kind === 'not-found') { + span.setAttribute('tanstack.workflow.result_kind', 'not-found') + return { + kind: 'not-found', + runId: args.runId, + workflowId: args.workflowId, + eventCount: 0, + events: [], + } + } + if (claim.kind === 'not-claimable') { + span.setAttribute('tanstack.workflow.result_kind', 'not-claimable') + return { + kind: 'not-claimable', + runId: args.runId, + workflowId: args.workflowId, + run: claim.run, + eventCount: 0, + events: [], + } + } + + const runStore = createRunStoreAdapter(config.store, telemetry) + const collected = await (async () => { + try { + const events = await collectWorkflowEvents( + runWorkflow({ + workflow: args.workflow, + runStore, + runId: args.runId, + input: args.input, + signalDelivery: args.signalDelivery, + approval: args.approval, + threadId: args.threadId, + telemetry: config.telemetry, + deadline: args.deadline, + minYieldRemainingMs: args.minYieldRemainingMs, + yieldResumeAt: args.yieldResumeAt, + }), + { + includeEvents: args.includeEvents ?? true, + maxEvents: args.maxEvents, + }, + ) + + await syncTimerFromRunState( + config, + telemetry, + args.runId, + args.workflowId, + args.now, + ) + return events + } finally { + await traceStoreOperation( + telemetry, + 'store.release_run_lease', + { + workflowId: args.workflowId, + workflowVersion: args.workflow.version, + runId: args.runId, + leaseOwner, + }, + () => + config.store.releaseRunLease({ runId: args.runId, leaseOwner }), + ) + } + })() + + const run = await traceStoreOperation( + telemetry, + 'store.load_run', + { + workflowId: args.workflowId, + workflowVersion: args.workflow.version, + runId: args.runId, + }, + () => config.store.loadRun(args.runId), + ) + const result: WorkflowRuntimeRunResult = { + kind: classifyRun(run, collected.eventCount), + runId: args.runId, + workflowId: args.workflowId, + run, + events: collected.events, + eventCount: collected.eventCount, + eventsTruncated: collected.eventsTruncated || undefined, + } + span.setAttribute('tanstack.workflow.result_kind', result.kind) + span.setAttribute('tanstack.workflow.event_count', result.eventCount) + if (result.eventsTruncated !== undefined) { + span.setAttribute( + 'tanstack.workflow.events_truncated', + result.eventsTruncated, + ) + } + return result }, ) - - await syncTimerFromRunState(config, args.runId, args.workflowId, args.now) - await config.store.releaseRunLease({ runId: args.runId, leaseOwner }) - - const run = await config.store.loadRun(args.runId) - return { - kind: classifyRun(run, collected.eventCount), - runId: args.runId, - workflowId: args.workflowId, - run, - events: collected.events, - eventCount: collected.eventCount, - eventsTruncated: collected.eventsTruncated || undefined, - } } async function syncTimerFromRunState< TWorkflows extends Record, >( config: WorkflowRuntimeConfig, + telemetry: WorkflowTelemetry, runId: string, workflowId: string, now: number, ) { - const state = await config.store.loadRunState(runId) - const deadline = state?.waitingFor?.deadline - if (state?.waitingFor?.signalName !== '__timer' || deadline === undefined) { + const state = await traceStoreOperation( + telemetry, + 'store.load_run_state', + { workflowId, runId }, + () => config.store.loadRunState(runId), + ) + const waitingFor = state?.waitingFor + const deadline = waitingFor?.deadline + if ( + !state || + waitingFor?.signalName !== '__timer' || + deadline === undefined + ) { return } - await config.store.scheduleTimer({ - runId, - workflowId, - workflowVersion: state.workflowVersion, - wakeAt: deadline, - signalId: `timer:${runId}:${deadline}`, - now, - }) + await traceStoreOperation( + telemetry, + 'store.schedule_timer', + { workflowId, workflowVersion: state.workflowVersion, runId }, + () => + config.store.scheduleTimer({ + runId, + workflowId, + workflowVersion: state.workflowVersion, + wakeAt: deadline, + signalId: `timer:${runId}:${waitingFor.stepId}:${deadline}`, + now, + }), + ) } async function loadWorkflow< @@ -413,6 +783,23 @@ function normalizeWorkflowLoaderResult( return result.workflow } +async function traceStoreOperation( + telemetry: WorkflowTelemetry, + operation: string, + spanContext: { + workflowId?: string + workflowVersion?: string + runId?: string + signalName?: string + scheduleId?: string + bucketId?: string + leaseOwner?: string + }, + fn: () => Promise, +) { + return await telemetry.startActiveSpan(operation, spanContext, fn) +} + function resultFromSignalDelivery( runId: string, result: Exclude, @@ -427,6 +814,32 @@ function resultFromSignalDelivery( } } +function resultFromExistingRun( + run: WorkflowExecution, +): WorkflowRuntimeRunResult { + return { + kind: 'not-claimable', + runId: run.runId, + workflowId: run.workflowId, + run, + events: [], + eventCount: 0, + } +} + +function resultFromRunSnapshot( + run: WorkflowExecution, +): WorkflowRuntimeRunResult { + return { + kind: classifyRun(run, 0), + runId: run.runId, + workflowId: run.workflowId, + run, + events: [], + eventCount: 0, + } +} + function resultFromApprovalDelivery( runId: string, result: Exclude, @@ -464,11 +877,46 @@ function normalizeSweepLimit( return limit } -function isPastSweepDeadline( +function resolveRuntimeDeadline( + args: { deadline?: number; maxDurationMs?: number }, startedAt: number, - maxDurationMs: number | undefined, ) { - return maxDurationMs !== undefined && Date.now() - startedAt >= maxDurationMs + if (args.deadline !== undefined && !Number.isFinite(args.deadline)) { + throw new Error('Workflow runtime deadline must be a finite timestamp.') + } + if ( + args.maxDurationMs !== undefined && + (!Number.isFinite(args.maxDurationMs) || args.maxDurationMs < 0) + ) { + throw new Error( + 'Workflow runtime maxDurationMs must be a non-negative finite number.', + ) + } + + const durationDeadline = + args.maxDurationMs === undefined + ? undefined + : startedAt + args.maxDurationMs + if (args.deadline === undefined) return durationDeadline + if (durationDeadline === undefined) return args.deadline + return Math.min(args.deadline, durationDeadline) +} + +function normalizeMinYieldRemainingMs(value: number | undefined) { + const normalized = value ?? DEFAULT_MIN_YIELD_REMAINING_MS + if (!Number.isFinite(normalized) || normalized < 0) { + throw new Error( + 'Workflow runtime minYieldRemainingMs must be a non-negative finite number.', + ) + } + return normalized +} + +function shouldStopForDeadline( + deadline: number | undefined, + minYieldRemainingMs: number, +) { + return deadline !== undefined && deadline - Date.now() <= minYieldRemainingMs } function summarizeSweep( diff --git a/packages/workflow-runtime/src/types.ts b/packages/workflow-runtime/src/types.ts index 792fac9..593e639 100644 --- a/packages/workflow-runtime/src/types.ts +++ b/packages/workflow-runtime/src/types.ts @@ -8,6 +8,7 @@ import type { SerializedError, SignalDelivery, WorkflowEvent, + WorkflowTelemetryOptions, } from '@tanstack/workflow-core' export type WorkflowId = string @@ -357,6 +358,7 @@ export interface WorkflowRuntimeConfig< workflows: TWorkflows store: WorkflowExecutionStore defaultLeaseMs?: number + telemetry?: false | WorkflowTelemetryOptions } export interface WorkflowRuntimeDefinition< @@ -382,6 +384,9 @@ export interface WorkflowRuntimeStartRunArgs { runId: RunId input: unknown now?: number + deadline?: number + maxDurationMs?: number + minYieldRemainingMs?: number leaseOwner?: LeaseOwner leaseMs?: number threadId?: string @@ -397,6 +402,9 @@ export interface WorkflowRuntimeDeliverSignalArgs { payload: TPayload meta?: Record now?: number + deadline?: number + maxDurationMs?: number + minYieldRemainingMs?: number leaseOwner?: LeaseOwner leaseMs?: number threadId?: string @@ -408,6 +416,9 @@ export interface WorkflowRuntimeDeliverApprovalArgs { runId: RunId approval: ApprovalResult now?: number + deadline?: number + maxDurationMs?: number + minYieldRemainingMs?: number leaseOwner?: LeaseOwner leaseMs?: number threadId?: string @@ -440,7 +451,9 @@ export interface WorkflowRuntimeSweepArgs { limit?: number maxScheduledRuns?: number maxTimers?: number + deadline?: number maxDurationMs?: number + minYieldRemainingMs?: number leaseOwner?: LeaseOwner leaseMs?: number includeEvents?: boolean diff --git a/packages/workflow-runtime/tests/runtime-driver.test.ts b/packages/workflow-runtime/tests/runtime-driver.test.ts index a17f46d..ef604da 100644 --- a/packages/workflow-runtime/tests/runtime-driver.test.ts +++ b/packages/workflow-runtime/tests/runtime-driver.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it } from 'vitest' +import { describe, expect, it, vi } from 'vitest' import { createWorkflow } from '@tanstack/workflow-core' import { defineWorkflowRuntime, inMemoryWorkflowExecutionStore } from '../src' @@ -85,6 +85,525 @@ describe('workflow runtime driver', () => { expect(cappedEvents.eventsTruncated).toBe(true) }) + it('exposes runtime deadline helpers on handler and step contexts', async () => { + const store = inMemoryWorkflowExecutionStore() + const deadline = Date.now() + 60_000 + const workflow = createWorkflow({ id: 'deadline-context' }).handler( + async (ctx) => { + return ctx.step('read-deadline', (stepCtx) => ({ + handlerDeadline: ctx.runtime.deadline, + stepDeadline: stepCtx.runtime.deadline, + handlerRemaining: ctx.runtime.timeRemaining(), + stepRemaining: stepCtx.runtime.timeRemaining(), + shouldYieldWithSmallBuffer: ctx.runtime.shouldYield({ + minRemainingMs: 1, + }), + shouldYieldWithLargeBuffer: stepCtx.runtime.shouldYield({ + minRemainingMs: 120_000, + }), + })) + }, + ) + const runtime = defineWorkflowRuntime({ + store, + workflows: { + 'deadline-context': { + load: async () => { + await Promise.resolve() + return workflow + }, + }, + }, + }) + + const result = await runtime.startRun({ + workflowId: 'deadline-context', + runId: 'deadline-context:1', + input: {}, + deadline, + }) + + expect(result.kind).toBe('completed') + expect(result.run?.output).toMatchObject({ + handlerDeadline: deadline, + stepDeadline: deadline, + shouldYieldWithSmallBuffer: false, + shouldYieldWithLargeBuffer: true, + }) + expect( + (result.run?.output as { handlerRemaining?: number }).handlerRemaining, + ).toBeGreaterThan(0) + expect( + (result.run?.output as { stepRemaining?: number }).stepRemaining, + ).toBeGreaterThan(0) + }) + + it('explicit yield persists, releases the lease, and resumes on a later sweep', async () => { + const store = inMemoryWorkflowExecutionStore() + let executions = 0 + const workflow = createWorkflow({ id: 'manual-yield' }).handler( + async (ctx) => { + const first = await ctx.step('first', () => ++executions) + await ctx.runtime.yield({ reason: 'manual' }) + const second = await ctx.step('second', () => ++executions) + return { first, second } + }, + ) + const runtime = defineWorkflowRuntime({ + store, + workflows: { + 'manual-yield': { + load: async () => { + await Promise.resolve() + return workflow + }, + }, + }, + }) + + const start = await runtime.startRun({ + workflowId: 'manual-yield', + runId: 'manual-yield:1', + input: {}, + now: 100, + leaseOwner: 'worker-a', + }) + const yielded = await store.loadRun('manual-yield:1') + const resumed = await runtime.sweep({ + now: 101, + leaseOwner: 'worker-b', + }) + + expect(start.kind).toBe('paused') + expect(yielded).toMatchObject({ + status: 'paused', + lease: undefined, + waitingFor: { signalName: '__timer', deadline: 101 }, + }) + expect(resumed.timers[0]).toMatchObject({ + kind: 'completed', + runId: 'manual-yield:1', + }) + expect(resumed.timers[0]?.run?.output).toEqual({ first: 1, second: 2 }) + expect(executions).toBe(2) + }) + + it('does not turn a yielded run back into a stale running row on duplicate start', async () => { + const store = inMemoryWorkflowExecutionStore() + const workflow = createWorkflow({ id: 'yield-retry' }).handler( + async (ctx) => { + await ctx.runtime.yield() + return { done: true } + }, + ) + const runtime = defineWorkflowRuntime({ + store, + workflows: { + 'yield-retry': { load: async () => workflow }, + }, + }) + + const first = await runtime.startRun({ + workflowId: 'yield-retry', + runId: 'yield-retry:1', + input: {}, + now: 100, + }) + const duplicate = await runtime.startRun({ + workflowId: 'yield-retry', + runId: 'yield-retry:1', + input: {}, + now: 100, + }) + + expect(first.kind).toBe('paused') + expect(duplicate.kind).toBe('not-claimable') + expect(await store.loadRun('yield-retry:1')).toMatchObject({ + status: 'paused', + lease: undefined, + }) + }) + + it('releases the lease when scheduling a yielded run fails', async () => { + const store = inMemoryWorkflowExecutionStore() + store.scheduleTimer = async () => { + throw new Error('timer store unavailable') + } + const workflow = createWorkflow({ id: 'yield-schedule-failure' }).handler( + async (ctx) => { + await ctx.runtime.yield() + return {} + }, + ) + const runtime = defineWorkflowRuntime({ + store, + workflows: { + 'yield-schedule-failure': { load: async () => workflow }, + }, + }) + + await expect( + runtime.startRun({ + workflowId: 'yield-schedule-failure', + runId: 'yield-schedule-failure:1', + input: {}, + now: 100, + }), + ).rejects.toThrow('timer store unavailable') + + expect(await store.loadRun('yield-schedule-failure:1')).toMatchObject({ + status: 'paused', + lease: undefined, + }) + }) + + it('automatically yields before fresh step work when the runtime deadline is exhausted', async () => { + const store = inMemoryWorkflowExecutionStore() + let executions = 0 + const workflow = createWorkflow({ id: 'auto-yield' }).handler( + async (ctx) => { + const value = await ctx.step('fresh-work', () => ++executions) + return { value } + }, + ) + const runtime = defineWorkflowRuntime({ + store, + workflows: { + 'auto-yield': { + load: async () => { + await Promise.resolve() + return workflow + }, + }, + }, + }) + + const start = await runtime.startRun({ + workflowId: 'auto-yield', + runId: 'auto-yield:1', + input: {}, + now: 200, + deadline: Date.now() - 1, + leaseOwner: 'worker-a', + }) + const yielded = await store.loadRun('auto-yield:1') + + expect(start.kind).toBe('paused') + expect(executions).toBe(0) + expect(yielded).toMatchObject({ + status: 'paused', + lease: undefined, + waitingFor: { signalName: '__timer', deadline: 201 }, + }) + + const resumed = await runtime.sweep({ + now: 201, + deadline: Date.now() + 60_000, + leaseOwner: 'worker-b', + }) + + expect(resumed.timers[0]?.kind).toBe('completed') + expect(resumed.timers[0]?.run?.output).toEqual({ value: 1 }) + expect(executions).toBe(1) + }) + + it('stays paused when workflow code catches the automatic yield sentinel', async () => { + const store = inMemoryWorkflowExecutionStore() + let executions = 0 + let caught = 0 + const workflow = createWorkflow({ id: 'caught-auto-yield' }).handler( + async (ctx) => { + try { + await ctx.step('work', () => { + executions++ + return 'done' + }) + } catch { + caught++ + } + return 'completed' + }, + ) + const runtime = defineWorkflowRuntime({ + store, + workflows: { + 'caught-auto-yield': { load: async () => workflow }, + }, + }) + + const paused = await runtime.startRun({ + workflowId: 'caught-auto-yield', + runId: 'caught-auto-yield:1', + input: {}, + now: 100, + deadline: Date.now() - 1, + }) + + expect(paused.kind).toBe('paused') + expect(executions).toBe(0) + expect(caught).toBe(1) + + const resumed = await runtime.sweep({ + now: 101, + deadline: Date.now() + 60_000, + }) + + expect(resumed.timers[0]?.kind).toBe('completed') + expect(resumed.timers[0]?.run?.output).toBe('completed') + expect(executions).toBe(1) + expect(caught).toBe(1) + }) + + it('keeps automatic and explicit yield checkpoints independent', async () => { + const store = inMemoryWorkflowExecutionStore() + let executions = 0 + const workflow = createWorkflow({ id: 'mixed-yield' }).handler( + async (ctx) => { + const first = await ctx.step('first', () => ++executions) + await ctx.runtime.yield({ reason: 'manual' }) + const second = await ctx.step('second', () => ++executions) + return { first, second } + }, + ) + const runtime = defineWorkflowRuntime({ + store, + workflows: { + 'mixed-yield': { + load: async () => workflow, + }, + }, + }) + + const automaticYield = await runtime.startRun({ + workflowId: 'mixed-yield', + runId: 'mixed-yield:1', + input: {}, + now: 300, + deadline: Date.now() - 1, + }) + const explicitYield = await runtime.sweep({ + now: 301, + deadline: Date.now() + 60_000, + }) + + expect(automaticYield.kind).toBe('paused') + expect(explicitYield.timers[0]?.kind).toBe('paused') + expect(executions).toBe(1) + + const completed = await runtime.sweep({ + now: 302, + deadline: Date.now() + 60_000, + }) + + expect(completed.timers[0]?.kind).toBe('completed') + expect(completed.timers[0]?.run?.output).toEqual({ first: 1, second: 2 }) + expect(executions).toBe(2) + }) + + it('drains unknown-length work until the deadline margin, then resumes without a batch size', async () => { + let wallNow = 1_000 + const nowSpy = vi.spyOn(Date, 'now').mockImplementation(() => wallNow) + + try { + const store = inMemoryWorkflowExecutionStore() + const queue = ['a', 'b', 'c', 'd'] + const processed: Array = [] + const workflow = createWorkflow({ id: 'queue-drain' }).handler( + async (ctx) => { + for (let index = 0; ; index++) { + const item = await ctx.step(`take-${index}`, () => { + return queue.shift() ?? null + }) + if (item === null) return { processed } + await ctx.step(`process-${item}`, () => { + processed.push(item) + wallNow += 600 + }) + } + }, + ) + const runtime = defineWorkflowRuntime({ + store, + workflows: { + 'queue-drain': { load: async () => workflow }, + }, + }) + + const first = await runtime.startRun({ + workflowId: 'queue-drain', + runId: 'queue-drain:1', + input: {}, + now: 10, + deadline: 2_500, + minYieldRemainingMs: 500, + }) + + expect(first.kind).toBe('paused') + expect(processed).toEqual(['a', 'b']) + + const resumed = await runtime.sweep({ + now: 11, + deadline: 10_000, + minYieldRemainingMs: 500, + }) + + expect(resumed.timers[0]?.kind).toBe('completed') + expect(resumed.timers[0]?.run?.output).toEqual({ + processed: ['a', 'b', 'c', 'd'], + }) + } finally { + nowSpy.mockRestore() + } + }) + + it('does not resume a sweep-yielded run again in the same sweep', async () => { + const store = inMemoryWorkflowExecutionStore() + let executions = 0 + const workflow = createWorkflow({ id: 'sweep-yield' }).handler( + async (ctx) => { + await ctx.runtime.yield() + const value = await ctx.step('fresh-work', () => ++executions) + return { value } + }, + ) + const runtime = defineWorkflowRuntime({ + store, + workflows: { + 'sweep-yield': { + load: async () => { + await Promise.resolve() + return workflow + }, + }, + }, + }) + await store.upsertSchedule({ + scheduleId: 'sweep-yield-now', + workflowId: 'sweep-yield', + schedule: { kind: 'interval', everyMs: 60_000 }, + overlapPolicy: 'skip', + input: {}, + nextFireAt: 500, + enabled: true, + now: 0, + }) + + const first = await runtime.sweep({ + now: 500, + leaseOwner: 'worker-a', + }) + + expect(first.scheduled[0]?.kind).toBe('paused') + expect(first.timers).toEqual([]) + expect(executions).toBe(0) + + const second = await runtime.sweep({ + now: 501, + deadline: Date.now() + 60_000, + leaseOwner: 'worker-b', + }) + + expect(second.timers[0]?.kind).toBe('completed') + expect(executions).toBe(1) + }) + + it('keeps sequential timers at the same timestamp independently deliverable', async () => { + const store = inMemoryWorkflowExecutionStore() + const workflow = createWorkflow({ id: 'same-time-timers' }).handler( + async (ctx) => { + await ctx.sleepUntil(700, { id: 'first' }) + await ctx.sleepUntil(700, { id: 'second' }) + return { done: true } + }, + ) + const runtime = defineWorkflowRuntime({ + store, + workflows: { + 'same-time-timers': { load: async () => workflow }, + }, + }) + + const started = await runtime.startRun({ + workflowId: 'same-time-timers', + runId: 'same-time-timers:1', + input: {}, + now: 700, + }) + const wake = await runtime.sweep({ now: 700 }) + + expect(started.kind).toBe('paused') + expect(wake.timers.map((result) => result.kind)).toEqual([ + 'paused', + 'completed', + ]) + }) + + it('stops a sweep before claiming work inside the cooperative margin', async () => { + const store = inMemoryWorkflowExecutionStore() + const workflow = createWorkflow({ id: 'margin-stop' }).handler(async () => { + return { done: true } + }) + const runtime = defineWorkflowRuntime({ + store, + workflows: { + 'margin-stop': { load: async () => workflow }, + }, + }) + await store.upsertSchedule({ + scheduleId: 'margin-stop-now', + workflowId: 'margin-stop', + schedule: { kind: 'interval', everyMs: 60_000 }, + overlapPolicy: 'skip', + input: {}, + nextFireAt: 600, + enabled: true, + now: 0, + }) + + const result = await runtime.sweep({ + now: 600, + deadline: Date.now() + 60_000, + minYieldRemainingMs: 120_000, + }) + + expect(result.scheduled).toEqual([]) + expect(result.deadlineReached).toBe(true) + expect(result.remainingMayExist).toBe(true) + expect( + await store.loadRun('margin-stop:margin-stop-now:600'), + ).toBeUndefined() + }) + + it('uses the earlier of an absolute deadline and max duration', async () => { + const store = inMemoryWorkflowExecutionStore() + let executions = 0 + const workflow = createWorkflow({ id: 'earliest-deadline' }).handler( + async (ctx) => { + await ctx.step('work', () => ++executions) + return {} + }, + ) + const runtime = defineWorkflowRuntime({ + store, + workflows: { + 'earliest-deadline': { load: async () => workflow }, + }, + }) + + const result = await runtime.startRun({ + workflowId: 'earliest-deadline', + runId: 'earliest-deadline:1', + input: {}, + now: 700, + deadline: Date.now() + 60_000, + maxDurationMs: 0, + minYieldRemainingMs: 0, + }) + + expect(result.kind).toBe('paused') + expect(executions).toBe(0) + }) + it('delivers a signal and resumes a paused workflow', async () => { const store = inMemoryWorkflowExecutionStore() const workflow = createWorkflow({ id: 'signal' }).handler(async (ctx) => { diff --git a/packages/workflow-runtime/tests/telemetry.test.ts b/packages/workflow-runtime/tests/telemetry.test.ts new file mode 100644 index 0000000..b047f48 --- /dev/null +++ b/packages/workflow-runtime/tests/telemetry.test.ts @@ -0,0 +1,292 @@ +import { describe, expect, it } from 'vitest' +import { createWorkflow } from '@tanstack/workflow-core' +import { defineWorkflowRuntime, inMemoryWorkflowExecutionStore } from '../src' +import type { + Attributes, + Context, + Link, + Span, + SpanContext, + SpanOptions, + SpanStatus, + TimeInput, + Tracer, +} from '@opentelemetry/api' + +describe('workflow runtime OpenTelemetry tracing', () => { + it('records runtime, store, drive, and step spans with safe attributes', async () => { + const tracer = new TestTracer() + const store = inMemoryWorkflowExecutionStore() + const workflow = createWorkflow({ + id: 'otel-runtime', + version: 'v1', + }).handler(async (ctx) => { + await ctx.step('reserve', () => ({ secret: 'never' })) + return { ok: true } + }) + const runtime = defineWorkflowRuntime({ + store, + telemetry: { + tracer, + attributes: ({ workflowId }) => ({ + 'app.workflow_seen': workflowId ?? 'unknown', + }), + }, + workflows: { + 'otel-runtime': { + load: async () => workflow, + }, + }, + }) + + const result = await runtime.startRun({ + workflowId: 'otel-runtime', + runId: 'otel-runtime:1', + input: { secret: 'input' }, + now: 0, + leaseOwner: 'test-worker', + }) + + expect(result.kind).toBe('completed') + expect(tracer.spans.map((span) => span.name)).toEqual( + expect.arrayContaining([ + 'tanstack.workflow.start_run', + 'tanstack.workflow.drive_run', + 'tanstack.workflow.store.create_run', + 'tanstack.workflow.store.claim_run', + 'tanstack.workflow.store.append_events', + 'tanstack.workflow.store.release_run_lease', + 'tanstack.workflow.store.load_run', + 'tanstack.workflow.step', + ]), + ) + const startSpan = tracer.spans.find( + (span) => span.name === 'tanstack.workflow.start_run', + ) + expect(startSpan?.attributes).toMatchObject({ + 'tanstack.workflow.workflow_id': 'otel-runtime', + 'tanstack.workflow.workflow_version': 'v1', + 'tanstack.workflow.run_id': 'otel-runtime:1', + 'tanstack.workflow.lease_owner': 'test-worker', + 'tanstack.workflow.result_kind': 'completed', + 'app.workflow_seen': 'otel-runtime', + }) + expect( + JSON.stringify(tracer.spans.map((span) => span.attributes)), + ).not.toContain('secret') + }) + + it('can disable telemetry', async () => { + const tracer = new TestTracer() + const workflow = createWorkflow({ id: 'otel-disabled' }).handler( + async () => ({}), + ) + const runtime = defineWorkflowRuntime({ + store: inMemoryWorkflowExecutionStore(), + telemetry: { tracer, enabled: false }, + workflows: { + 'otel-disabled': { + load: async () => workflow, + }, + }, + }) + + await runtime.startRun({ + workflowId: 'otel-disabled', + runId: 'otel-disabled:1', + input: {}, + now: 0, + }) + + expect(tracer.spans).toEqual([]) + }) + + it('records signal and sweep spans', async () => { + const tracer = new TestTracer() + const store = inMemoryWorkflowExecutionStore() + const signalWorkflow = createWorkflow({ id: 'signal' }).handler( + async (ctx) => ctx.waitForEvent('go'), + ) + const timerWorkflow = createWorkflow({ id: 'timer' }).handler( + async (ctx) => { + await ctx.sleepUntil(10) + return { ok: true } + }, + ) + const runtime = defineWorkflowRuntime({ + store, + telemetry: { tracer }, + workflows: { + signal: { load: async () => signalWorkflow }, + timer: { load: async () => timerWorkflow }, + }, + }) + + await runtime.startRun({ + workflowId: 'signal', + runId: 'signal:1', + input: {}, + now: 0, + }) + await runtime.deliverSignal({ + runId: 'signal:1', + signalId: 'go:1', + name: 'go', + payload: { secret: 'payload' }, + now: 1, + }) + await runtime.startRun({ + workflowId: 'timer', + runId: 'timer:1', + input: {}, + now: 0, + }) + await runtime.sweep({ now: 10, leaseOwner: 'sweep-worker' }) + + expect(tracer.spans.map((span) => span.name)).toEqual( + expect.arrayContaining([ + 'tanstack.workflow.deliver_signal', + 'tanstack.workflow.sweep', + 'tanstack.workflow.store.claim_due_timers', + ]), + ) + const signalSpan = tracer.spans.find( + (span) => span.name === 'tanstack.workflow.deliver_signal', + ) + expect(signalSpan?.attributes).toMatchObject({ + 'tanstack.workflow.run_id': 'signal:1', + 'tanstack.workflow.signal_name': 'go', + 'tanstack.workflow.result_kind': 'completed', + }) + expect(JSON.stringify(signalSpan?.attributes)).not.toContain('payload') + }) + + it('marks store operation failures as span errors and rethrows', async () => { + const tracer = new TestTracer() + const store = inMemoryWorkflowExecutionStore() + const originalClaimRun = store.claimRun + store.claimRun = async (args) => { + if (args.runId === 'broken:1') throw new Error('claim failed') + return await originalClaimRun(args) + } + const workflow = createWorkflow({ id: 'broken' }).handler(async () => ({})) + const runtime = defineWorkflowRuntime({ + store, + telemetry: { tracer }, + workflows: { + broken: { + load: async () => workflow, + }, + }, + }) + + await expect( + runtime.startRun({ + workflowId: 'broken', + runId: 'broken:1', + input: {}, + now: 0, + }), + ).rejects.toThrow('claim failed') + + const claimSpan = tracer.spans.find( + (span) => span.name === 'tanstack.workflow.store.claim_run', + ) + expect(claimSpan?.status.code).toBe(2) + expect(claimSpan?.exceptions).toHaveLength(1) + }) +}) + +class TestTracer implements Tracer { + spans: Array = [] + + startSpan(name: string, options?: SpanOptions): Span { + const span = new TestSpan(name, options?.attributes) + this.spans.push(span) + return span + } + + startActiveSpan unknown>( + name: string, + optionsOrFn: SpanOptions | F, + contextOrFn?: Context | F, + maybeFn?: F, + ): ReturnType { + const options = typeof optionsOrFn === 'function' ? undefined : optionsOrFn + const fn = + typeof optionsOrFn === 'function' + ? optionsOrFn + : typeof contextOrFn === 'function' + ? contextOrFn + : maybeFn + if (!fn) throw new Error('Missing span callback') + return fn(this.startSpan(name, options)) as ReturnType + } +} + +class TestSpan implements Span { + attributes: Attributes = {} + events: Array<{ name: string; attributes?: Attributes }> = [] + exceptions: Array = [] + status: SpanStatus = { code: 0 } + ended = false + + constructor( + readonly name: string, + attributes: Attributes | undefined, + ) { + if (attributes) this.attributes = { ...attributes } + } + + spanContext(): SpanContext { + return { + traceId: '00000000000000000000000000000001', + spanId: '0000000000000001', + traceFlags: 1, + } + } + + setAttribute(key: string, value: Attributes[string]): this { + this.attributes[key] = value + return this + } + + setAttributes(attributes: Attributes): this { + Object.assign(this.attributes, attributes) + return this + } + + addEvent(name: string, attributes?: Attributes): this { + this.events.push({ name, attributes }) + return this + } + + addLink(_link: Link): this { + return this + } + + addLinks(_links: Array): this { + return this + } + + setStatus(status: SpanStatus): this { + this.status = status + return this + } + + updateName(_name: string): this { + return this + } + + end(_endTime?: TimeInput): void { + this.ended = true + } + + isRecording(): boolean { + return true + } + + recordException(exception: unknown): void { + this.exceptions.push(exception) + } +} diff --git a/packages/workflow-store-cloudflare-d1/package.json b/packages/workflow-store-cloudflare-d1/package.json index 8115597..86a5f36 100644 --- a/packages/workflow-store-cloudflare-d1/package.json +++ b/packages/workflow-store-cloudflare-d1/package.json @@ -1,6 +1,6 @@ { "name": "@tanstack/workflow-store-cloudflare-d1", - "version": "0.0.2", + "version": "0.0.3", "description": "Cloudflare D1 durable execution store for TanStack Workflow.", "author": "Tanner Linsley", "license": "MIT", diff --git a/packages/workflow-store-drizzle-postgres/package.json b/packages/workflow-store-drizzle-postgres/package.json index cca40a8..963634e 100644 --- a/packages/workflow-store-drizzle-postgres/package.json +++ b/packages/workflow-store-drizzle-postgres/package.json @@ -1,6 +1,6 @@ { "name": "@tanstack/workflow-store-drizzle-postgres", - "version": "0.0.3", + "version": "0.0.4", "description": "Drizzle/Postgres durable execution store for TanStack Workflow.", "author": "Tanner Linsley", "license": "MIT", diff --git a/packages/workflow-vercel/package.json b/packages/workflow-vercel/package.json index 7361679..c47ed91 100644 --- a/packages/workflow-vercel/package.json +++ b/packages/workflow-vercel/package.json @@ -1,6 +1,6 @@ { "name": "@tanstack/workflow-vercel", - "version": "0.0.2", + "version": "0.0.3", "description": "Vercel host adapter for TanStack Workflow runtimes.", "author": "Tanner Linsley", "license": "MIT", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 5cafc4f..45ea457 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -76,7 +76,7 @@ importers: version: 6.0.3 vitest: specifier: ^4.1.5 - version: 4.1.5(@types/node@25.6.0)(happy-dom@20.9.0)(vite@8.0.10(@types/node@25.6.0)(esbuild@0.28.0)(jiti@2.6.1)(yaml@2.8.2)) + version: 4.1.5(@opentelemetry/api@1.9.1)(@types/node@25.6.0)(happy-dom@20.9.0)(vite@8.0.10(@types/node@25.6.0)(esbuild@0.28.0)(jiti@2.6.1)(yaml@2.8.2)) examples/deployment-pocs/cloudflare-d1: dependencies: @@ -119,6 +119,9 @@ importers: specifier: ^1.1.0 version: 1.1.0 devDependencies: + '@opentelemetry/api': + specifier: ^1.9.0 + version: 1.9.1 zod: specifier: ^4.2.0 version: 4.3.6 @@ -148,6 +151,10 @@ importers: '@tanstack/workflow-core': specifier: workspace:* version: link:../workflow-core + devDependencies: + '@opentelemetry/api': + specifier: ^1.9.0 + version: 1.9.1 packages/workflow-store-cloudflare-d1: dependencies: @@ -168,7 +175,7 @@ importers: version: link:../workflow-runtime drizzle-orm: specifier: ^0.45.1 - version: 0.45.2(@cloudflare/workers-types@4.20260530.1)(@electric-sql/pglite@0.3.16) + version: 0.45.2(@cloudflare/workers-types@4.20260530.1)(@electric-sql/pglite@0.3.16)(@opentelemetry/api@1.9.1) devDependencies: '@electric-sql/pglite': specifier: ^0.3.14 @@ -1009,6 +1016,10 @@ packages: cpu: [x64] os: [win32] + '@opentelemetry/api@1.9.1': + resolution: {integrity: sha512-gLyJlPHPZYdAk1JENA9LeHejZe1Ti77/pTeFm/nMXmQH/HFZlcS/O2XJB+L8fkbrNSqhdtlvjBVjxwUYanNH5Q==} + engines: {node: '>=8.0.0'} + '@oxc-parser/binding-android-arm-eabi@0.127.0': resolution: {integrity: sha512-0LC7ye4hvqbIKxAzThzvswgHLFu2AURKzYLeSVvLdu2TBOYWQDmHnTqPLeA597BcUCxiLqLsS4CJ5uoI5WYWCQ==} engines: {node: ^20.19.0 || >=22.12.0} @@ -4281,6 +4292,8 @@ snapshots: '@nx/nx-win32-x64-msvc@22.7.0': optional: true + '@opentelemetry/api@1.9.1': {} + '@oxc-parser/binding-android-arm-eabi@0.127.0': optional: true @@ -5053,10 +5066,11 @@ snapshots: dotenv@16.6.1: {} - drizzle-orm@0.45.2(@cloudflare/workers-types@4.20260530.1)(@electric-sql/pglite@0.3.16): + drizzle-orm@0.45.2(@cloudflare/workers-types@4.20260530.1)(@electric-sql/pglite@0.3.16)(@opentelemetry/api@1.9.1): optionalDependencies: '@cloudflare/workers-types': 4.20260530.1 '@electric-sql/pglite': 0.3.16 + '@opentelemetry/api': 1.9.1 dts-resolver@2.1.3(oxc-resolver@11.19.1): optionalDependencies: @@ -6554,7 +6568,7 @@ snapshots: jiti: 2.6.1 yaml: 2.8.2 - vitest@4.1.5(@types/node@25.6.0)(happy-dom@20.9.0)(vite@8.0.10(@types/node@25.6.0)(esbuild@0.28.0)(jiti@2.6.1)(yaml@2.8.2)): + vitest@4.1.5(@opentelemetry/api@1.9.1)(@types/node@25.6.0)(happy-dom@20.9.0)(vite@8.0.10(@types/node@25.6.0)(esbuild@0.28.0)(jiti@2.6.1)(yaml@2.8.2)): dependencies: '@vitest/expect': 4.1.5 '@vitest/mocker': 4.1.5(vite@8.0.10(@types/node@25.6.0)(esbuild@0.28.0)(jiti@2.6.1)(yaml@2.8.2)) @@ -6577,6 +6591,7 @@ snapshots: vite: 8.0.10(@types/node@25.6.0)(esbuild@0.28.0)(jiti@2.6.1)(yaml@2.8.2) why-is-node-running: 2.3.0 optionalDependencies: + '@opentelemetry/api': 1.9.1 '@types/node': 25.6.0 happy-dom: 20.9.0 transitivePeerDependencies: