From 02825c21f9cccc6403f967a1d7a7a457cfec5d4b Mon Sep 17 00:00:00 2001 From: aXenDeveloper Date: Wed, 26 Aug 2026 17:41:44 +0200 Subject: [PATCH] feat: Add workflow stage 0 --- .../0001-workflow-engine-wave-1-ownership.md | 176 +++++ docs/architecture/0001-workflow-engine.md | 625 ++++++++++++++++++ docs/architecture/README.md | 10 + packages/vitnode/src/api/lib/module.ts | 17 + packages/vitnode/src/api/lib/plugin.test.ts | 77 +++ packages/vitnode/src/api/lib/plugin.ts | 30 + .../src/api/middlewares/global.middleware.ts | 29 + packages/vitnode/src/api/models/workflow.ts | 130 ++++ .../queue/helpers/process-queue-tasks.ts | 74 +++ .../workflows/tasks/workflow-step.task.ts | 41 ++ .../api/modules/workflows/workflows.module.ts | 18 + packages/vitnode/src/api/plugin.ts | 2 + packages/vitnode/src/api/workflows/const.ts | 82 +++ .../src/api/workflows/define.test-d.ts | 102 +++ .../vitnode/src/api/workflows/define.test.ts | 182 +++++ packages/vitnode/src/api/workflows/define.ts | 173 +++++ packages/vitnode/src/api/workflows/errors.ts | 98 +++ .../src/api/workflows/idempotency.test.ts | 99 +++ .../vitnode/src/api/workflows/idempotency.ts | 93 +++ packages/vitnode/src/api/workflows/index.ts | 12 + .../vitnode/src/api/workflows/plan.test.ts | 184 ++++++ packages/vitnode/src/api/workflows/plan.ts | 120 ++++ .../vitnode/src/api/workflows/queue-task.ts | 26 + .../src/api/workflows/registry.test.ts | 225 +++++++ .../vitnode/src/api/workflows/registry.ts | 189 ++++++ .../vitnode/src/api/workflows/retry.test.ts | 109 +++ packages/vitnode/src/api/workflows/retry.ts | 135 ++++ .../src/api/workflows/state-machine.test.ts | 89 +++ .../src/api/workflows/state-machine.ts | 170 +++++ .../src/api/workflows/step-outputs.test.ts | 60 ++ .../vitnode/src/api/workflows/step-outputs.ts | 43 ++ packages/vitnode/src/api/workflows/store.ts | 34 + .../vitnode/src/api/workflows/triggers.ts | 132 ++++ packages/vitnode/src/api/workflows/types.ts | 369 +++++++++++ packages/vitnode/src/database/relations.ts | 13 + packages/vitnode/src/database/workflows.ts | 206 ++++++ .../lib/api/resolve-stale-queue-lease.test.ts | 65 ++ .../src/lib/api/resolve-stale-queue-lease.ts | 63 ++ 38 files changed, 4302 insertions(+) create mode 100644 docs/architecture/0001-workflow-engine-wave-1-ownership.md create mode 100644 docs/architecture/0001-workflow-engine.md create mode 100644 docs/architecture/README.md create mode 100644 packages/vitnode/src/api/models/workflow.ts create mode 100644 packages/vitnode/src/api/modules/workflows/tasks/workflow-step.task.ts create mode 100644 packages/vitnode/src/api/modules/workflows/workflows.module.ts create mode 100644 packages/vitnode/src/api/workflows/const.ts create mode 100644 packages/vitnode/src/api/workflows/define.test-d.ts create mode 100644 packages/vitnode/src/api/workflows/define.test.ts create mode 100644 packages/vitnode/src/api/workflows/define.ts create mode 100644 packages/vitnode/src/api/workflows/errors.ts create mode 100644 packages/vitnode/src/api/workflows/idempotency.test.ts create mode 100644 packages/vitnode/src/api/workflows/idempotency.ts create mode 100644 packages/vitnode/src/api/workflows/index.ts create mode 100644 packages/vitnode/src/api/workflows/plan.test.ts create mode 100644 packages/vitnode/src/api/workflows/plan.ts create mode 100644 packages/vitnode/src/api/workflows/queue-task.ts create mode 100644 packages/vitnode/src/api/workflows/registry.test.ts create mode 100644 packages/vitnode/src/api/workflows/registry.ts create mode 100644 packages/vitnode/src/api/workflows/retry.test.ts create mode 100644 packages/vitnode/src/api/workflows/retry.ts create mode 100644 packages/vitnode/src/api/workflows/state-machine.test.ts create mode 100644 packages/vitnode/src/api/workflows/state-machine.ts create mode 100644 packages/vitnode/src/api/workflows/step-outputs.test.ts create mode 100644 packages/vitnode/src/api/workflows/step-outputs.ts create mode 100644 packages/vitnode/src/api/workflows/store.ts create mode 100644 packages/vitnode/src/api/workflows/triggers.ts create mode 100644 packages/vitnode/src/api/workflows/types.ts create mode 100644 packages/vitnode/src/database/workflows.ts create mode 100644 packages/vitnode/src/lib/api/resolve-stale-queue-lease.test.ts create mode 100644 packages/vitnode/src/lib/api/resolve-stale-queue-lease.ts diff --git a/docs/architecture/0001-workflow-engine-wave-1-ownership.md b/docs/architecture/0001-workflow-engine-wave-1-ownership.md new file mode 100644 index 000000000..6ab5ca248 --- /dev/null +++ b/docs/architecture/0001-workflow-engine-wave-1-ownership.md @@ -0,0 +1,176 @@ +# Workflow Engine - Wave 1 ownership map + +Companion to [ADR 0001](./0001-workflow-engine.md). Wave 0 froze the contracts +and added the skeleton; this document says who implements what, and which files +nobody but the lead may touch while that happens. + +**Rule:** the contracts in ADR 0001 and in `src/api/workflows/types.ts`, +`const.ts` and `state-machine.ts` are settled. An agent that believes one is +wrong raises it with the lead rather than changing it - a unilateral change +breaks the other three agents silently. + +--- + +## Agent A - SDK + registry + +**Owns** + +```text +packages/vitnode/src/api/workflows/define.ts +packages/vitnode/src/api/workflows/registry.ts +packages/vitnode/src/api/workflows/step-outputs.ts +packages/vitnode/src/api/workflows/plan.ts +packages/vitnode/src/api/workflows/errors.ts +packages/vitnode/src/api/workflows/retry.ts +packages/vitnode/src/api/workflows/index.ts +packages/vitnode/src/api/workflows/*.test.ts (for the above) +apps/docs/content/docs/dev/workflows/** (new, user-facing docs) +``` + +**Scope** + +- Harden definition validation as real workflows appear. +- Registry ergonomics: listing, diagnostics, "which versions are deployed". +- The public `@vitnode/core/api/workflows` surface and its documentation. +- The first documented example workflow, in `plugins/example`. + +**Must not touch:** the two database tables, the store implementation, the +runner, the queue worker. + +**Depends on nothing.** Can start immediately. + +--- + +## Agent B - Database persistence + +**Owns** + +```text +packages/vitnode/src/database/workflows.ts +packages/vitnode/src/api/workflows/store.ts (replaces the throwing stub) +packages/vitnode/src/api/workflows/store.test.ts +apps/docs/migrations/** (the workflow migration) +``` + +**Scope** + +- Generate and commit the migration for the two tables. `apps/docs` uses + versioned migrations; `apps/api` uses `drizzle-kit push`. Never regenerate an + already-applied migration in place - the journal timestamp changes and the + migrator replays the file. +- Implement every `WorkflowStore` method against Drizzle. +- `createExecution` must honour the partial unique index: on conflict, return + the existing execution with `deduplicated: true` rather than throwing. +- `createExecution` must use `options.tx` when given, and dispatch the first + `workflow-step` task through `c.get("queue").dispatch({ ..., tx })` inside the + same unit of work. +- `claimStep` returning `undefined` is the normal answer for a duplicate + delivery, not an error. + +**Must not touch:** `types.ts` (the interface it implements), `define.ts`, +`registry.ts`, the runner, the queue worker. + +**Depends on nothing.** The interface is frozen; can start immediately. + +--- + +## Agent C - Queue reliability + runner plumbing + +**Owns** + +```text +packages/vitnode/src/api/modules/workflows/** (runner, task handler) +packages/vitnode/src/api/workflows/queue-task.ts +packages/vitnode/src/api/modules/queue/** (further queue hardening) +packages/vitnode/src/lib/api/resolve-stale-queue-lease.ts +packages/vitnode/src/lib/api/*queue*.test.ts +``` + +**Scope** + +- Implement the runner behind `workflow-step.task.ts`: resolve the definition + by exact version, claim the step, build the step context, run it, record the + outcome, chain the next step or finish the execution. +- Failure handling: `nextWorkflowAttemptAt` decides retry vs fail; a failed + step fails the execution and marks the remaining steps `skipped`. +- Cancellation check **between** steps only. +- `WORKFLOW_DEFINITION_NOT_FOUND` fails the execution and preserves every row. +- Wave 0 already added stale-lease recovery to `processQueueTasks`; extend it + rather than replacing the approach, and keep it generic queue behaviour. + +**Must not touch:** workflow definition semantics (`define.ts`, `registry.ts`), +the database tables, the store implementation. The runner talks to persistence +only through the `WorkflowStore` interface. + +**Depends on:** nothing to start (code against `WorkflowStore`); needs Agent B +merged to run end to end. + +--- + +## Agent D - Trigger contracts + +**Owns** + +```text +packages/vitnode/src/api/workflows/triggers.ts +packages/vitnode/src/api/workflows/trigger-adapters.ts (new) +packages/vitnode/src/api/workflows/triggers.test.ts +``` + +**Scope** + +- Turn a `WorkflowEventTriggerDefinition` into an ordinary + `BuildEventListenerReturn`, and a `WorkflowCronTriggerDefinition` into an + ordinary `BuildCronReturn`. +- Wire the dedupe contract: `event:{eventId}` and `cron:{name}:{tick}`. +- Decide how triggers are registered on a module (`workflowTriggers: []`, or + folded into `events`/`cronJobs` at build time) **and agree it with the lead**, + because it touches `buildModule`. + +**Must not touch:** workflow runtime semantics, the runner, the store, the +tables. + +**Depends on nothing.** Adapters call `c.get("workflow").start(...)`; whether +the store behind it is implemented does not affect the adapter's shape. + +--- + +## Shared files - LEAD ONLY + +No agent edits these without the lead. They are the coupling points: two agents +editing one of them in parallel produces a merge that compiles and is wrong. + +```text +packages/vitnode/src/api/lib/module.ts buildModule surface +packages/vitnode/src/api/lib/plugin.ts buildApiPlugin collection +packages/vitnode/src/api/middlewares/global.middleware.ts + c.get("core"), c.get("workflow") +packages/vitnode/src/api/plugin.ts core module registration +packages/vitnode/src/database/relations.ts coreSchema barrel + relations +packages/vitnode/src/api/workflows/types.ts every cross-agent contract +packages/vitnode/src/api/workflows/const.ts frozen vocabulary +packages/vitnode/src/api/workflows/state-machine.ts transition tables +packages/vitnode/src/api/models/workflow.ts c.get("workflow") surface +apps/*/migrations/meta/** drizzle journal + snapshots +``` + +Wave 0 already made every change these files need for Wave 1 to proceed: +`workflows` is collected, validated, exposed on `c.get("core")`, the model is +registered, the tables are in `coreSchema`, and the `workflow-step` task is +registered. If an agent finds one of them genuinely insufficient, that is a +lead ticket, not a local edit. + +--- + +## Parallel-work verification + +| Claim | Why it holds | +| --- | --- | +| A can implement the SDK without touching DB internals | `define.ts`/`registry.ts`/`plan.ts` are pure; `plan.ts` produces a `WorkflowStartPlan` and never inserts. | +| B can implement persistence without redesigning the SDK | `WorkflowStore` in `types.ts` is the whole surface; `WorkflowStartPlan` is handed in finished. | +| C can implement the runner without changing definition semantics | The runner reads a definition through `registry.ts` and persists through `WorkflowStore`; it authors neither. | +| D can implement trigger adapters without changing runtime semantics | Adapters build an envelope/tick into an input and call `start()`; they never touch execution state. | + +The one negotiation left is D's registration surface on `buildModule`, which is +flagged above as a lead decision precisely because it is the only remaining +overlap. diff --git a/docs/architecture/0001-workflow-engine.md b/docs/architecture/0001-workflow-engine.md new file mode 100644 index 000000000..d6e97faf3 --- /dev/null +++ b/docs/architecture/0001-workflow-engine.md @@ -0,0 +1,625 @@ +# ADR 0001 - Workflow Engine + +**Status:** contracts frozen (Wave 0). Runner, persistence, triggers and AdminCP are not implemented. +**Applies to:** `@vitnode/core` + +Wave 1 agents treat this document as the contract. Anything it fixes may not be +redesigned without changing this document first. + +--- + +## 1. Goals + +- Durable, restartable orchestration of multi-step business operations. +- Sequential, deterministic execution: step 1, then step 2, then step 3. +- Explicit versioning; an in-flight execution never migrates to newer code. +- At-least-once execution with a first-class idempotency contract. +- Per-step business retry, separate from queue delivery retry. +- A transaction boundary that lets a business row and its workflow commit together. +- Compensation and cancellation contracts frozen now, implemented later. + +## 2. Non-goals (explicitly out of the first engine) + +Parallel branches, DAG execution, loops, visual builders, `wait-for-signal`, +human approval steps, child workflows, dynamic graphs, webhooks, and +exactly-once semantics. Each is a future extension; none may be assumed by +Wave 1 code. + +## 3. Responsibility boundaries + +```text +Events = something happened +Queue = execute something later +CRON = initiate something on a schedule +Workflow = orchestrate durable multi-step business operations +``` + +The Workflow Engine is an **orchestration layer over existing VitNode systems**. +It does not ship: + +- a second queue - steps are delivered through `core_queue`; +- a second scheduler - cron triggers register ordinary `buildCron` jobs; +- a second event bus - event triggers register ordinary `buildEventListener`s; +- separate worker infrastructure - the existing `process-queue` cron drains it. + +Queue and Events stay fully usable without the Workflow Engine. Nothing in core +starts a workflow. + +## 4. Execution model + +`durable | sequential | queue-backed | versioned | idempotent | retryable | restart-safe` + +```text +c.get("workflow").start(workflow, input, { tx }) + │ validate input against the definition's zod schema + │ resolve the owning plugin from the registry (by object identity) + │ write core_workflow_executions (status: pending) + │ write core_workflow_step_executions (one row per step, pending, in order) + │ dispatch one @vitnode/core:workflow-step task { executionId, stepId } + └─ return { executionId, status: "pending", deduplicated } + +... later, in the queue worker's cron request ... + +workflow-step task + │ load execution + steps + │ resolve pluginId + workflowId + workflowVersion -> registered definition + │ claim the step (pending -> running, attempts += 1) + │ run step.run(ctx) + │ record output, complete the step + │ cancellation requested? -> skip the rest, execution -> cancelled + │ otherwise dispatch the next step's workflow-step task + └─ no next step -> execution -> completed +``` + +**`start()` never runs a step.** Step 1 does not execute inside the caller's +HTTP request: a slow inventory call must not become a slow checkout response, +and a crash a millisecond after the commit must still leave the work queued. + +The whole step plan is written at start rather than one row at a time. That is +what makes the engine restart-safe - after a crash the runner reads state +instead of re-deriving it - and what lets an operator see where a stuck +execution stopped. + +## 5. State machines + +### Execution status + +```text +pending -> running | cancelled +running -> completed | failed | cancelled +completed -> (terminal) +failed -> (terminal) +cancelled -> (terminal) +``` + +`failed` is terminal. Operator-initiated resume (`failed -> running`) is a +deliberate future extension and must be added to +`WORKFLOW_EXECUTION_TRANSITIONS` first. + +### Step status + +```text +pending -> running | skipped +running -> completed | failed | pending +completed -> (terminal) +failed -> (terminal) +skipped -> (terminal) +``` + +`running -> pending` is a scheduled retry (`nextAttemptAt` says when). +`pending -> skipped` is a step the runner never started - cancellation, or an +earlier step failed. + +### Compensation status (separate property, on both tables) + +```text +none -> pending +pending -> running +running -> completed | failed | pending +``` + +Compensation is tracked **beside** status, never folded into it. A combined +vocabulary (`failed_compensating`, `failed_compensated`, +`failed_compensation_failed`) makes every "did this succeed" query enumerate +compensation states it does not care about, and doubles each time a new one +appears. + +Code: `src/api/workflows/state-machine.ts`. + +## 6. Versioning + +Every definition declares `version: number` (positive integer, bumped by hand). +Definition identity is: + +```text +pluginId + workflowId + version +``` + +- Duplicates are rejected at plugin build (`buildApiPlugin`) and again across + all plugins at boot (`globalMiddleware`). +- `place-order@1` and `place-order@2` may be registered side by side. That is + the supported deployment, not an edge case. +- The runner resolves **only** the exact version on the execution row. There is + no `workflowId -> latest` lookup anywhere in the codebase, and + `resolveWorkflowDefinition` deliberately has no "latest" variant. + +Bump the version whenever the step list changes meaning: a step added, removed, +renamed or reordered. + +### Missing version behaviour + +```text +DB execution: @vitnode/shop / place-order / v1 +deployed code: only v2 exists +``` + +The runner must **not** run v2. It fails the execution with +`WORKFLOW_DEFINITION_NOT_FOUND` (`WorkflowDefinitionNotFoundError`), writes the +code into `lastError`, and leaves every row in place for an operator. + +**Deployment guidance:** keep old workflow versions registered until no active +execution requires them. `core_workflow_executions_definition_idx` covers the +query that answers "is anything still running on v1". + +## 7. Public SDK contract + +```ts +export const placeOrderWorkflow = defineWorkflow({ + id: "place-order", + version: 1, + + input: z.object({ orderId: z.number().int().positive() }), + + steps: ({ step }) => [ + step({ + id: "reserve-inventory", + retry: { maxAttempts: 3, strategy: "exponential" }, + run: async ({ input, idempotencyKey }) => ({ reservationId: 1 }), + compensate: async ({ output, idempotencyKey }) => { + // future wave + }, + }), + + step({ + id: "authorize-payment", + output: z.object({ chargeId: z.string() }), + run: async ({ outputs, idempotencyKey }) => { + const { reservationId } = outputs.parse( + "reserve-inventory", + z.object({ reservationId: z.number() }), + ); + + return { chargeId: `ch_${reservationId}` }; + }, + }), + ], +}); +``` + +Frozen concepts: `workflow.id`, `workflow.version`, `workflow.input`, +`workflow.steps`, `step.id`, `step.run`, `step.output`, `step.retry`, +`step.compensate`. + +### Definition-time validation (all throw `WorkflowError`) + +| Rule | Code | +| --- | --- | +| Non-empty id matching `^[a-z0-9][a-z0-9._-]*$`, <= 100 chars | `WORKFLOW_INVALID_ID` | +| `version` is a positive integer | `WORKFLOW_INVALID_VERSION` | +| At least one step | `WORKFLOW_EMPTY` | +| Step ids unique within a workflow | `WORKFLOW_DUPLICATE_STEP` | +| Retry policy is coherent | `WORKFLOW_INVALID_RETRY_POLICY` | + +Declaration order is frozen into each step's `position` at definition time. + +### Step context + +```ts +interface WorkflowStepContext { + readonly actor: WorkflowActor; // metadata only, never authorization + readonly attempt: number; // 1 on the first run + readonly c: Context; // background request: no user, no admin + readonly execution: WorkflowExecutionRef; + readonly idempotencyKey: string; // workflow:{executionId}:{stepId} + readonly input: TInput; + readonly outputs: WorkflowStepOutputs; // get / has / parse(stepId, schema) + readonly step: { id: string; position: number }; + readonly trigger: WorkflowTriggerRef; +} +``` + +There is no compile-time typing of *other* steps' outputs. The array form +`steps: ({ step }) => [...]` cannot thread a growing tuple through, and outputs +come back out of JSONB after a restart anyway - so `outputs.get()` is `unknown` +and `outputs.parse(stepId, schema)` is the supported way across that boundary. +This is the only place the engine uses `unknown`, and it is a genuine runtime +boundary. + +### Ownership and registration + +```ts +buildModule({ + pluginId: "@vitnode/shop", + name: "orders", + routes: [], + workflows: [placeOrderWorkflow], +}); +``` + +Collected **recursively** through the module tree (like `contentTypes`, +`contentModels` and `searchIndexers`, unlike `cronJobs`/`events`/`queueTasks`), +keeping the *owning* module's name. Exposed to background work as +`c.get("core").workflows`, the same way `contentModels` is - the runner has no +plugin context and needs a lookup from the row back to the code. + +`start()` resolves the owning plugin by **object identity**, not by +`c.get("plugin")`: guessing would be wrong the moment one plugin starts +another's workflow, and would write an execution row nothing can resolve. An +unregistered definition is a hard error. + +## 8. Storage + +Definitions are **source code**. Postgres stores identity and runtime state +only. JSONB is used for genuinely schema-dynamic values (workflow input, step +output) and nothing else. + +### `core_workflow_executions` + +```text +id serial pk +pluginId varchar(100) not null +module varchar(100) not null +workflowId varchar(100) not null +workflowVersion integer not null +status enum not null default 'pending' +compensationStatus enum not null default 'none' +triggerType enum not null default 'manual' +triggerName varchar(255) null +triggerId varchar(255) null +actorType enum not null default 'system' +actorId integer null (no FK - see below) +input jsonb not null default {} +output jsonb null +idempotencyKey varchar(255) null +lastError text null +cancellationRequestedAt timestamp null +createdAt timestamp not null default now() +startedAt timestamp null +completedAt timestamp null +cancelledAt timestamp null +updatedAt timestamp not null +``` + +Indexes: + +- `core_workflow_executions_idempotency_unique` - **partial** unique on + `(pluginId, workflowId, workflowVersion, idempotencyKey)` + `WHERE "idempotencyKey" is not null`. An execution with no key is not + de-duplicated at all, and any number of them may exist. +- `core_workflow_executions_status_idx` on `(status, createdAt)` - the AdminCP + list and "what is stuck". +- `core_workflow_executions_definition_idx` on + `(pluginId, workflowId, workflowVersion, status)` - "is anything still + running on v1", the question a deploy must answer before removing an old + version. + +`actorId` carries no foreign key: the actor is a fact about the past and must +stay readable after the account is gone. + +### `core_workflow_step_executions` + +```text +id serial pk +executionId integer not null -> core_workflow_executions.id (cascade) +stepId varchar(100) not null +position integer not null -- 0-based, frozen at plan time +status enum not null default 'pending' +attempts integer not null default 0 +maxAttempts integer not null default 3 +output jsonb null +lastError text null +nextAttemptAt timestamp null +compensationStatus enum not null default 'none' +compensationAttempts integer not null default 0 +compensationError text null +startedAt timestamp null +completedAt timestamp null +updatedAt timestamp not null +``` + +Invariants and indexes: + +- `UNIQUE(executionId, stepId)` - the invariant the runner rests on. "Has this + step already run" becomes a key lookup, and a duplicated queue delivery + cannot create a second attempt row. +- `(executionId, position)` - the runner's ordered read. +- `(status, nextAttemptAt)` - retries that have come due. + +The three `compensation*` columns extend the originally sketched field list. +They are required, not decorative: compensation is resumable per step, so a +crash halfway through a rollback has to continue where it stopped without +undoing anything twice. `nextAttemptAt` is shared between run-retry and +compensation-retry because a step is never running and compensating at once; +the error columns are separate so a rollback failure does not erase the +original one. + +Code: `src/database/workflows.ts`, registered in `src/database/relations.ts`. + +## 9. Transaction boundary + +`WorkflowModel.start()` accepts `{ tx }` with exactly the semantics of +`QueueModel.dispatch({ tx })`. + +```text +transaction +│ +├── business row +├── workflow execution +├── workflow step rows +└── workflow queue task +│ +COMMIT +``` + +Everything or nothing. Without `tx` the execution row can commit while the row +it refers to rolls back, and the runner wakes up to orchestrate something that +does not exist. + +## 10. Queue integration + +One generic core task: + +```text +@vitnode/core:workflow-step payload: { executionId, stepId } +``` + +Never one task per step. The queue resolves handlers by +`` `${pluginId}:${name}` ``, so `shop-authorize-payment`-style names would put +every plugin's business vocabulary into core's namespace and force registration +to happen before the step list is known. Here the payload names the execution +and the step, and the runner reads plugin, workflow, version, input and prior +outputs from the row - which is also why a task that sat in the queue across a +deploy still resolves against the version its execution started on. + +Code: `src/api/workflows/queue-task.ts`, +`src/api/modules/workflows/tasks/workflow-step.task.ts`. + +### Queue lease recovery (implemented in Wave 0) + +The pre-existing worker had a durability gap: `processQueueTasks` claims rows by +flipping them `pending -> processing` and stamping `reservedAt`, then runs the +handlers. If the process dies in between - deploy, OOM kill, container +reschedule - nothing writes the finishing update and the row is invisible to +every later tick, because the claim query only selects `pending`. That task is +lost permanently. + +A workflow step stuck in `processing` forever is an execution that never +advances *and* never fails - the one state a durable engine must not have. So +the invariant is established now, as **generic queue infrastructure**: + +```text +pending -> processing, reservedAt = now +worker dies +processing AND reservedAt < now - QUEUE_LEASE_TIMEOUT_MS + -> attempts < maxAttempts ? pending (available immediately) + -> otherwise : failed (lastError says the lease expired) +``` + +- Lease timeout: **15 minutes**. Long enough that a legitimately slow batch of + 25 tasks is never reclaimed while still running; short enough that a crashed + worker's tasks resume within a deploy cycle. +- Recovery counts against the task's own `maxAttempts` - the attempt was + already incremented at claim time - so a handler that reliably kills its + process fails eventually instead of cycling forever. +- Runs before the claim, so a recovered task can be picked up in the same tick. +- No heartbeats. Nothing in the current architecture justifies them; the lease + plus the attempt counter is sufficient. + +Code: `src/lib/api/resolve-stale-queue-lease.ts`, +`src/api/modules/queue/helpers/process-queue-tasks.ts`. + +## 11. Retry ownership + +```text +Queue retry = runner/infrastructure delivery retry (core_queue.maxAttempts) +Workflow step retry = business step retry (step.retry, core_workflow_step_executions) +Compensation retry = its own budget again, independent of the step's +``` + +```ts +retry: { + maxAttempts: 5, // total runs of the step body, not retries after the first + strategy: "exponential" | "fixed", + initialDelayMs: 1000, + maxDelayMs: 60_000, +} +``` + +Defaults: `{ maxAttempts: 3, strategy: "exponential", initialDelayMs: 1000, maxDelayMs: 60_000 }`. +Validated at definition time (`maxAttempts` a positive integer <= 25, +non-negative delays, `maxDelayMs >= initialDelayMs`), so a nonsense backoff +fails at boot rather than the first time a step happens to throw. + +`workflowRetryDelayMs` and `nextWorkflowAttemptAt` are the only places the +backoff curve is written down. No runner may compute `2 ** n` itself. +`nextWorkflowAttemptAt` returning `null` is what fails a step. + +Code: `src/api/workflows/retry.ts`. + +## 12. At-least-once semantics and idempotency + +The engine is **at-least-once**. It does not claim exactly-once, and no +documentation may say otherwise. + +```text +external side effect succeeds + ↓ +worker dies before the DB step completion is recorded + ↓ +the step runs again +``` + +Idempotency is therefore a **developer contract**, not a nicety. Every step +context carries a deterministic key: + +```text +step workflow:{executionId}:{stepId} +compensation workflow:{executionId}:{stepId}:compensate +``` + +```ts +await stripe.paymentIntents.create(payload, { + idempotencyKey: ctx.idempotencyKey, +}); +``` + +Compensation gets its own key because sharing one would make the provider +answer "refund this charge" with the cached response of "create this charge". + +Execution-level de-duplication is scoped by +`pluginId + workflowId + workflowVersion + idempotencyKey`, enforced by the +partial unique index. Two workflows may react to the same event, and +`place-order@2` is a different subscriber from `place-order@1`. + +Code: `src/api/workflows/idempotency.ts`. + +## 13. Actor handling + +`actorType` is `admin | user | system`; `actorId` is `number | null`. Defaults +from `c.get("admin")`, then `c.get("user")`, then `system`. + +**Queued steps never impersonate the original actor.** The runner executes as +system infrastructure: its request has no session, and `c.get("admin")` / +`c.get("user")` stay `null`. Reconstructing a fake request auth would make +every permission check, log line and model sharing that context lie about who +is present. The original actor is metadata, reachable as `ctx.actor`. + +## 14. Triggers + +Definition and trigger are separate. `defineWorkflow({ trigger })` would bind +one workflow to one way of starting it; the same `place-order` has to be +reachable from direct code, an event, a cron tick, the AdminCP, the API and a +test. + +```ts +buildWorkflowEventTrigger({ + name: "start-place-order", + workflow: placeOrderWorkflow, + event: "order.created", + input: payload => ({ orderId: payload.orderId }), + when: payload => payload.total > 0, // optional +}); + +buildWorkflowCronTrigger({ + name: "nightly-reconciliation", + schedule: "0 3 * * *", + workflow: reconcileLedgerWorkflow, + input: tick => ({ day: tick.toISOString().slice(0, 10) }), +}); +``` + +Both are **adapters over existing infrastructure**: the event trigger becomes +an ordinary `buildEventListener`, the cron trigger an ordinary `buildCron` job +whose body does nothing but call `start()`. No new scheduler, no new bus. + +Dedupe contract: + +| Trigger | `idempotencyKey` | `triggerName` | `triggerId` | +| --- | --- | --- | --- | +| event | `event:{envelope.eventId}` | event name | `envelope.eventId` | +| cron | `cron:{name}:{tick to the minute}` | cron job name | tick key | +| manual | none unless the caller passes one | caller label | `null` | + +A broker delivering the same envelope twice must produce **one** execution. +Mapper functions (`input`, `when`) must be pure: they run once, at start, and +their result is what every attempt of every step sees. + +Code: `src/api/workflows/triggers.ts`. + +## 15. Compensation contract (not implemented) + +```ts +step({ + id: "reserve-inventory", + run: async () => ({ reservationId: 123 }), + compensate: async ({ output, idempotencyKey }) => { /* future */ }, +}); +``` + +Frozen rules: + +1. Only **completed** steps are compensated. +2. Compensation runs in **reverse completion order**. +3. Compensation retries **independently** of the step's retry budget. +4. Compensation uses **its own idempotency key**. +5. Compensation is **not a SQL rollback** - the step already committed, and + possibly charged a card. +6. Progress is tracked per step (`compensationStatus`, `compensationAttempts`, + `compensationError`) so a crash mid-rollback resumes where it stopped. + +## 16. Cancellation contract (not implemented) + +VitNode cannot interrupt arbitrary running JavaScript and does not claim to. + +```text +cancel(executionId) -> cancellationRequestedAt = now + +Step B running + -> Step B completes normally +runner checks the cancellation request between steps + -> Step C never starts, becomes "skipped" + -> execution -> cancelled, cancelledAt = now +``` + +`cancellationRequestedAt` and `cancelledAt` are separate columns precisely +because the request and the effect are different moments. + +## 17. AdminCP integration boundaries (not implemented) + +Reserved, so Wave 1 does not invent conflicting names: + +- Routes: `admin/advanced/workflows`, alongside the existing `cron` and `queue` + advanced modules. +- Staff permission module: `workflows`, with + `can_view`, `{ can_cancel, dependsOn: [can_view] }`, + `{ can_retry, dependsOn: [can_view] }`. +- Read-only listing first (`core_workflow_executions` + its steps), following + `admin/advanced/queue/routes/get.route.ts` and `withPagination`. +- Cancel and retry go through `WorkflowModel`, never through direct table + writes - the state machine lives in code, not in a route handler. + +Nothing above is registered in Wave 0: a permission with no route behind it is +noise in the staff catalogue. + +## 18. Files + +```text +packages/vitnode/src/api/workflows/ + const.ts frozen vocabulary; imports nothing (drizzle-kit loads it) + types.ts every contract type, including WorkflowStore + errors.ts WorkflowError + stable codes + define.ts defineWorkflow + step builder + definition-time validation + registry.ts identity, duplicate validation, exact-version resolution + retry.ts RetryPolicy, resolution, the one backoff curve + idempotency.ts key builders and scope + state-machine.ts transition tables and assertions + step-outputs.ts the unknown -> typed boundary + plan.ts planWorkflowStart (pure) + triggers.ts event and cron trigger builders + queue-task.ts workflow-step payload schema + store.ts WorkflowStore implementation (Wave 0: throws) + index.ts + +packages/vitnode/src/api/models/workflow.ts c.get("workflow") +packages/vitnode/src/database/workflows.ts the two tables +packages/vitnode/src/api/modules/workflows/ core's runtime module +packages/vitnode/src/lib/api/resolve-stale-queue-lease.ts +``` + +## 19. Future extensions + +Parallel branches and DAG execution (would need `dependsOn` on steps and a +different runner), loops, `wait-for-signal` and human approval steps (would +need an execution status such as `waiting`), child workflows, a visual builder, +operator resume of a `failed` execution (`failed -> running`), and per-step +timeouts. Each needs a new ADR or an amendment here; none may be assumed. diff --git a/docs/architecture/README.md b/docs/architecture/README.md new file mode 100644 index 000000000..15992a624 --- /dev/null +++ b/docs/architecture/README.md @@ -0,0 +1,10 @@ +# Architecture documents + +Internal, implementation-focused records of decisions that span several +subsystems. Not published to `apps/docs` - user-facing documentation lands +there once the feature exists. + +| Document | Status | +| --- | --- | +| [0001 - Workflow Engine](./0001-workflow-engine.md) | Contracts frozen (Wave 0). Implementation in progress. | +| [0001a - Workflow Engine, Wave 1 ownership map](./0001-workflow-engine-wave-1-ownership.md) | Active | diff --git a/packages/vitnode/src/api/lib/module.ts b/packages/vitnode/src/api/lib/module.ts index 00fcd8246..c9212b447 100644 --- a/packages/vitnode/src/api/lib/module.ts +++ b/packages/vitnode/src/api/lib/module.ts @@ -4,6 +4,7 @@ import type { AnyContentModel } from "@/content/server/model"; import type { AnyContentTypeDefinition } from "@/content/types"; import type { SearchIndexer } from "../models/search"; +import type { AnyWorkflowDefinition } from "../workflows/types"; import type { BuildCronReturn } from "./cron"; import type { BuildEventListenerReturn } from "./events"; import type { BuildQueueTaskReturn } from "./queue"; @@ -47,6 +48,19 @@ export interface BaseBuildModuleReturn< routes: Routes; searchIndexers?: SearchIndexer[]; webSockets: BuildWebSocketReturn[]; + /** + * Durable workflows this module owns. + * + * Collected recursively by `buildApiPlugin`, like `contentTypes` and + * `searchIndexers`, because a workflow usually reads best next to the domain + * module it orchestrates - which is often nested inside the plugin's `admin` + * tree. + * + * Registration is what makes a definition addressable: the runner resolves a + * queued step by `pluginId + workflowId + version` from the execution row, so + * a definition no module registers can never be picked up again. + */ + workflows?: AnyWorkflowDefinition[]; } export interface BuildModuleReturn< @@ -75,6 +89,7 @@ export function buildModule< queueTasks = [], searchIndexers, webSockets = [], + workflows, }: { contentModels?: AnyContentModel[]; contentTypes?: AnyContentTypeDefinition[]; @@ -87,6 +102,7 @@ export function buildModule< routes: Routes; searchIndexers?: SearchIndexer[]; webSockets?: BuildWebSocketReturn[]; + workflows?: AnyWorkflowDefinition[]; }): BuildModuleReturn { const hono = new OpenAPIHono(); @@ -115,5 +131,6 @@ export function buildModule< queueTasks, searchIndexers, webSockets, + workflows, }; } diff --git a/packages/vitnode/src/api/lib/plugin.test.ts b/packages/vitnode/src/api/lib/plugin.test.ts index e00296f1a..d840acd11 100644 --- a/packages/vitnode/src/api/lib/plugin.test.ts +++ b/packages/vitnode/src/api/lib/plugin.test.ts @@ -1,5 +1,6 @@ // @vitest-environment node import { describe, expect, it } from "vitest"; +import { z } from "zod"; import { testArticleContentType, @@ -9,6 +10,7 @@ import { import type { SearchIndexer } from "../models/search"; import { validateSearchIndexers } from "../models/search"; +import { defineWorkflow } from "../workflows/define"; import { buildModule } from "./module"; import { buildApiPlugin } from "./plugin"; @@ -217,3 +219,78 @@ describe("validateSearchIndexers", () => { ).toEqual(["blog_post", "test.article"]); }); }); + +describe("buildApiPlugin workflows", () => { + const workflow = (id: string, version: number) => + defineWorkflow({ + id, + input: z.object({ orderId: z.number() }), + steps: ({ step }) => [step({ id: "reserve", run: () => undefined })], + version, + }); + + it("collects workflows from nested modules, keeping the owning module", () => { + const plugin = buildApiPlugin({ + pluginId: "@vitnode/example", + modules: [ + buildModule({ + pluginId: "@vitnode/example", + name: "admin", + routes: [], + modules: [ + buildModule({ + pluginId: "@vitnode/example", + name: "orders", + routes: [], + workflows: [workflow("place-order", 1)], + }), + ], + }), + ], + }); + + expect( + plugin.workflows?.map(entry => [entry.definition.id, entry.module]), + ).toEqual([["place-order", "orders"]]); + }); + + it("registers two versions of one workflow side by side", () => { + const plugin = buildApiPlugin({ + pluginId: "@vitnode/example", + modules: [ + buildModule({ + pluginId: "@vitnode/example", + name: "orders", + routes: [], + workflows: [workflow("place-order", 1), workflow("place-order", 2)], + }), + ], + }); + + expect(plugin.workflows?.map(entry => entry.definition.version)).toEqual([ + 1, 2, + ]); + }); + + it("refuses the same workflow id and version twice inside one plugin", () => { + expect(() => + buildApiPlugin({ + pluginId: "@vitnode/example", + modules: [ + buildModule({ + pluginId: "@vitnode/example", + name: "orders", + routes: [], + workflows: [workflow("place-order", 1)], + }), + buildModule({ + pluginId: "@vitnode/example", + name: "checkout", + routes: [], + workflows: [workflow("place-order", 1)], + }), + ], + }), + ).toThrow(/already registered by module "orders" and again by "checkout"/); + }); +}); diff --git a/packages/vitnode/src/api/lib/plugin.ts b/packages/vitnode/src/api/lib/plugin.ts index 3571171d4..db41c61db 100644 --- a/packages/vitnode/src/api/lib/plugin.ts +++ b/packages/vitnode/src/api/lib/plugin.ts @@ -11,6 +11,7 @@ import { } from "@/content/registry"; import type { SearchIndexer } from "../models/search"; +import type { RegisteredWorkflowDefinition } from "../workflows/types"; import type { CronJobConfig } from "./cron"; import type { EventListenerConfig } from "./events"; import type { BaseBuildModuleReturn, BuildModuleReturn } from "./module"; @@ -19,6 +20,7 @@ import type { QueueTaskConfig } from "./queue"; import type { WebSocketConfig } from "./websocket"; import { validateSearchIndexers } from "../models/search"; +import { validateWorkflowDefinitions } from "../workflows/registry"; import { checkPluginId } from "./check-plugin-id"; import { applyModuleTags } from "./openapi-tags"; @@ -35,6 +37,7 @@ export interface BuildPluginApiReturn { queueTasks?: Omit[]; searchIndexers?: SearchIndexer[]; webSockets?: Omit[]; + workflows?: Omit[]; } export function buildApiPlugin

({ @@ -68,6 +71,7 @@ export function buildApiPlugin

({ const openApiTags: string[] = []; const queueTasks: BuildPluginApiReturn["queueTasks"] = []; const webSockets: BuildPluginApiReturn["webSockets"] = []; + const workflows: BuildPluginApiReturn["workflows"] = []; modules.forEach(handler => { openApiTags.push(...applyModuleTags(handler, pluginId)); @@ -76,6 +80,7 @@ export function buildApiPlugin

({ contentModels.push(...collectContentModels(handler)); contentTypes.push(...collectContentTypes(handler)); indexers.push(...collectSearchIndexers(handler)); + workflows.push(...collectWorkflows(handler)); handler.cronJobs?.forEach(cron => { cronJobs.push({ ...cron, module: handler.name }); @@ -100,6 +105,13 @@ export function buildApiPlugin

({ validateSearchIndexers(indexers.map(indexer => ({ ...indexer, pluginId }))); + // Collisions inside one plugin. The global middleware repeats the check + // across every installed plugin, which is the only place two plugins can be + // seen at once. + validateWorkflowDefinitions( + workflows.map(workflow => ({ ...workflow, pluginId })), + ); + return { pluginId, messages, @@ -112,6 +124,7 @@ export function buildApiPlugin

({ queueTasks, searchIndexers: indexers, webSockets, + workflows, // Every content type contributes can_view/can_create/can_edit/can_delete // unless the plugin declared that module itself. permissionStaff: withContentPermissions(permissionStaff, registered), @@ -149,3 +162,20 @@ function collectSearchIndexers(module: BaseBuildModuleReturn): SearchIndexer[] { ...(module.modules ?? []).flatMap(collectSearchIndexers), ]; } + +/** + * Same recursive walk, and it also keeps the *owning* module's name rather than + * the top-level one - a workflow nested three modules deep should say where it + * actually lives when the AdminCP lists it. + */ +function collectWorkflows( + module: BaseBuildModuleReturn, +): Omit[] { + return [ + ...(module.workflows ?? []).map(definition => ({ + definition, + module: module.name, + })), + ...(module.modules ?? []).flatMap(collectWorkflows), + ]; +} diff --git a/packages/vitnode/src/api/middlewares/global.middleware.ts b/packages/vitnode/src/api/middlewares/global.middleware.ts index 4d895fd47..64c7ba92b 100644 --- a/packages/vitnode/src/api/middlewares/global.middleware.ts +++ b/packages/vitnode/src/api/middlewares/global.middleware.ts @@ -25,6 +25,8 @@ import { import { SessionModel } from "@/api/models/session"; import { SessionAdminModel } from "@/api/models/session-admin"; import { StorageModel } from "@/api/models/storage"; +import { WorkflowModel } from "@/api/models/workflow"; +import { validateWorkflowDefinitions } from "@/api/workflows/registry"; import { validateContentTypes } from "@/content/registry"; import { ensureContentLocalizationLanguages } from "@/content/server/language-resolver"; import { warnAboutContentPreviewConfig } from "@/content/server/preview-config"; @@ -45,6 +47,7 @@ import type { SearchProviderApiPlugin, } from "../models/search"; import type { SSOApiPlugin } from "../models/sso"; +import type { RegisteredWorkflowDefinition } from "../workflows/types"; import { collectCronJobs } from "../lib/cron"; import { @@ -136,6 +139,15 @@ export interface EnvVariablesVitNode { searchIndexers: SearchIndexerConfig[]; storage?: VitNodeApiConfig["storage"]; webSockets: WebSocketConfig[]; + /** + * Every registered workflow definition, with the plugin that owns it. + * + * The runner picks a `workflow-step` task up in a cron request that has no + * plugin context at all, so the only way back from an execution row to the + * code it names is a lookup by `pluginId + workflowId + version` - which + * has to live somewhere that request can reach. + */ + workflows: RegisteredWorkflowDefinition[]; }; db: Pick["dbProvider"]; email: EmailModel; @@ -163,6 +175,7 @@ export interface EnvVariablesVitNode { newsletter: boolean; roleId: number; }; + workflow: WorkflowModel; } export const globalMiddleware = ({ @@ -236,6 +249,20 @@ export const globalMiddleware = ({ })), ); + // Validated once more across *all* plugins, for the same reason content types + // and search indexers are: `buildApiPlugin` can only see one plugin's modules, + // and `pluginId + workflowId + version` has to be unique installation-wide + // before any execution row can be resolved back to code. + const workflowsMetadata: RegisteredWorkflowDefinition[] = + validateWorkflowDefinitions( + plugins.flatMap(plugin => + (plugin.workflows ?? []).map(workflow => ({ + ...workflow, + pluginId: plugin.pluginId, + })), + ), + ); + const webSocketsMetadata: WebSocketConfig[] = plugins.flatMap(plugin => (plugin.webSockets ?? []).map(webSocket => ({ ...webSocket, @@ -363,6 +390,7 @@ export const globalMiddleware = ({ c.set("queue", new QueueModel(c)); c.set("search", new SearchModel(c)); c.set("storage", new StorageModel(c)); + c.set("workflow", new WorkflowModel(c)); c.set("realtime", realtime); // Resolved before `core` is set rather than per mint, so the integrations @@ -409,6 +437,7 @@ export const globalMiddleware = ({ contentModels: contentModelsMetadata, contentRevalidateOrigins: content?.revalidateOrigins, contentTypes: contentTypesMetadata, + workflows: workflowsMetadata, }); // Whether a localized content type's `defaultLocale` names a row in diff --git a/packages/vitnode/src/api/models/workflow.ts b/packages/vitnode/src/api/models/workflow.ts new file mode 100644 index 000000000..3c00ea165 --- /dev/null +++ b/packages/vitnode/src/api/models/workflow.ts @@ -0,0 +1,130 @@ +import type { Context } from "hono"; +import type { z } from "zod"; + +import type { + RegisteredWorkflowDefinition, + WorkflowActor, + WorkflowDefinition, + WorkflowDefinitionRef, + WorkflowStartOptions, + WorkflowStartResult, +} from "@/api/workflows/types"; + +import { planWorkflowStart } from "@/api/workflows/plan"; +import { + requireWorkflowDefinition, + resolveWorkflowRegistration, +} from "@/api/workflows/registry"; +import { workflowStore } from "@/api/workflows/store"; + +/** + * Durable multi-step business operations, exposed on the request context as + * `c.get("workflow")`. + * + * The engine orchestrates; it does not re-implement what VitNode already has. + * Events say something happened, the queue executes something later, cron + * initiates something on a schedule - a workflow strings those into one + * restartable, versioned, idempotent sequence and owns nothing else. Steps are + * delivered through the existing `core_queue`; there is no second queue, no + * second scheduler and no second event bus. + * + * Execution is **at-least-once**, never exactly-once. A step's side effect can + * succeed and the worker can die before the completion is recorded, in which + * case the step runs again - which is why every step context carries an + * `idempotencyKey`. + */ +export class WorkflowModel { + constructor(c: Context) { + this.c = c; + } + + protected readonly c: Context; + + private registered(): RegisteredWorkflowDefinition[] { + return this.c.get("core").workflows; + } + + /** + * The actor recorded on the execution: metadata only. + * + * A queued step never impersonates them. The runner's request has no session + * at all, and reconstructing one would make `c.get("admin")` lie for every + * other model sharing that context. + */ + private requestActor(): WorkflowActor { + const admin = this.c.get("admin"); + if (admin) return { id: admin.user.id, type: "admin" }; + + const user = this.c.get("user"); + if (user) return { id: user.id, type: "user" }; + + return { type: "system" }; + } + + /** + * Ask for an execution to stop. + * + * VitNode cannot interrupt JavaScript that is already running, and does not + * pretend to: this records the request, the runner notices it *between* + * steps, the step in flight finishes normally, and everything still pending + * is skipped. + */ + async cancel(executionId: number): Promise { + await workflowStore.requestCancellation(this.c, executionId); + } + + /** + * Exact-version lookup, for the runner and the AdminCP. + * + * Throws `WORKFLOW_DEFINITION_NOT_FOUND` when this deployment no longer + * registers that version. It never falls back to a newer one - an execution + * planned against v1 would then run v2's steps against v1's rows. + */ + definition(ref: WorkflowDefinitionRef): RegisteredWorkflowDefinition { + return requireWorkflowDefinition(this.registered(), ref); + } + + /** + * Create a durable execution and hand the first step to the queue. + * + * ```ts + * await c.get("workflow").start(placeOrderWorkflow, { orderId: 42 }); + * ``` + * + * Pass `tx` to join the caller's transaction, exactly like + * `QueueModel.dispatch({ tx })`, so the business row, the execution, its step + * rows and the queue task commit together or not at all: + * + * ```ts + * await db.transaction(async tx => { + * const order = await createOrder(tx); + * await c.get("workflow").start(placeOrderWorkflow, { orderId: order.id }, { tx }); + * }); + * ``` + * + * **No step runs here.** `start()` validates the input, writes the execution + * and its steps, queues one `workflow-step` task and returns - so a checkout + * request never waits on an inventory call, and a crash a millisecond after + * the commit still leaves the work queued. + */ + async start( + workflow: WorkflowDefinition, + input: z.input, + options: WorkflowStartOptions = {}, + ): Promise { + const entry = resolveWorkflowRegistration(this.registered(), workflow); + const plan = planWorkflowStart({ + entry, + input, + options: { + actor: options.actor ?? this.requestActor(), + idempotencyKey: options.idempotencyKey, + trigger: options.trigger, + }, + }); + + return await workflowStore.createExecution(this.c, plan, { + tx: options.tx, + }); + } +} diff --git a/packages/vitnode/src/api/modules/queue/helpers/process-queue-tasks.ts b/packages/vitnode/src/api/modules/queue/helpers/process-queue-tasks.ts index 2bd9fca7e..f0b2e8e21 100644 --- a/packages/vitnode/src/api/modules/queue/helpers/process-queue-tasks.ts +++ b/packages/vitnode/src/api/modules/queue/helpers/process-queue-tasks.ts @@ -6,6 +6,10 @@ import type { EnvVitNode } from "@/api/middlewares/global.middleware"; import { core_queue } from "@/database/queue"; import { resolveQueueTaskOutcome } from "@/lib/api/resolve-queue-task-outcome"; +import { + queueLeaseCutoff, + resolveStaleQueueLease, +} from "@/lib/api/resolve-stale-queue-lease"; const QUEUE_BATCH_SIZE = 25; const QUEUE_LOCK_KEY = "queue:process"; @@ -30,6 +34,8 @@ export const processQueueTasks = async ( const db = c.get("db"); const now = new Date(); + await recoverStaleQueueLeases(c, now); + const claimed = await db.transaction(async tx => { const rows = await tx .select({ id: core_queue.id }) @@ -120,3 +126,71 @@ export const processQueueTasks = async ( await c.get("cache").releaseLock(QUEUE_LOCK_KEY); } }; + +/** + * Give back tasks whose worker never came back. + * + * The claim in `processQueueTasks` is a lease: a row is flipped to `processing` + * before its handler runs, and the finishing update only happens if the process + * survives. A deploy, an OOM kill or a container reschedule in between leaves + * the row `processing` forever, and every later tick ignores it - the claim + * query only ever selects `pending`. + * + * Runs before the claim so a recovered task can be picked up in the same tick. + * Recovery counts against the task's own `maxAttempts` (the attempt was already + * incremented when it was claimed), so a handler that reliably takes its + * process down still fails eventually instead of cycling forever. + */ +const recoverStaleQueueLeases = async ( + c: Context, + now: Date, +): Promise => { + const db = c.get("db"); + const stale = await db + .select({ + attempts: core_queue.attempts, + id: core_queue.id, + maxAttempts: core_queue.maxAttempts, + name: core_queue.name, + pluginId: core_queue.pluginId, + }) + .from(core_queue) + .where( + and( + eq(core_queue.status, "processing"), + lt(core_queue.reservedAt, queueLeaseCutoff(now)), + ), + ) + .limit(QUEUE_BATCH_SIZE); + + if (stale.length === 0) return; + + for (const task of stale) { + const outcome = resolveStaleQueueLease({ + attempts: task.attempts, + maxAttempts: task.maxAttempts, + now, + }); + + await db + .update(core_queue) + .set({ + availableAt: outcome.availableAt ?? now, + completedAt: outcome.completedAt ?? null, + lastError: outcome.lastError, + reservedAt: null, + status: outcome.status, + }) + .where( + and(eq(core_queue.id, task.id), eq(core_queue.status, "processing")), + ); + } + + await c + .get("log") + .warn( + `Recovered ${stale.length} queue task(s) left in "processing" by a worker that stopped: ${stale + .map(task => `${task.pluginId}:${task.name}#${task.id}`) + .join(", ")}`, + ); +}; diff --git a/packages/vitnode/src/api/modules/workflows/tasks/workflow-step.task.ts b/packages/vitnode/src/api/modules/workflows/tasks/workflow-step.task.ts new file mode 100644 index 000000000..c0d9d3d0e --- /dev/null +++ b/packages/vitnode/src/api/modules/workflows/tasks/workflow-step.task.ts @@ -0,0 +1,41 @@ +import { buildQueueTask } from "@/api/lib/queue"; +import { + WORKFLOW_STEP_QUEUE_MAX_ATTEMPTS, + WORKFLOW_STEP_QUEUE_TASK, +} from "@/api/workflows/const"; +import { WORKFLOW_ERROR_CODES, WorkflowError } from "@/api/workflows/errors"; +import { workflowStepTaskPayloadSchema } from "@/api/workflows/queue-task"; + +/** + * The one queue task every workflow step in the installation is delivered + * through, resolved by the worker as `@vitnode/core:workflow-step`. + * + * One generic task rather than one per step. The queue resolves handlers by + * `` `${pluginId}:${name}` ``, so a task named after each business step would + * put `shop-authorize-payment` into core's namespace and force registration to + * happen before anybody knows the step list. Here the payload names the + * execution and the step, and the runner reads plugin, workflow, version, input + * and previous outputs from the execution row - which is also why a task that + * sat in the queue across a deploy still resolves against the version its + * execution started on. + * + * `maxAttempts` here is *delivery* retry: the worker could not reach the runner + * at all. A step body that fails is a different budget, owned by the step's own + * `retry` policy. + * + * The runner itself is Wave 1. This registration exists so its home, its name + * and its payload are already fixed. + */ +export const workflowStepQueueTask = buildQueueTask({ + description: "Run one step of a durable workflow execution", + maxAttempts: WORKFLOW_STEP_QUEUE_MAX_ATTEMPTS, + name: WORKFLOW_STEP_QUEUE_TASK, + handler: (_c, payload) => { + const parsed = workflowStepTaskPayloadSchema.parse(payload); + + throw new WorkflowError( + WORKFLOW_ERROR_CODES.NOT_IMPLEMENTED, + `no runner is registered for execution ${parsed.executionId}, step "${parsed.stepId}". The Workflow Engine's contracts are frozen but its runner is not implemented - see docs/architecture/0001-workflow-engine.md.`, + ); + }, +}); diff --git a/packages/vitnode/src/api/modules/workflows/workflows.module.ts b/packages/vitnode/src/api/modules/workflows/workflows.module.ts new file mode 100644 index 000000000..fce964443 --- /dev/null +++ b/packages/vitnode/src/api/modules/workflows/workflows.module.ts @@ -0,0 +1,18 @@ +import { buildModule } from "@/api/lib/module"; +import { CONFIG_PLUGIN } from "@/config"; + +import { workflowStepQueueTask } from "./tasks/workflow-step.task"; + +/** + * Core's home for the Workflow Engine's runtime. + * + * Deliberately thin: the engine orchestrates over the existing queue, cron and + * event systems rather than shipping its own, so the only thing core registers + * is the single generic step task the runner is delivered through. + */ +export const workflowsModule = buildModule({ + pluginId: CONFIG_PLUGIN.pluginId, + name: "workflows", + routes: [], + queueTasks: [workflowStepQueueTask], +}); diff --git a/packages/vitnode/src/api/plugin.ts b/packages/vitnode/src/api/plugin.ts index 45cd3deb4..6f3a32950 100644 --- a/packages/vitnode/src/api/plugin.ts +++ b/packages/vitnode/src/api/plugin.ts @@ -8,6 +8,7 @@ import { middlewareModule } from "./modules/middleware/middleware.module"; import { queueModule } from "./modules/queue/queue.module"; import { searchModule } from "./modules/search/search.module"; import { usersModule } from "./modules/users/users.module"; +import { workflowsModule } from "./modules/workflows/workflows.module"; export const newBuildPluginApiCore = buildApiPlugin({ pluginId: CONFIG_PLUGIN.pluginId, @@ -19,6 +20,7 @@ export const newBuildPluginApiCore = buildApiPlugin({ cronModule, queueModule, searchModule, + workflowsModule, ], permissionStaff: { moderator: { diff --git a/packages/vitnode/src/api/workflows/const.ts b/packages/vitnode/src/api/workflows/const.ts new file mode 100644 index 000000000..912526a11 --- /dev/null +++ b/packages/vitnode/src/api/workflows/const.ts @@ -0,0 +1,82 @@ +/** + * Frozen Workflow Engine vocabulary. + * + * Deliberately dependency-free: `src/database/workflows.ts` is loaded by + * `drizzle-kit` in plain Node, so the column enums have to come from a module + * that imports nothing. + */ + +export const WORKFLOW_EXECUTION_STATUSES = [ + "pending", + "running", + "completed", + "failed", + "cancelled", +] as const; + +export const WORKFLOW_STEP_STATUSES = [ + "pending", + "running", + "completed", + "failed", + "skipped", +] as const; + +/** + * Compensation is tracked *beside* the status, never folded into it. + * + * A combined vocabulary (`failed_compensating`, `failed_compensated`, + * `failed_compensation_failed`, ...) multiplies out: every question about + * "did this workflow succeed" then has to enumerate compensation states it + * does not care about, and every new compensation state doubles the list. + */ +export const WORKFLOW_COMPENSATION_STATUSES = [ + "none", + "pending", + "running", + "completed", + "failed", +] as const; + +export const WORKFLOW_TRIGGER_TYPES = ["manual", "event", "cron"] as const; + +export const WORKFLOW_ACTOR_TYPES = ["admin", "system", "user"] as const; + +export const WORKFLOW_RETRY_STRATEGIES = ["fixed", "exponential"] as const; + +/** + * The single generic queue task every workflow step is delivered through, + * resolved by the worker as `` `@vitnode/core:${WORKFLOW_STEP_QUEUE_TASK}` ``. + * + * One task, not one per step: the queue resolves handlers by + * `` `${pluginId}:${name}` ``, so a task per step would put every plugin's + * business vocabulary into core's queue namespace and make the runner + * un-writable - it would have to know each step's name at registration time + * rather than resolve it from the execution row. + */ +export const WORKFLOW_STEP_QUEUE_TASK = "workflow-step"; + +/** Queue delivery attempts for one `workflow-step` task. See `retry.ts`. */ +export const WORKFLOW_STEP_QUEUE_MAX_ATTEMPTS = 3; + +export const WORKFLOW_IDEMPOTENCY_PREFIX = "workflow"; +export const WORKFLOW_COMPENSATION_IDEMPOTENCY_SUFFIX = "compensate"; +export const WORKFLOW_EVENT_IDEMPOTENCY_PREFIX = "event"; +export const WORKFLOW_CRON_IDEMPOTENCY_PREFIX = "cron"; + +/** Matches the `varchar(100)` columns the identifiers are stored in. */ +export const WORKFLOW_ID_MAX_LENGTH = 100; +export const WORKFLOW_STEP_ID_MAX_LENGTH = 100; +export const WORKFLOW_IDEMPOTENCY_KEY_MAX_LENGTH = 255; + +/** + * Identifier shape for workflow and step ids. + * + * Both end up in database columns, in log lines and - through + * `workflow:{executionId}:{stepId}` - inside idempotency keys handed to + * third-party APIs, so they are restricted to characters that survive all + * three unchanged. + */ +export const WORKFLOW_ID_PATTERN = /^[a-z0-9][a-z0-9._-]*$/; + +export const WORKFLOW_MAX_RETRY_ATTEMPTS = 25; diff --git a/packages/vitnode/src/api/workflows/define.test-d.ts b/packages/vitnode/src/api/workflows/define.test-d.ts new file mode 100644 index 000000000..e51b5cd97 --- /dev/null +++ b/packages/vitnode/src/api/workflows/define.test-d.ts @@ -0,0 +1,102 @@ +import { describe, expectTypeOf, it } from "vitest"; +import { z } from "zod"; + +import { defineWorkflow } from "./define"; + +const input = z.object({ + currency: z.string(), + orderId: z.number().int().positive(), +}); + +describe("workflow input typing", () => { + it("gives every step the parsed input type", () => { + defineWorkflow({ + id: "place-order", + input, + steps: ({ step }) => [ + step({ + id: "reserve-inventory", + run: ctx => { + expectTypeOf(ctx.input).toEqualTypeOf<{ + currency: string; + orderId: number; + }>(); + expectTypeOf(ctx.idempotencyKey).toEqualTypeOf(); + expectTypeOf(ctx.attempt).toEqualTypeOf(); + + return undefined; + }, + }), + ], + version: 1, + }); + }); + + it("keeps the schema on the definition, so `start()` can type its argument", () => { + const workflow = defineWorkflow({ + id: "place-order", + input, + steps: ({ step }) => [step({ id: "reserve", run: () => undefined })], + version: 1, + }); + + expectTypeOf(workflow.input).toEqualTypeOf(input); + expectTypeOf>().toEqualTypeOf<{ + currency: string; + orderId: number; + }>(); + }); +}); + +describe("step output typing", () => { + it("types `compensate` from what `run` returns", () => { + defineWorkflow({ + id: "place-order", + input, + steps: ({ step }) => [ + step({ + id: "reserve-inventory", + compensate: ctx => { + expectTypeOf(ctx.output).toEqualTypeOf<{ reservationId: number }>(); + expectTypeOf(ctx.idempotencyKey).toEqualTypeOf(); + }, + run: () => ({ reservationId: 1 }), + }), + ], + version: 1, + }); + }); + + it("types `compensate` through an async `run`", () => { + defineWorkflow({ + id: "place-order", + input, + steps: ({ step }) => [ + step({ + id: "authorize-payment", + compensate: ctx => { + expectTypeOf(ctx.output).toEqualTypeOf<{ chargeId: string }>(); + }, + run: async () => await Promise.resolve({ chargeId: "ch_1" }), + }), + ], + version: 1, + }); + }); + + it("checks the declared `output` schema against what `run` returns", () => { + defineWorkflow({ + id: "place-order", + input, + steps: ({ step }) => [ + step({ + id: "reserve-inventory", + output: z.object({ reservationId: z.number() }), + // @ts-expect-error `run` must return what `output` describes. + run: () => ({ reservationId: "not-a-number" }), + }), + ], + version: 1, + }); + }); +}); diff --git a/packages/vitnode/src/api/workflows/define.test.ts b/packages/vitnode/src/api/workflows/define.test.ts new file mode 100644 index 000000000..7f9ff96bd --- /dev/null +++ b/packages/vitnode/src/api/workflows/define.test.ts @@ -0,0 +1,182 @@ +// @vitest-environment node +import { describe, expect, it } from "vitest"; +import { z } from "zod"; + +import { defineWorkflow } from "./define"; +import { WORKFLOW_ERROR_CODES, WorkflowError } from "./errors"; + +const input = z.object({ orderId: z.number().int().positive() }); + +const codeOf = (run: () => unknown): string => { + try { + run(); + } catch (error) { + return error instanceof WorkflowError ? error.code : "not-a-workflow-error"; + } + + return "did-not-throw"; +}; + +describe("defineWorkflow identity", () => { + it("keeps the declared id and version", () => { + const workflow = defineWorkflow({ + id: "place-order", + input, + steps: ({ step }) => [step({ id: "reserve", run: () => undefined })], + version: 1, + }); + + expect(workflow.id).toBe("place-order"); + expect(workflow.version).toBe(1); + }); + + it.each([ + ["", "empty"], + ["Place-Order", "uppercase"], + ["-place-order", "leading dash"], + ["place order", "whitespace"], + ["a".repeat(101), "over 100 characters"], + ])("rejects the %s workflow id (%s)", id => { + expect( + codeOf(() => + defineWorkflow({ + id, + input, + steps: ({ step }) => [step({ id: "reserve", run: () => undefined })], + version: 1, + }), + ), + ).toBe(WORKFLOW_ERROR_CODES.INVALID_ID); + }); + + it.each([0, -1, 1.5, Number.NaN])( + "rejects the non positive-integer version %s", + version => { + expect( + codeOf(() => + defineWorkflow({ + id: "place-order", + input, + steps: ({ step }) => [ + step({ id: "reserve", run: () => undefined }), + ], + version, + }), + ), + ).toBe(WORKFLOW_ERROR_CODES.INVALID_VERSION); + }, + ); +}); + +describe("defineWorkflow steps", () => { + it("freezes declaration order into deterministic positions", () => { + const workflow = defineWorkflow({ + id: "place-order", + input, + steps: ({ step }) => [ + step({ id: "reserve-inventory", run: () => undefined }), + step({ id: "authorize-payment", run: () => undefined }), + step({ id: "create-fulfillment", run: () => undefined }), + ], + version: 1, + }); + + expect(workflow.steps.map(step => [step.id, step.position])).toEqual([ + ["reserve-inventory", 0], + ["authorize-payment", 1], + ["create-fulfillment", 2], + ]); + }); + + it("rejects duplicate step ids", () => { + expect( + codeOf(() => + defineWorkflow({ + id: "place-order", + input, + steps: ({ step }) => [ + step({ id: "reserve", run: () => undefined }), + step({ id: "reserve", run: () => undefined }), + ], + version: 1, + }), + ), + ).toBe(WORKFLOW_ERROR_CODES.DUPLICATE_STEP); + }); + + it("rejects a workflow with no steps", () => { + expect( + codeOf(() => + defineWorkflow({ + id: "place-order", + input, + steps: () => [], + version: 1, + }), + ), + ).toBe(WORKFLOW_ERROR_CODES.EMPTY_WORKFLOW); + }); + + it("rejects an invalid step id", () => { + expect( + codeOf(() => + defineWorkflow({ + id: "place-order", + input, + steps: ({ step }) => [ + step({ id: "Reserve Inventory", run: () => undefined }), + ], + version: 1, + }), + ), + ).toBe(WORKFLOW_ERROR_CODES.INVALID_ID); + }); + + it("resolves every step's retry policy at definition time", () => { + const workflow = defineWorkflow({ + id: "place-order", + input, + steps: ({ step }) => [ + step({ id: "default-retry", run: () => undefined }), + step({ + id: "custom-retry", + retry: { maxAttempts: 5, strategy: "fixed" }, + run: () => undefined, + }), + ], + version: 1, + }); + + expect(workflow.steps[0].retry).toEqual({ + initialDelayMs: 1_000, + maxAttempts: 3, + maxDelayMs: 60_000, + strategy: "exponential", + }); + expect(workflow.steps[1].retry).toEqual({ + initialDelayMs: 1_000, + maxAttempts: 5, + maxDelayMs: 60_000, + strategy: "fixed", + }); + }); + + it("refuses an invalid retry policy at definition time, not at first failure", () => { + expect( + codeOf(() => + defineWorkflow({ + id: "place-order", + input, + steps: ({ step }) => [ + step({ + id: "reserve", + retry: { maxAttempts: 0 }, + run: () => undefined, + }), + ], + version: 1, + }), + ), + ).toBe(WORKFLOW_ERROR_CODES.INVALID_RETRY_POLICY); + }); +}); diff --git a/packages/vitnode/src/api/workflows/define.ts b/packages/vitnode/src/api/workflows/define.ts new file mode 100644 index 000000000..8eb90394d --- /dev/null +++ b/packages/vitnode/src/api/workflows/define.ts @@ -0,0 +1,173 @@ +import type { z } from "zod"; + +import type { + ResolvedWorkflowStep, + WorkflowDefinition, + WorkflowStepDefinition, + WorkflowStepsBuilder, +} from "./types"; + +import { + WORKFLOW_ID_MAX_LENGTH, + WORKFLOW_ID_PATTERN, + WORKFLOW_STEP_ID_MAX_LENGTH, +} from "./const"; +import { WORKFLOW_ERROR_CODES, WorkflowError } from "./errors"; +import { resolveWorkflowRetryPolicy } from "./retry"; + +const assertIdentifier = ({ + kind, + maxLength, + value, + workflowId, +}: { + kind: "step" | "workflow"; + maxLength: number; + value: string; + workflowId?: string; +}): void => { + if (typeof value !== "string" || value.length === 0) { + throw new WorkflowError( + WORKFLOW_ERROR_CODES.INVALID_ID, + `a ${kind} needs a non-empty \`id\`.`, + { workflowId }, + ); + } + + if (value.length > maxLength) { + throw new WorkflowError( + WORKFLOW_ERROR_CODES.INVALID_ID, + `${kind} id "${value}" is longer than ${maxLength} characters.`, + { workflowId }, + ); + } + + if (!WORKFLOW_ID_PATTERN.test(value)) { + throw new WorkflowError( + WORKFLOW_ERROR_CODES.INVALID_ID, + `${kind} id "${value}" must start with a lowercase letter or digit and contain only lowercase letters, digits, ".", "-" and "_". Ids end up in database columns, log lines and third-party idempotency keys, which is why they are restricted.`, + { workflowId }, + ); + } +}; + +/** + * Declares a durable, sequential workflow. + * + * The result is plain data - a zod schema, step objects and two identifiers - + * so the same definition is importable by the plugin that owns it, by the + * runner (which has no plugin context at all) and by tests, without any of + * them pulling in the others. + * + * ```ts + * export const placeOrderWorkflow = defineWorkflow({ + * id: "place-order", + * version: 1, + * input: z.object({ orderId: z.number().int().positive() }), + * steps: ({ step }) => [ + * step({ + * id: "reserve-inventory", + * retry: { maxAttempts: 3, strategy: "exponential" }, + * run: async ({ input, idempotencyKey }) => ({ reservationId: 1 }), + * }), + * step({ + * id: "authorize-payment", + * run: async ({ outputs }) => { + * const { reservationId } = outputs.parse( + * "reserve-inventory", + * z.object({ reservationId: z.number() }), + * ); + * + * return { chargeId: `ch_${reservationId}` }; + * }, + * }), + * ], + * }); + * ``` + */ +export const defineWorkflow = < + const TId extends string, + TInputSchema extends z.ZodType, +>({ + description, + id, + input, + steps, + version, +}: { + description?: string; + id: TId; + /** Runtime-validated by `start()`, and the source of every step's `input` type. */ + input: TInputSchema; + /** + * Declaration order *is* execution order, and it is frozen into each step's + * `position` here rather than re-derived later. Reordering steps changes what + * the workflow means, so it needs a new `version`. + */ + steps: ( + builder: WorkflowStepsBuilder>, + ) => WorkflowStepDefinition, unknown>[]; + /** A positive integer, bumped by hand. There is no "latest" anywhere. */ + version: number; +}): WorkflowDefinition => { + assertIdentifier({ + kind: "workflow", + maxLength: WORKFLOW_ID_MAX_LENGTH, + value: id, + }); + + if (!Number.isSafeInteger(version) || version < 1) { + throw new WorkflowError( + WORKFLOW_ERROR_CODES.INVALID_VERSION, + `\`version\` must be a positive integer, received ${String(version)}. Executions store the version they started with, so it cannot be inferred or defaulted.`, + { workflowId: id }, + ); + } + + const declared = steps({ + // The per-step output type is checked at this call site and erased + // afterwards, so one array can hold steps that return different things. + step: definition => + definition as WorkflowStepDefinition, unknown>, + }); + + if (declared.length === 0) { + throw new WorkflowError( + WORKFLOW_ERROR_CODES.EMPTY_WORKFLOW, + "a workflow needs at least one step. An execution with nothing to run would be created, queued and completed without ever doing anything, which is a bug worth catching at boot.", + { workflowId: id, version }, + ); + } + + const seen = new Set(); + const resolved: ResolvedWorkflowStep>[] = declared.map( + (definition, position) => { + assertIdentifier({ + kind: "step", + maxLength: WORKFLOW_STEP_ID_MAX_LENGTH, + value: definition.id, + workflowId: id, + }); + + if (seen.has(definition.id)) { + throw new WorkflowError( + WORKFLOW_ERROR_CODES.DUPLICATE_STEP, + `two steps share the id "${definition.id}". Step ids address one row per execution (\`UNIQUE(executionId, stepId)\`) and form the step's idempotency key, so they have to be unique inside a workflow.`, + { workflowId: id, version }, + ); + } + + seen.add(definition.id); + + return { + ...definition, + position, + retry: resolveWorkflowRetryPolicy(definition.retry, { + stepId: definition.id, + }), + }; + }, + ); + + return { description, id, input, steps: resolved, version }; +}; diff --git a/packages/vitnode/src/api/workflows/errors.ts b/packages/vitnode/src/api/workflows/errors.ts new file mode 100644 index 000000000..7deed254e --- /dev/null +++ b/packages/vitnode/src/api/workflows/errors.ts @@ -0,0 +1,98 @@ +/** + * Structured Workflow Engine failures. + * + * Every one carries a stable `code`. The runner writes that code into + * `core_workflow_executions.lastError`, so an operator reading a stuck + * execution sees the same token the tests assert on rather than a sentence + * that changed between releases. + */ +export const WORKFLOW_ERROR_CODES = { + DEFINITION_NOT_FOUND: "WORKFLOW_DEFINITION_NOT_FOUND", + DUPLICATE_DEFINITION: "WORKFLOW_DUPLICATE_DEFINITION", + DUPLICATE_STEP: "WORKFLOW_DUPLICATE_STEP", + EMPTY_WORKFLOW: "WORKFLOW_EMPTY", + INVALID_ID: "WORKFLOW_INVALID_ID", + INVALID_INPUT: "WORKFLOW_INVALID_INPUT", + INVALID_RETRY_POLICY: "WORKFLOW_INVALID_RETRY_POLICY", + INVALID_TRANSITION: "WORKFLOW_INVALID_TRANSITION", + INVALID_TRIGGER: "WORKFLOW_INVALID_TRIGGER", + INVALID_VERSION: "WORKFLOW_INVALID_VERSION", + NOT_IMPLEMENTED: "WORKFLOW_NOT_IMPLEMENTED", + STEP_NOT_FOUND: "WORKFLOW_STEP_NOT_FOUND", + STEP_OUTPUT_INVALID: "WORKFLOW_STEP_OUTPUT_INVALID", +} as const; + +export type WorkflowErrorCode = + (typeof WORKFLOW_ERROR_CODES)[keyof typeof WORKFLOW_ERROR_CODES]; + +export interface WorkflowErrorOptions { + cause?: unknown; + /** `pluginId + workflowId + version`, when the failure is about one. */ + pluginId?: string; + version?: number; + workflowId?: string; +} + +const describe = ({ pluginId, version, workflowId }: WorkflowErrorOptions) => { + if (!workflowId) return ""; + const owner = pluginId ? `${pluginId} -> ` : ""; + const at = version === undefined ? "" : `@${version}`; + + return `${owner}${workflowId}${at}: `; +}; + +/** + * Thrown while a definition is built or registered (import/boot time), or by + * the runtime when an execution cannot be resolved or moved. + */ +export class WorkflowError extends Error { + constructor( + code: WorkflowErrorCode, + message: string, + options: WorkflowErrorOptions = {}, + ) { + super(`[Workflow] ${describe(options)}${message}`, { + cause: options.cause, + }); + + this.name = "WorkflowError"; + this.code = code; + this.pluginId = options.pluginId; + this.version = options.version; + this.workflowId = options.workflowId; + } + + readonly code: WorkflowErrorCode; + readonly pluginId: string | undefined; + readonly version: number | undefined; + readonly workflowId: string | undefined; +} + +/** + * The execution names a definition this deployment does not have. + * + * Its own class because it is the one workflow failure an operator is expected + * to *act* on rather than debug: the code that ran this execution was removed + * or renamed, and the fix is to put that version back, not to change data. The + * runner never falls back to another version - see + * {@link resolveWorkflowDefinition}. + */ +export class WorkflowDefinitionNotFoundError extends WorkflowError { + constructor({ + pluginId, + version, + workflowId, + }: { + pluginId: string; + version: number; + workflowId: string; + }) { + super( + WORKFLOW_ERROR_CODES.DEFINITION_NOT_FOUND, + "No workflow definition is registered for this exact version. Keep old workflow versions registered until no execution needs them - the runner never upgrades an in-flight execution to a newer version.", + { pluginId, version, workflowId }, + ); + + this.name = "WorkflowDefinitionNotFoundError"; + } +} diff --git a/packages/vitnode/src/api/workflows/idempotency.test.ts b/packages/vitnode/src/api/workflows/idempotency.test.ts new file mode 100644 index 000000000..5e12c0dfd --- /dev/null +++ b/packages/vitnode/src/api/workflows/idempotency.test.ts @@ -0,0 +1,99 @@ +// @vitest-environment node +import { describe, expect, it } from "vitest"; + +import { + workflowCronIdempotencyKey, + workflowEventIdempotencyKey, + workflowIdempotencyScopeKey, + workflowStepCompensationIdempotencyKey, + workflowStepIdempotencyKey, +} from "./idempotency"; + +describe("step idempotency keys", () => { + it("is deterministic for one execution and step", () => { + const args = { executionId: 182, stepId: "reserve-inventory" }; + + expect(workflowStepIdempotencyKey(args)).toBe( + "workflow:182:reserve-inventory", + ); + expect(workflowStepIdempotencyKey(args)).toBe( + workflowStepIdempotencyKey(args), + ); + }); + + it("gives compensation its own key so it cannot collide with the step's", () => { + const args = { executionId: 182, stepId: "authorize-payment" }; + + expect(workflowStepCompensationIdempotencyKey(args)).toBe( + "workflow:182:authorize-payment:compensate", + ); + expect(workflowStepCompensationIdempotencyKey(args)).not.toBe( + workflowStepIdempotencyKey(args), + ); + }); + + it("separates two executions of the same workflow", () => { + expect( + workflowStepIdempotencyKey({ executionId: 1, stepId: "reserve" }), + ).not.toBe( + workflowStepIdempotencyKey({ executionId: 2, stepId: "reserve" }), + ); + }); +}); + +describe("trigger idempotency keys", () => { + it("derives an event key from the envelope id", () => { + expect( + workflowEventIdempotencyKey("6f1c1f6e-0000-4000-8000-000000000000"), + ).toBe("event:6f1c1f6e-0000-4000-8000-000000000000"); + }); + + it("derives a cron key from the tick, to the minute", () => { + expect( + workflowCronIdempotencyKey({ + name: "nightly-reconciliation", + tick: new Date("2026-08-26T03:00:41.512Z"), + }), + ).toBe("cron:nightly-reconciliation:2026-08-26T03:00"); + }); + + it("collapses two deliveries of the same tick", () => { + const first = workflowCronIdempotencyKey({ + name: "nightly", + tick: new Date("2026-08-26T03:00:01.000Z"), + }); + const second = workflowCronIdempotencyKey({ + name: "nightly", + tick: new Date("2026-08-26T03:00:59.000Z"), + }); + + expect(first).toBe(second); + }); +}); + +describe("workflowIdempotencyScopeKey", () => { + const scope = { + idempotencyKey: "event:abc", + pluginId: "@vitnode/shop", + workflowId: "place-order", + workflowVersion: 1, + }; + + it("scopes a key to one definition version", () => { + expect(workflowIdempotencyScopeKey(scope)).toBe( + "@vitnode/shop:place-order@1:event:abc", + ); + }); + + it("treats a new workflow version as a different subscriber", () => { + expect(workflowIdempotencyScopeKey(scope)).not.toBe( + workflowIdempotencyScopeKey({ ...scope, workflowVersion: 2 }), + ); + }); + + it("does not collide across plugins reacting to the same event", () => { + expect(workflowIdempotencyScopeKey(scope)).not.toBe( + workflowIdempotencyScopeKey({ ...scope, pluginId: "@vitnode/blog" }), + ); + }); +}); diff --git a/packages/vitnode/src/api/workflows/idempotency.ts b/packages/vitnode/src/api/workflows/idempotency.ts new file mode 100644 index 000000000..16c4a8d26 --- /dev/null +++ b/packages/vitnode/src/api/workflows/idempotency.ts @@ -0,0 +1,93 @@ +import { + WORKFLOW_COMPENSATION_IDEMPOTENCY_SUFFIX, + WORKFLOW_CRON_IDEMPOTENCY_PREFIX, + WORKFLOW_EVENT_IDEMPOTENCY_PREFIX, + WORKFLOW_IDEMPOTENCY_PREFIX, +} from "./const"; + +/** + * The key a step body hands to a third-party API. + * + * Deterministic, so every attempt of the same step of the same execution - + * including the attempt that happens after a worker died mid-step - presents + * the same key and the provider collapses them into one side effect. + * + * ```ts + * await stripe.paymentIntents.create(payload, { + * idempotencyKey: ctx.idempotencyKey, + * }); + * ``` + */ +export const workflowStepIdempotencyKey = ({ + executionId, + stepId, +}: { + executionId: number; + stepId: string; +}): string => `${WORKFLOW_IDEMPOTENCY_PREFIX}:${executionId}:${stepId}`; + +/** + * Compensation gets its own key, never the step's. + * + * Sharing one would make "refund this charge" collide with "create this + * charge" at the provider: the second call would be answered with the first + * call's cached response and the refund would silently never happen. + */ +export const workflowStepCompensationIdempotencyKey = ({ + executionId, + stepId, +}: { + executionId: number; + stepId: string; +}): string => + `${workflowStepIdempotencyKey({ executionId, stepId })}:${WORKFLOW_COMPENSATION_IDEMPOTENCY_SUFFIX}`; + +/** + * Execution-level key for a workflow started by an event. + * + * `EventEnvelope.eventId` is the unit of delivery, so it is also the unit of + * de-duplication: a broker that delivers the same envelope twice must produce + * one execution. Uniqueness is enforced per definition + * (`pluginId + workflowId + workflowVersion + idempotencyKey`), so two + * different workflows may both react to the same event - and `place-order@2` + * is a different subscriber from `place-order@1`. + */ +export const workflowEventIdempotencyKey = (eventId: string): string => + `${WORKFLOW_EVENT_IDEMPOTENCY_PREFIX}:${eventId}`; + +/** + * Execution-level key for a workflow started by a cron tick. + * + * A cron job has no envelope id, so the tick itself is the unit: the endpoint + * can be triggered twice for the same minute (an external scheduler retrying, + * two instances racing) and only one execution may come out of it. + */ +export const workflowCronIdempotencyKey = ({ + name, + tick, +}: { + name: string; + tick: Date; +}): string => + `${WORKFLOW_CRON_IDEMPOTENCY_PREFIX}:${name}:${tick.toISOString().slice(0, 16)}`; + +export interface WorkflowIdempotencyScope { + idempotencyKey: string; + pluginId: string; + workflowId: string; + workflowVersion: number; +} + +/** + * The scope the unique index covers, as one string. + * + * Only for logs and tests - the database enforces the real constraint - but it + * keeps "what does idempotent mean here" answerable in one place. + */ +export const workflowIdempotencyScopeKey = ({ + idempotencyKey, + pluginId, + workflowId, + workflowVersion, +}: WorkflowIdempotencyScope): string => + `${pluginId}:${workflowId}@${workflowVersion}:${idempotencyKey}`; diff --git a/packages/vitnode/src/api/workflows/index.ts b/packages/vitnode/src/api/workflows/index.ts new file mode 100644 index 000000000..ddb3308ff --- /dev/null +++ b/packages/vitnode/src/api/workflows/index.ts @@ -0,0 +1,12 @@ +export * from "./const"; +export * from "./define"; +export * from "./errors"; +export * from "./idempotency"; +export * from "./plan"; +export * from "./queue-task"; +export * from "./registry"; +export * from "./retry"; +export * from "./state-machine"; +export * from "./step-outputs"; +export * from "./triggers"; +export type * from "./types"; diff --git a/packages/vitnode/src/api/workflows/plan.test.ts b/packages/vitnode/src/api/workflows/plan.test.ts new file mode 100644 index 000000000..27869e163 --- /dev/null +++ b/packages/vitnode/src/api/workflows/plan.test.ts @@ -0,0 +1,184 @@ +// @vitest-environment node +import { describe, expect, it } from "vitest"; +import { z } from "zod"; + +import type { RegisteredWorkflowDefinition } from "./types"; + +import { defineWorkflow } from "./define"; +import { WORKFLOW_ERROR_CODES, WorkflowError } from "./errors"; +import { planWorkflowStart } from "./plan"; + +const placeOrder = defineWorkflow({ + id: "place-order", + input: z.object({ orderId: z.number().int().positive() }), + steps: ({ step }) => [ + step({ + id: "reserve-inventory", + retry: { maxAttempts: 5 }, + run: () => undefined, + }), + step({ id: "authorize-payment", run: () => undefined }), + ], + version: 2, +}); + +const entry: RegisteredWorkflowDefinition = { + definition: placeOrder, + module: "orders", + pluginId: "@vitnode/shop", +}; + +const codeOf = (run: () => unknown): string => { + try { + run(); + } catch (error) { + return error instanceof WorkflowError ? error.code : "not-a-workflow-error"; + } + + return "did-not-throw"; +}; + +describe("planWorkflowStart execution row", () => { + it("stamps the exact definition identity the runner has to resolve", () => { + const plan = planWorkflowStart({ entry, input: { orderId: 42 } }); + + expect(plan.execution).toMatchObject({ + module: "orders", + pluginId: "@vitnode/shop", + status: "pending", + workflowId: "place-order", + workflowVersion: 2, + }); + }); + + it("starts with no compensation and no trigger metadata", () => { + const plan = planWorkflowStart({ entry, input: { orderId: 42 } }); + + expect(plan.execution.compensationStatus).toBe("none"); + expect(plan.execution.triggerType).toBe("manual"); + expect(plan.execution.triggerId).toBeNull(); + expect(plan.execution.triggerName).toBeNull(); + }); + + it("records the actor as metadata, defaulting to system", () => { + expect( + planWorkflowStart({ entry, input: { orderId: 42 } }).execution, + ).toMatchObject({ actorId: null, actorType: "system" }); + + expect( + planWorkflowStart({ + entry, + input: { orderId: 42 }, + options: { actor: { id: 9, type: "admin" } }, + }).execution, + ).toMatchObject({ actorId: 9, actorType: "admin" }); + }); + + it("stores the parsed input, so steps see exactly what was validated", () => { + const plan = planWorkflowStart({ + entry, + input: { extra: "dropped", orderId: 42 }, + }); + + expect(plan.execution.input).toEqual({ orderId: 42 }); + }); + + it("carries an event trigger's envelope id as the idempotency key", () => { + const plan = planWorkflowStart({ + entry, + input: { orderId: 42 }, + options: { + idempotencyKey: "event:abc", + trigger: { id: "abc", name: "order.created", type: "event" }, + }, + }); + + expect(plan.execution).toMatchObject({ + idempotencyKey: "event:abc", + triggerId: "abc", + triggerName: "order.created", + triggerType: "event", + }); + }); +}); + +describe("planWorkflowStart step rows", () => { + it("writes the whole plan up front, pending and in order", () => { + const plan = planWorkflowStart({ entry, input: { orderId: 42 } }); + + expect(plan.steps).toEqual([ + { + maxAttempts: 5, + position: 0, + status: "pending", + stepId: "reserve-inventory", + }, + { + maxAttempts: 3, + position: 1, + status: "pending", + stepId: "authorize-payment", + }, + ]); + }); + + it("queues one generic core task for the first step only", () => { + const plan = planWorkflowStart({ entry, input: { orderId: 42 } }); + + expect(plan.queue).toEqual({ + maxAttempts: 3, + name: "workflow-step", + pluginId: "@vitnode/core", + stepId: "reserve-inventory", + }); + }); +}); + +describe("planWorkflowStart validation", () => { + it("refuses input the workflow's schema rejects", () => { + expect( + codeOf(() => planWorkflowStart({ entry, input: { orderId: -1 } })), + ).toBe(WORKFLOW_ERROR_CODES.INVALID_INPUT); + }); + + it("refuses input that does not parse to an object", () => { + const scalar: RegisteredWorkflowDefinition = { + definition: defineWorkflow({ + id: "scalar-input", + input: z.number(), + steps: ({ step }) => [step({ id: "run", run: () => undefined })], + version: 1, + }), + module: "orders", + pluginId: "@vitnode/shop", + }; + + expect(codeOf(() => planWorkflowStart({ entry: scalar, input: 1 }))).toBe( + WORKFLOW_ERROR_CODES.INVALID_INPUT, + ); + }); + + it("refuses an empty idempotency key rather than silently ignoring it", () => { + expect( + codeOf(() => + planWorkflowStart({ + entry, + input: { orderId: 42 }, + options: { idempotencyKey: "" }, + }), + ), + ).toBe(WORKFLOW_ERROR_CODES.INVALID_INPUT); + }); + + it("refuses an idempotency key longer than the column", () => { + expect( + codeOf(() => + planWorkflowStart({ + entry, + input: { orderId: 42 }, + options: { idempotencyKey: "a".repeat(256) }, + }), + ), + ).toBe(WORKFLOW_ERROR_CODES.INVALID_INPUT); + }); +}); diff --git a/packages/vitnode/src/api/workflows/plan.ts b/packages/vitnode/src/api/workflows/plan.ts new file mode 100644 index 000000000..eff14299b --- /dev/null +++ b/packages/vitnode/src/api/workflows/plan.ts @@ -0,0 +1,120 @@ +import type { + RegisteredWorkflowDefinition, + WorkflowStartOptions, + WorkflowStartPlan, +} from "./types"; + +import { + WORKFLOW_IDEMPOTENCY_KEY_MAX_LENGTH, + WORKFLOW_STEP_QUEUE_MAX_ATTEMPTS, + WORKFLOW_STEP_QUEUE_TASK, +} from "./const"; +import { WORKFLOW_ERROR_CODES, WorkflowError } from "./errors"; + +const CORE_PLUGIN_ID = "@vitnode/core"; + +const isPlainObject = (value: unknown): value is Record => + typeof value === "object" && value !== null && !Array.isArray(value); + +/** + * Everything `WorkflowModel.start()` decides, with nothing written. + * + * Pure on purpose. "What does starting this workflow mean" is a property of + * the definition and the caller's arguments, not of the database, so it is + * decided - and unit-tested - without one. The persistence layer receives a + * finished plan and only has to insert it. + * + * Note what is *not* here: running a step. `start()` validates, writes rows and + * queues the first `workflow-step` task. Step 1 never executes inside the + * caller's HTTP request, so a slow inventory call cannot become a slow + * checkout response, and a crash after the commit still leaves the work queued. + */ +export const planWorkflowStart = ({ + entry, + input, + options = {}, +}: { + entry: RegisteredWorkflowDefinition; + input: unknown; + options?: Omit; +}): WorkflowStartPlan => { + const { definition, module, pluginId } = entry; + const parsed = definition.input.safeParse(input); + + if (!parsed.success) { + throw new WorkflowError( + WORKFLOW_ERROR_CODES.INVALID_INPUT, + `input does not match the workflow's schema: ${parsed.error.issues + .map(issue => `${issue.path.join(".") || "(root)"} ${issue.message}`) + .join("; ")}.`, + { pluginId, version: definition.version, workflowId: definition.id }, + ); + } + + if (!isPlainObject(parsed.data)) { + throw new WorkflowError( + WORKFLOW_ERROR_CODES.INVALID_INPUT, + "`input` must parse to a plain object. The execution's input is stored as JSONB and rendered field by field in the AdminCP, so a bare scalar or array has nothing to key on - wrap it, e.g. `z.object({ orderId: z.number() })`.", + { pluginId, version: definition.version, workflowId: definition.id }, + ); + } + + const idempotencyKey = options.idempotencyKey ?? null; + + if (idempotencyKey !== null) { + if (idempotencyKey.length === 0) { + throw new WorkflowError( + WORKFLOW_ERROR_CODES.INVALID_INPUT, + "`idempotencyKey` cannot be an empty string. Omit it entirely to start an execution that is not de-duplicated.", + { pluginId, version: definition.version, workflowId: definition.id }, + ); + } + + if (idempotencyKey.length > WORKFLOW_IDEMPOTENCY_KEY_MAX_LENGTH) { + throw new WorkflowError( + WORKFLOW_ERROR_CODES.INVALID_INPUT, + `\`idempotencyKey\` is longer than ${WORKFLOW_IDEMPOTENCY_KEY_MAX_LENGTH} characters.`, + { pluginId, version: definition.version, workflowId: definition.id }, + ); + } + } + + const trigger = options.trigger ?? { type: "manual" }; + const actor = options.actor ?? { type: "system" }; + const [firstStep] = definition.steps; + + return { + execution: { + actorId: actor.id ?? null, + actorType: actor.type, + compensationStatus: "none", + idempotencyKey, + input: parsed.data, + module, + pluginId, + status: "pending", + triggerId: trigger.id ?? null, + triggerName: trigger.name ?? null, + triggerType: trigger.type, + workflowId: definition.id, + workflowVersion: definition.version, + }, + // One generic core task, whoever owns the workflow. The runner resolves + // execution -> plugin -> workflow -> version -> step -> code from the row. + queue: { + maxAttempts: WORKFLOW_STEP_QUEUE_MAX_ATTEMPTS, + name: WORKFLOW_STEP_QUEUE_TASK, + pluginId: CORE_PLUGIN_ID, + stepId: firstStep.id, + }, + // Every step is written up front, `pending`, in declaration order. The + // execution's plan is therefore visible in the database from the moment it + // is created, rather than appearing one row at a time as it runs. + steps: definition.steps.map(step => ({ + maxAttempts: step.retry.maxAttempts, + position: step.position, + status: "pending", + stepId: step.id, + })), + }; +}; diff --git a/packages/vitnode/src/api/workflows/queue-task.ts b/packages/vitnode/src/api/workflows/queue-task.ts new file mode 100644 index 000000000..50d13b316 --- /dev/null +++ b/packages/vitnode/src/api/workflows/queue-task.ts @@ -0,0 +1,26 @@ +import { z } from "zod"; + +import { WORKFLOW_STEP_QUEUE_TASK } from "./const"; + +/** + * The payload of the one generic `@vitnode/core:workflow-step` task. + * + * Two identifiers and nothing else. Everything the runner needs - the plugin, + * the workflow, the version, the input, the outputs of earlier steps - is read + * from the execution row, so a task that has been sitting in the queue across a + * deploy still resolves against the version its execution started on. + */ +export const workflowStepTaskPayloadSchema = z.object({ + executionId: z.number().int().positive(), + stepId: z.string().min(1), +}); + +export type WorkflowStepTaskPayload = z.infer< + typeof workflowStepTaskPayloadSchema +>; + +export const workflowStepTaskPayload = ( + payload: WorkflowStepTaskPayload, +): Record => ({ ...payload }); + +export { WORKFLOW_STEP_QUEUE_TASK }; diff --git a/packages/vitnode/src/api/workflows/registry.test.ts b/packages/vitnode/src/api/workflows/registry.test.ts new file mode 100644 index 000000000..8d91a9105 --- /dev/null +++ b/packages/vitnode/src/api/workflows/registry.test.ts @@ -0,0 +1,225 @@ +// @vitest-environment node +import { describe, expect, it } from "vitest"; +import { z } from "zod"; + +import type { RegisteredWorkflowDefinition } from "./types"; + +import { defineWorkflow } from "./define"; +import { + WORKFLOW_ERROR_CODES, + WorkflowDefinitionNotFoundError, + WorkflowError, +} from "./errors"; +import { + nextWorkflowStep, + requireWorkflowDefinition, + requireWorkflowStep, + resolveWorkflowDefinition, + resolveWorkflowRegistration, + validateWorkflowDefinitions, + workflowDefinitionKey, + workflowDefinitionVersions, +} from "./registry"; + +const workflow = (id: string, version: number) => + defineWorkflow({ + id, + input: z.object({ orderId: z.number() }), + steps: ({ step }) => [ + step({ id: "reserve-inventory", run: () => undefined }), + step({ id: "authorize-payment", run: () => undefined }), + ], + version, + }); + +const entry = ( + pluginId: string, + id: string, + version: number, + module = "orders", +): RegisteredWorkflowDefinition => ({ + definition: workflow(id, version), + module, + pluginId, +}); + +const codeOf = (run: () => unknown): string => { + try { + run(); + } catch (error) { + return error instanceof WorkflowError ? error.code : "not-a-workflow-error"; + } + + return "did-not-throw"; +}; + +describe("validateWorkflowDefinitions", () => { + it("rejects the same pluginId + workflowId + version twice", () => { + expect( + codeOf(() => + validateWorkflowDefinitions([ + entry("@vitnode/shop", "place-order", 1, "orders"), + entry("@vitnode/shop", "place-order", 1, "checkout"), + ]), + ), + ).toBe(WORKFLOW_ERROR_CODES.DUPLICATE_DEFINITION); + }); + + it("allows v1 and v2 of the same workflow side by side", () => { + const entries = validateWorkflowDefinitions([ + entry("@vitnode/shop", "place-order", 1), + entry("@vitnode/shop", "place-order", 2), + ]); + + expect(entries).toHaveLength(2); + }); + + it("allows the same workflow id in two different plugins", () => { + expect( + validateWorkflowDefinitions([ + entry("@vitnode/shop", "place-order", 1), + entry("@vitnode/blog", "place-order", 1), + ]), + ).toHaveLength(2); + }); +}); + +describe("resolveWorkflowDefinition", () => { + const entries = [ + entry("@vitnode/shop", "place-order", 1), + entry("@vitnode/shop", "place-order", 2), + ]; + + it("resolves the exact version asked for", () => { + expect( + resolveWorkflowDefinition(entries, { + pluginId: "@vitnode/shop", + version: 1, + workflowId: "place-order", + })?.definition.version, + ).toBe(1); + }); + + it("never falls back to the latest version", () => { + // The whole point of storing `workflowVersion` on the execution: v1 was + // removed from the deployment, and running v2's steps against a plan made + // for v1 would be silent corruption. + const onlyV2 = [entry("@vitnode/shop", "place-order", 2)]; + + expect( + resolveWorkflowDefinition(onlyV2, { + pluginId: "@vitnode/shop", + version: 1, + workflowId: "place-order", + }), + ).toBeUndefined(); + }); + + it("does not resolve across plugins", () => { + expect( + resolveWorkflowDefinition(entries, { + pluginId: "@vitnode/blog", + version: 1, + workflowId: "place-order", + }), + ).toBeUndefined(); + }); + + it("gives a structured error for a missing version", () => { + let thrown: unknown; + + try { + requireWorkflowDefinition([entry("@vitnode/shop", "place-order", 2)], { + pluginId: "@vitnode/shop", + version: 1, + workflowId: "place-order", + }); + } catch (error) { + thrown = error; + } + + expect(thrown).toBeInstanceOf(WorkflowDefinitionNotFoundError); + expect((thrown as WorkflowDefinitionNotFoundError).code).toBe( + WORKFLOW_ERROR_CODES.DEFINITION_NOT_FOUND, + ); + expect((thrown as WorkflowDefinitionNotFoundError).version).toBe(1); + }); + + it("lists deployed versions oldest first", () => { + expect( + workflowDefinitionVersions( + [ + entry("@vitnode/shop", "place-order", 2), + entry("@vitnode/shop", "place-order", 1), + entry("@vitnode/shop", "refund-order", 1), + ], + { pluginId: "@vitnode/shop", workflowId: "place-order" }, + ), + ).toEqual([1, 2]); + }); + + it("builds an identity key from plugin, id and version", () => { + expect( + workflowDefinitionKey({ + pluginId: "@vitnode/shop", + version: 3, + workflowId: "place-order", + }), + ).toBe("@vitnode/shop:place-order@3"); + }); +}); + +describe("resolveWorkflowRegistration", () => { + it("finds the owning plugin of a definition the caller holds", () => { + const definition = workflow("place-order", 1); + const entries = [ + { definition, module: "orders", pluginId: "@vitnode/shop" }, + ]; + + expect(resolveWorkflowRegistration(entries, definition).pluginId).toBe( + "@vitnode/shop", + ); + }); + + it("refuses a definition no module registered", () => { + expect( + codeOf(() => resolveWorkflowRegistration([], workflow("place-order", 1))), + ).toBe(WORKFLOW_ERROR_CODES.DEFINITION_NOT_FOUND); + }); + + it("refuses a definition two plugins registered", () => { + const definition = workflow("place-order", 1); + + expect( + codeOf(() => + resolveWorkflowRegistration( + [ + { definition, module: "orders", pluginId: "@vitnode/shop" }, + { definition, module: "orders", pluginId: "@vitnode/blog" }, + ], + definition, + ), + ), + ).toBe(WORKFLOW_ERROR_CODES.DUPLICATE_DEFINITION); + }); +}); + +describe("step lookup", () => { + const definition = workflow("place-order", 1); + + it("walks steps in declaration order", () => { + expect(nextWorkflowStep(definition, "reserve-inventory")?.id).toBe( + "authorize-payment", + ); + }); + + it("has no next step after the last one", () => { + expect(nextWorkflowStep(definition, "authorize-payment")).toBeUndefined(); + }); + + it("refuses a step the definition does not declare", () => { + expect(codeOf(() => requireWorkflowStep(definition, "ship-order"))).toBe( + WORKFLOW_ERROR_CODES.STEP_NOT_FOUND, + ); + }); +}); diff --git a/packages/vitnode/src/api/workflows/registry.ts b/packages/vitnode/src/api/workflows/registry.ts new file mode 100644 index 000000000..4b3c48123 --- /dev/null +++ b/packages/vitnode/src/api/workflows/registry.ts @@ -0,0 +1,189 @@ +import type { + AnyWorkflowDefinition, + RegisteredWorkflowDefinition, + ResolvedWorkflowStep, + WorkflowDefinitionRef, +} from "./types"; + +import { + WORKFLOW_ERROR_CODES, + WorkflowDefinitionNotFoundError, + WorkflowError, +} from "./errors"; + +/** + * Identity of one registered definition: `pluginId + workflowId + version`. + * + * `workflowId` alone is never an identity. Two versions of the same workflow + * are two different programs that happen to share a name, and an execution + * that started on v1 must keep running v1 forever. + */ +export const workflowDefinitionKey = ({ + pluginId, + version, + workflowId, +}: WorkflowDefinitionRef): string => `${pluginId}:${workflowId}@${version}`; + +export const workflowDefinitionRef = ( + entry: RegisteredWorkflowDefinition, +): WorkflowDefinitionRef => ({ + pluginId: entry.pluginId, + version: entry.definition.version, + workflowId: entry.definition.id, +}); + +/** + * Refuses two definitions with the same identity, and returns the rest + * untouched. + * + * Run inside `buildApiPlugin` (collisions within one plugin) and again in the + * global middleware across every installed plugin, for the same reason content + * types and search indexers are: a plugin can only see its own modules. + * + * Registering `place-order@1` and `place-order@2` side by side is not just + * allowed, it is the supported deployment: old versions stay registered until + * no execution needs them. + */ +export const validateWorkflowDefinitions = ( + entries: readonly RegisteredWorkflowDefinition[], +): RegisteredWorkflowDefinition[] => { + const seen = new Map(); + + for (const entry of entries) { + const key = workflowDefinitionKey(workflowDefinitionRef(entry)); + const owner = seen.get(key); + + if (owner) { + throw new WorkflowError( + WORKFLOW_ERROR_CODES.DUPLICATE_DEFINITION, + `already registered by module "${owner.module}" and again by "${entry.module}". A workflow is identified by plugin, id and version - bump \`version\` to register a second definition under the same id.`, + { + pluginId: entry.pluginId, + version: entry.definition.version, + workflowId: entry.definition.id, + }, + ); + } + + seen.set(key, entry); + } + + return [...entries]; +}; + +/** + * Exact-version lookup. There is deliberately no "latest" variant: an + * execution row carries the version it started with, and resolving anything + * else would silently migrate in-flight work onto code it was never planned + * against. + */ +export const resolveWorkflowDefinition = ( + entries: readonly RegisteredWorkflowDefinition[], + ref: WorkflowDefinitionRef, +): RegisteredWorkflowDefinition | undefined => + entries.find( + entry => + entry.pluginId === ref.pluginId && + entry.definition.id === ref.workflowId && + entry.definition.version === ref.version, + ); + +/** + * Same lookup, but throws {@link WorkflowDefinitionNotFoundError} instead of + * returning `undefined`. The runner catches it, fails the execution with the + * `WORKFLOW_DEFINITION_NOT_FOUND` code and leaves every row in place for an + * operator to inspect. + */ +export const requireWorkflowDefinition = ( + entries: readonly RegisteredWorkflowDefinition[], + ref: WorkflowDefinitionRef, +): RegisteredWorkflowDefinition => { + const entry = resolveWorkflowDefinition(entries, ref); + if (entry) return entry; + + throw new WorkflowDefinitionNotFoundError({ + pluginId: ref.pluginId, + version: ref.version, + workflowId: ref.workflowId, + }); +}; + +/** Every version of one workflow that is currently deployed, oldest first. */ +export const workflowDefinitionVersions = ( + entries: readonly RegisteredWorkflowDefinition[], + { pluginId, workflowId }: Omit, +): number[] => + entries + .filter( + entry => + entry.pluginId === pluginId && entry.definition.id === workflowId, + ) + .map(entry => entry.definition.version) + .sort((a, b) => a - b); + +/** + * Finds the registration of a definition the caller already holds. + * + * Resolution is by object identity rather than by id, because `start()` has to + * answer a question the caller cannot: *which plugin owns this*. Guessing from + * `c.get("plugin")` would be wrong the moment one plugin starts another's + * workflow, and would write an execution row the runner can never resolve. + * + * Not being registered is a hard error for the same reason: an unregistered + * definition would produce an execution whose steps no deployment can find. + */ +export const resolveWorkflowRegistration = ( + entries: readonly RegisteredWorkflowDefinition[], + definition: AnyWorkflowDefinition, +): RegisteredWorkflowDefinition => { + const matches = entries.filter(entry => entry.definition === definition); + const [entry] = matches; + + if (!entry) { + throw new WorkflowError( + WORKFLOW_ERROR_CODES.DEFINITION_NOT_FOUND, + "is not registered. Add it to a module's `workflows: []` - the runner resolves a queued step by plugin, id and version from the execution row, so a definition nothing registered could never be picked up again.", + { version: definition.version, workflowId: definition.id }, + ); + } + + if (matches.length > 1) { + throw new WorkflowError( + WORKFLOW_ERROR_CODES.DUPLICATE_DEFINITION, + `is registered by more than one plugin (${matches.map(match => match.pluginId).join(", ")}), so its owner is ambiguous. Register a definition in exactly one plugin.`, + { version: definition.version, workflowId: definition.id }, + ); + } + + return entry; +}; + +export const findWorkflowStep = ( + definition: AnyWorkflowDefinition, + stepId: string, +): ResolvedWorkflowStep | undefined => + definition.steps.find(step => step.id === stepId); + +export const requireWorkflowStep = ( + definition: AnyWorkflowDefinition, + stepId: string, +): ResolvedWorkflowStep => { + const step = findWorkflowStep(definition, stepId); + if (step) return step; + + throw new WorkflowError( + WORKFLOW_ERROR_CODES.STEP_NOT_FOUND, + `has no step "${stepId}". The execution was planned against a different step list, which means its version was changed in place instead of being bumped.`, + { version: definition.version, workflowId: definition.id }, + ); +}; + +/** The step that runs after `stepId`, or `undefined` when the workflow is done. */ +export const nextWorkflowStep = ( + definition: AnyWorkflowDefinition, + stepId: string, +): ResolvedWorkflowStep | undefined => { + const current = requireWorkflowStep(definition, stepId); + + return definition.steps.find(step => step.position === current.position + 1); +}; diff --git a/packages/vitnode/src/api/workflows/retry.test.ts b/packages/vitnode/src/api/workflows/retry.test.ts new file mode 100644 index 000000000..3fe3a1cfb --- /dev/null +++ b/packages/vitnode/src/api/workflows/retry.test.ts @@ -0,0 +1,109 @@ +// @vitest-environment node +import { describe, expect, it } from "vitest"; + +import type { WorkflowRetryPolicyInput } from "./retry"; + +import { WORKFLOW_ERROR_CODES, WorkflowError } from "./errors"; +import { + DEFAULT_WORKFLOW_RETRY_POLICY, + nextWorkflowAttemptAt, + resolveWorkflowRetryPolicy, + workflowRetryDelayMs, +} from "./retry"; + +const codeOf = (run: () => unknown): string => { + try { + run(); + } catch (error) { + return error instanceof WorkflowError ? error.code : "not-a-workflow-error"; + } + + return "did-not-throw"; +}; + +describe("resolveWorkflowRetryPolicy", () => { + it("fills every key in from the default", () => { + expect(resolveWorkflowRetryPolicy(undefined)).toEqual( + DEFAULT_WORKFLOW_RETRY_POLICY, + ); + }); + + it("keeps what the step declared", () => { + expect( + resolveWorkflowRetryPolicy({ maxAttempts: 5, strategy: "fixed" }), + ).toEqual({ + initialDelayMs: 1_000, + maxAttempts: 5, + maxDelayMs: 60_000, + strategy: "fixed", + }); + }); + + it.each([ + { maxAttempts: 0 }, + { maxAttempts: 2.5 }, + { maxAttempts: 26 }, + { initialDelayMs: -1 }, + { initialDelayMs: 10_000, maxDelayMs: 1_000 }, + { strategy: "linear" as never }, + ])("rejects %o", policy => { + expect(codeOf(() => resolveWorkflowRetryPolicy(policy))).toBe( + WORKFLOW_ERROR_CODES.INVALID_RETRY_POLICY, + ); + }); + + it("names the step in the error", () => { + try { + resolveWorkflowRetryPolicy({ maxAttempts: 0 }, { stepId: "reserve" }); + } catch (error) { + expect((error as WorkflowError).message).toContain('step "reserve"'); + } + }); +}); + +describe("workflowRetryDelayMs", () => { + const exponential = resolveWorkflowRetryPolicy({ + initialDelayMs: 1_000, + maxDelayMs: 10_000, + strategy: "exponential", + }); + + it("doubles each attempt", () => { + expect(workflowRetryDelayMs(exponential, 1)).toBe(1_000); + expect(workflowRetryDelayMs(exponential, 2)).toBe(2_000); + expect(workflowRetryDelayMs(exponential, 3)).toBe(4_000); + }); + + it("caps at maxDelayMs", () => { + expect(workflowRetryDelayMs(exponential, 10)).toBe(10_000); + }); + + it("stays flat for the fixed strategy", () => { + const fixed = resolveWorkflowRetryPolicy({ + initialDelayMs: 2_500, + strategy: "fixed", + }); + + expect(workflowRetryDelayMs(fixed, 1)).toBe(2_500); + expect(workflowRetryDelayMs(fixed, 7)).toBe(2_500); + }); +}); + +describe("nextWorkflowAttemptAt", () => { + const now = new Date("2026-01-01T00:00:00.000Z"); + const policy = resolveWorkflowRetryPolicy({ maxAttempts: 3 }); + + it("schedules the next attempt while the budget lasts", () => { + expect(nextWorkflowAttemptAt(policy, 1, now)?.toISOString()).toBe( + "2026-01-01T00:00:01.000Z", + ); + expect(nextWorkflowAttemptAt(policy, 2, now)?.toISOString()).toBe( + "2026-01-01T00:00:02.000Z", + ); + }); + + it("returns null once maxAttempts is reached, which is what fails the step", () => { + expect(nextWorkflowAttemptAt(policy, 3, now)).toBeNull(); + expect(nextWorkflowAttemptAt(policy, 4, now)).toBeNull(); + }); +}); diff --git a/packages/vitnode/src/api/workflows/retry.ts b/packages/vitnode/src/api/workflows/retry.ts new file mode 100644 index 000000000..9ef63f5cb --- /dev/null +++ b/packages/vitnode/src/api/workflows/retry.ts @@ -0,0 +1,135 @@ +import { + WORKFLOW_MAX_RETRY_ATTEMPTS, + WORKFLOW_RETRY_STRATEGIES, +} from "./const"; +import { WORKFLOW_ERROR_CODES, WorkflowError } from "./errors"; + +export type WorkflowRetryStrategy = (typeof WORKFLOW_RETRY_STRATEGIES)[number]; + +export interface WorkflowRetryPolicy { + initialDelayMs: number; + maxAttempts: number; + maxDelayMs: number; + strategy: WorkflowRetryStrategy; +} + +/** What a step declares. Every key has a default; see {@link DEFAULT_WORKFLOW_RETRY_POLICY}. */ +export type WorkflowRetryPolicyInput = Partial; + +/** + * `maxAttempts: 3` means the step body runs at most three times in total - + * one first run and two retries - not three retries after the first failure. + */ +export const DEFAULT_WORKFLOW_RETRY_POLICY: WorkflowRetryPolicy = { + initialDelayMs: 1_000, + maxAttempts: 3, + maxDelayMs: 60_000, + strategy: "exponential", +}; + +const invalid = (message: string, context: { stepId?: string } = {}) => + new WorkflowError( + WORKFLOW_ERROR_CODES.INVALID_RETRY_POLICY, + context.stepId + ? `step "${context.stepId}" has an invalid retry policy: ${message}` + : `invalid retry policy: ${message}`, + ); + +const positiveInteger = (value: number): boolean => + Number.isSafeInteger(value) && value > 0; + +/** + * Fills a step's `retry` in and refuses a policy that cannot be honoured. + * + * Validated once, at definition time, so a workflow with a nonsense backoff + * fails at boot rather than the first time a step happens to throw in + * production - which could be weeks later. + */ +export const resolveWorkflowRetryPolicy = ( + input: undefined | WorkflowRetryPolicyInput, + context: { stepId?: string } = {}, +): WorkflowRetryPolicy => { + const policy = { ...DEFAULT_WORKFLOW_RETRY_POLICY, ...input }; + + if (!positiveInteger(policy.maxAttempts)) { + throw invalid( + `\`maxAttempts\` must be a positive integer, received ${String(policy.maxAttempts)}.`, + context, + ); + } + + if (policy.maxAttempts > WORKFLOW_MAX_RETRY_ATTEMPTS) { + throw invalid( + `\`maxAttempts\` may not exceed ${WORKFLOW_MAX_RETRY_ATTEMPTS}, received ${policy.maxAttempts}.`, + context, + ); + } + + if (!WORKFLOW_RETRY_STRATEGIES.includes(policy.strategy)) { + throw invalid( + `\`strategy\` must be one of ${WORKFLOW_RETRY_STRATEGIES.join(", ")}, received "${policy.strategy}".`, + context, + ); + } + + if ( + !Number.isSafeInteger(policy.initialDelayMs) || + policy.initialDelayMs < 0 + ) { + throw invalid( + `\`initialDelayMs\` must be a non-negative integer, received ${String(policy.initialDelayMs)}.`, + context, + ); + } + + if (!Number.isSafeInteger(policy.maxDelayMs) || policy.maxDelayMs < 0) { + throw invalid( + `\`maxDelayMs\` must be a non-negative integer, received ${String(policy.maxDelayMs)}.`, + context, + ); + } + + if (policy.maxDelayMs < policy.initialDelayMs) { + throw invalid( + `\`maxDelayMs\` (${policy.maxDelayMs}) must be greater than or equal to \`initialDelayMs\` (${policy.initialDelayMs}).`, + context, + ); + } + + return policy; +}; + +/** + * Delay before attempt number `attempts + 1`, where `attempts` counts the runs + * already made (>= 1 when a retry is being scheduled). + * + * The one place the backoff curve is written down. Runners, the AdminCP's + * "next attempt" column and the tests all read it from here rather than + * recomputing `2 ** n` in three places that then disagree. + */ +export const workflowRetryDelayMs = ( + policy: WorkflowRetryPolicy, + attempts: number, +): number => { + const exponent = Math.max(0, attempts - 1); + const delay = + policy.strategy === "fixed" + ? policy.initialDelayMs + : policy.initialDelayMs * 2 ** exponent; + + return Math.min(delay, policy.maxDelayMs); +}; + +/** + * When the step may run again, or `null` when the policy is exhausted and the + * step has to be marked `failed`. + */ +export const nextWorkflowAttemptAt = ( + policy: WorkflowRetryPolicy, + attempts: number, + from: Date = new Date(), +): Date | null => { + if (attempts >= policy.maxAttempts) return null; + + return new Date(from.getTime() + workflowRetryDelayMs(policy, attempts)); +}; diff --git a/packages/vitnode/src/api/workflows/state-machine.test.ts b/packages/vitnode/src/api/workflows/state-machine.test.ts new file mode 100644 index 000000000..74c5c5855 --- /dev/null +++ b/packages/vitnode/src/api/workflows/state-machine.test.ts @@ -0,0 +1,89 @@ +// @vitest-environment node +import { describe, expect, it } from "vitest"; + +import { WORKFLOW_ERROR_CODES, WorkflowError } from "./errors"; +import { + assertWorkflowExecutionTransition, + assertWorkflowStepTransition, + canTransitionWorkflowCompensation, + canTransitionWorkflowExecution, + canTransitionWorkflowStep, + isWorkflowExecutionFinished, +} from "./state-machine"; + +const codeOf = (run: () => unknown): string => { + try { + run(); + } catch (error) { + return error instanceof WorkflowError ? error.code : "not-a-workflow-error"; + } + + return "did-not-throw"; +}; + +describe("execution state machine", () => { + it.each([ + ["pending", "running"], + ["pending", "cancelled"], + ["running", "completed"], + ["running", "failed"], + ["running", "cancelled"], + ] as const)("allows %s -> %s", (from, to) => { + expect(canTransitionWorkflowExecution(from, to)).toBe(true); + }); + + it.each([ + ["pending", "completed"], + ["completed", "running"], + ["cancelled", "running"], + ] as const)("refuses %s -> %s", (from, to) => { + expect(canTransitionWorkflowExecution(from, to)).toBe(false); + }); + + it("keeps failed terminal, so nothing resumes a workflow by accident", () => { + expect(canTransitionWorkflowExecution("failed", "running")).toBe(false); + expect( + codeOf(() => assertWorkflowExecutionTransition("failed", "running")), + ).toBe(WORKFLOW_ERROR_CODES.INVALID_TRANSITION); + }); + + it("knows which statuses are finished", () => { + expect(isWorkflowExecutionFinished("completed")).toBe(true); + expect(isWorkflowExecutionFinished("failed")).toBe(true); + expect(isWorkflowExecutionFinished("cancelled")).toBe(true); + expect(isWorkflowExecutionFinished("running")).toBe(false); + expect(isWorkflowExecutionFinished("pending")).toBe(false); + }); +}); + +describe("step state machine", () => { + it("lets a failed attempt go back to pending for a scheduled retry", () => { + expect(canTransitionWorkflowStep("running", "pending")).toBe(true); + }); + + it("lets a step that never started be skipped", () => { + expect(canTransitionWorkflowStep("pending", "skipped")).toBe(true); + }); + + it("refuses to skip a step that already ran", () => { + expect(canTransitionWorkflowStep("completed", "skipped")).toBe(false); + expect( + codeOf(() => assertWorkflowStepTransition("completed", "skipped")), + ).toBe(WORKFLOW_ERROR_CODES.INVALID_TRANSITION); + }); + + it("refuses to start a step directly from pending to completed", () => { + expect(canTransitionWorkflowStep("pending", "completed")).toBe(false); + }); +}); + +describe("compensation state machine", () => { + it("starts from none", () => { + expect(canTransitionWorkflowCompensation("none", "pending")).toBe(true); + expect(canTransitionWorkflowCompensation("none", "running")).toBe(false); + }); + + it("retries independently of the step", () => { + expect(canTransitionWorkflowCompensation("running", "pending")).toBe(true); + }); +}); diff --git a/packages/vitnode/src/api/workflows/state-machine.ts b/packages/vitnode/src/api/workflows/state-machine.ts new file mode 100644 index 000000000..918b5aea6 --- /dev/null +++ b/packages/vitnode/src/api/workflows/state-machine.ts @@ -0,0 +1,170 @@ +import type { + WORKFLOW_COMPENSATION_STATUSES, + WORKFLOW_EXECUTION_STATUSES, + WORKFLOW_STEP_STATUSES, +} from "./const"; + +import { WORKFLOW_ERROR_CODES, WorkflowError } from "./errors"; + +export type WorkflowExecutionStatus = + (typeof WORKFLOW_EXECUTION_STATUSES)[number]; +export type WorkflowStepStatus = (typeof WORKFLOW_STEP_STATUSES)[number]; +export type WorkflowCompensationStatus = + (typeof WORKFLOW_COMPENSATION_STATUSES)[number]; + +/** + * ```text + * pending -> running | cancelled + * running -> completed | failed | cancelled + * completed -> (terminal) + * failed -> (terminal) + * cancelled -> (terminal) + * ``` + * + * `failed` is terminal on purpose. Operator-initiated resume (`failed -> + * running`) is a deliberate future extension and has to be added here first, + * so nothing can quietly restart a workflow whose steps were never written to + * be re-entered. + * + * `pending -> cancelled` is the cancellation of a workflow the runner has not + * picked up yet; `running -> cancelled` is the one it noticed *between* steps. + * The engine never interrupts a step that is already executing - see + * {@link WorkflowExecutionRecord.cancellationRequestedAt}. + */ +export const WORKFLOW_EXECUTION_TRANSITIONS: Record< + WorkflowExecutionStatus, + readonly WorkflowExecutionStatus[] +> = { + cancelled: [], + completed: [], + failed: [], + pending: ["running", "cancelled"], + running: ["completed", "failed", "cancelled"], +}; + +/** + * ```text + * pending -> running | skipped + * running -> completed | failed | pending + * completed -> (terminal) + * failed -> (terminal) + * skipped -> (terminal) + * ``` + * + * `running -> pending` is a scheduled retry: the attempt failed, the policy + * still allows another, and `nextAttemptAt` says when. `pending -> skipped` is + * a step the runner never started because the execution was cancelled or an + * earlier step failed. + */ +export const WORKFLOW_STEP_TRANSITIONS: Record< + WorkflowStepStatus, + readonly WorkflowStepStatus[] +> = { + completed: [], + failed: [], + pending: ["running", "skipped"], + running: ["completed", "failed", "pending"], + skipped: [], +}; + +/** + * ```text + * none -> pending + * pending -> running + * running -> completed | failed | pending + * ``` + * + * `running -> pending` is compensation's own retry, which is independent of + * the step's: a step that ran three times and a rollback that has to be + * attempted five are unrelated budgets. + */ +export const WORKFLOW_COMPENSATION_TRANSITIONS: Record< + WorkflowCompensationStatus, + readonly WorkflowCompensationStatus[] +> = { + completed: [], + failed: [], + none: ["pending"], + pending: ["running"], + running: ["completed", "failed", "pending"], +}; + +const canTransition = ( + table: Record, + from: T, + to: T, +): boolean => table[from].includes(to); + +const assertTransition = ( + table: Record, + what: string, + from: T, + to: T, +): void => { + if (canTransition(table, from, to)) return; + + const allowed = table[from]; + + throw new WorkflowError( + WORKFLOW_ERROR_CODES.INVALID_TRANSITION, + `${what} cannot move from "${from}" to "${to}". ${ + allowed.length + ? `Allowed: ${allowed.join(", ")}.` + : `"${from}" is terminal.` + }`, + ); +}; + +export const canTransitionWorkflowExecution = ( + from: WorkflowExecutionStatus, + to: WorkflowExecutionStatus, +): boolean => canTransition(WORKFLOW_EXECUTION_TRANSITIONS, from, to); + +export const assertWorkflowExecutionTransition = ( + from: WorkflowExecutionStatus, + to: WorkflowExecutionStatus, +): void => + assertTransition( + WORKFLOW_EXECUTION_TRANSITIONS, + "A workflow execution", + from, + to, + ); + +export const canTransitionWorkflowStep = ( + from: WorkflowStepStatus, + to: WorkflowStepStatus, +): boolean => canTransition(WORKFLOW_STEP_TRANSITIONS, from, to); + +export const assertWorkflowStepTransition = ( + from: WorkflowStepStatus, + to: WorkflowStepStatus, +): void => + assertTransition(WORKFLOW_STEP_TRANSITIONS, "A workflow step", from, to); + +export const canTransitionWorkflowCompensation = ( + from: WorkflowCompensationStatus, + to: WorkflowCompensationStatus, +): boolean => canTransition(WORKFLOW_COMPENSATION_TRANSITIONS, from, to); + +export const assertWorkflowCompensationTransition = ( + from: WorkflowCompensationStatus, + to: WorkflowCompensationStatus, +): void => + assertTransition( + WORKFLOW_COMPENSATION_TRANSITIONS, + "Workflow compensation", + from, + to, + ); + +export const WORKFLOW_EXECUTION_TERMINAL_STATUSES = [ + "completed", + "failed", + "cancelled", +] as const satisfies readonly WorkflowExecutionStatus[]; + +export const isWorkflowExecutionFinished = ( + status: WorkflowExecutionStatus, +): boolean => + (WORKFLOW_EXECUTION_TERMINAL_STATUSES as readonly string[]).includes(status); diff --git a/packages/vitnode/src/api/workflows/step-outputs.test.ts b/packages/vitnode/src/api/workflows/step-outputs.test.ts new file mode 100644 index 000000000..5cc0fb6c5 --- /dev/null +++ b/packages/vitnode/src/api/workflows/step-outputs.test.ts @@ -0,0 +1,60 @@ +// @vitest-environment node +import { describe, expect, it } from "vitest"; +import { z } from "zod"; + +import { WORKFLOW_ERROR_CODES, WorkflowError } from "./errors"; +import { createWorkflowStepOutputs } from "./step-outputs"; + +const codeOf = (run: () => unknown): string => { + try { + run(); + } catch (error) { + return error instanceof WorkflowError ? error.code : "not-a-workflow-error"; + } + + return "did-not-throw"; +}; + +describe("createWorkflowStepOutputs", () => { + const outputs = createWorkflowStepOutputs({ + "reserve-inventory": { reservationId: 7 }, + }); + + it("reports what has already completed", () => { + expect(outputs.has("reserve-inventory")).toBe(true); + expect(outputs.has("authorize-payment")).toBe(false); + }); + + it("parses a previous step's output into a typed value", () => { + const { reservationId } = outputs.parse( + "reserve-inventory", + z.object({ reservationId: z.number() }), + ); + + expect(reservationId).toBe(7); + }); + + it("refuses to read a step that has not completed", () => { + expect(codeOf(() => outputs.parse("authorize-payment", z.object({})))).toBe( + WORKFLOW_ERROR_CODES.STEP_NOT_FOUND, + ); + }); + + it("refuses an output that no longer matches the expected shape", () => { + expect( + codeOf(() => + outputs.parse( + "reserve-inventory", + z.object({ reservationId: z.string() }), + ), + ), + ).toBe(WORKFLOW_ERROR_CODES.STEP_OUTPUT_INVALID); + }); + + it("treats a stored null as a real output rather than a missing one", () => { + const withNull = createWorkflowStepOutputs({ "send-email": null }); + + expect(withNull.has("send-email")).toBe(true); + expect(withNull.parse("send-email", z.null())).toBeNull(); + }); +}); diff --git a/packages/vitnode/src/api/workflows/step-outputs.ts b/packages/vitnode/src/api/workflows/step-outputs.ts new file mode 100644 index 000000000..1666b2a16 --- /dev/null +++ b/packages/vitnode/src/api/workflows/step-outputs.ts @@ -0,0 +1,43 @@ +import type { z } from "zod"; + +import type { WorkflowStepOutputs } from "./types"; + +import { WORKFLOW_ERROR_CODES, WorkflowError } from "./errors"; + +/** + * Wraps the outputs of already-completed steps, as they came back out of + * JSONB. + * + * The runner reads them from `core_workflow_step_executions.output`, so after + * a restart there is no in-memory object to hand on - only parsed JSON. That + * is why `get` is `unknown` and `parse` exists: a step that depends on an + * earlier step's shape says so, and finds out at the boundary rather than + * three lines later on a property that is suddenly a string. + */ +export const createWorkflowStepOutputs = ( + outputs: Readonly>, +): WorkflowStepOutputs => ({ + get: stepId => outputs[stepId], + has: stepId => Object.hasOwn(outputs, stepId), + parse: (stepId: string, schema: z.ZodType): TOutput => { + if (!Object.hasOwn(outputs, stepId)) { + throw new WorkflowError( + WORKFLOW_ERROR_CODES.STEP_NOT_FOUND, + `step "${stepId}" has no recorded output yet. Only steps that completed earlier in this execution can be read - a step cannot depend on one declared after it.`, + ); + } + + const result = schema.safeParse(outputs[stepId]); + + if (!result.success) { + throw new WorkflowError( + WORKFLOW_ERROR_CODES.STEP_OUTPUT_INVALID, + `the recorded output of step "${stepId}" does not match the expected schema: ${result.error.issues + .map(issue => `${issue.path.join(".") || "(root)"} ${issue.message}`) + .join("; ")}.`, + ); + } + + return result.data; + }, +}); diff --git a/packages/vitnode/src/api/workflows/store.ts b/packages/vitnode/src/api/workflows/store.ts new file mode 100644 index 000000000..297f2f860 --- /dev/null +++ b/packages/vitnode/src/api/workflows/store.ts @@ -0,0 +1,34 @@ +import type { WorkflowStore } from "./types"; + +import { WORKFLOW_ERROR_CODES, WorkflowError } from "./errors"; + +const notImplemented = (method: keyof WorkflowStore): never => { + throw new WorkflowError( + WORKFLOW_ERROR_CODES.NOT_IMPLEMENTED, + `\`WorkflowStore.${method}()\` has no implementation yet. The Workflow Engine's contracts are frozen but its persistence layer is not written - see docs/architecture/0001-workflow-engine.md.`, + ); +}; + +/** + * The Wave 0 placeholder for the persistence layer. + * + * Every method throws a structured `WORKFLOW_NOT_IMPLEMENTED`. Nothing in core + * calls them, so this changes no existing behaviour; it exists so the SDK, the + * runtime model and the runner can all be written and type-checked against a + * real boundary while the Drizzle implementation is written behind it. + * + * Replaced wholesale by the Drizzle store - the contract is + * {@link WorkflowStore} in `types.ts`, and it does not move when this file is + * implemented. + */ +export const workflowStore: WorkflowStore = { + claimStep: () => notImplemented("claimStep"), + completeStep: () => notImplemented("completeStep"), + createExecution: () => notImplemented("createExecution"), + failStep: () => notImplemented("failStep"), + finishExecution: () => notImplemented("finishExecution"), + loadExecution: () => notImplemented("loadExecution"), + requestCancellation: () => notImplemented("requestCancellation"), + skipPendingSteps: () => notImplemented("skipPendingSteps"), + startExecution: () => notImplemented("startExecution"), +}; diff --git a/packages/vitnode/src/api/workflows/triggers.ts b/packages/vitnode/src/api/workflows/triggers.ts new file mode 100644 index 000000000..1bd091590 --- /dev/null +++ b/packages/vitnode/src/api/workflows/triggers.ts @@ -0,0 +1,132 @@ +import type { z } from "zod"; + +import type { + EventEnvelope, + VitNodeEventName, + VitNodeEvents, +} from "../models/events"; +import type { AnyWorkflowDefinition, WorkflowDefinition } from "./types"; + +import { WORKFLOW_ERROR_CODES, WorkflowError } from "./errors"; + +/** + * A workflow reacting to a domain event. + * + * Kept out of the definition on purpose. `defineWorkflow({ trigger })` would + * bind one workflow to one way of starting it; the same `place-order` has to + * be reachable from a route, an event, a cron tick, the AdminCP and a test, + * and a definition that names its trigger can only ever be the first of those. + * + * This is an adapter over the existing event bus, not a second one: the + * trigger is turned into an ordinary `buildEventListener` at registration, so + * everything about delivery, ordering and adapters stays where it already is. + */ +export interface WorkflowEventTriggerDefinition { + description?: string; + event: VitNodeEventName; + /** + * Maps the event payload onto the workflow's input. Must be pure: it runs + * once, at start, and its result is what every attempt of every step sees. + */ + input: (payload: never, envelope: EventEnvelope) => unknown; + /** Listener name, unique within the registering module. */ + name: string; + /** Skip the event without starting anything. */ + when?: (payload: never, envelope: EventEnvelope) => boolean; + workflow: AnyWorkflowDefinition; +} + +/** + * A workflow started on a schedule. + * + * Registered as an ordinary `buildCron` job by the trigger adapter - there is + * no second scheduler. The job body does nothing but call + * `c.get("workflow").start(...)`, so the tick stays cheap and the work is + * durable from the first row. + */ +export interface WorkflowCronTriggerDefinition { + description?: string; + /** Built once per tick. Must be pure. */ + input?: (tick: Date) => unknown; + name: string; + /** Standard cron expression, handed straight to the configured cron adapter. */ + schedule: string; + workflow: AnyWorkflowDefinition; +} + +const assertTriggerName = (name: string, kind: "cron" | "event"): void => { + if (typeof name !== "string" || name.trim().length === 0) { + throw new WorkflowError( + WORKFLOW_ERROR_CODES.INVALID_TRIGGER, + `a workflow ${kind} trigger needs a non-empty \`name\`. It identifies the ${kind === "event" ? "listener" : "cron job"} in the AdminCP and in logs.`, + ); + } +}; + +/** + * ```ts + * buildWorkflowEventTrigger({ + * name: "start-place-order", + * workflow: placeOrderWorkflow, + * event: "order.created", + * input: payload => ({ orderId: payload.orderId }), + * }); + * ``` + * + * De-duplication is not optional here: the adapter starts the workflow with + * `idempotencyKey = event:{envelope.eventId}`, so a broker that delivers the + * same envelope twice produces one execution. + */ +export const buildWorkflowEventTrigger = < + const K extends VitNodeEventName, + TInputSchema extends z.ZodType, +>(args: { + description?: string; + event: K; + input: ( + payload: VitNodeEvents[K], + envelope: EventEnvelope, + ) => z.input; + name: string; + when?: (payload: VitNodeEvents[K], envelope: EventEnvelope) => boolean; + workflow: WorkflowDefinition; +}): WorkflowEventTriggerDefinition => { + assertTriggerName(args.name, "event"); + + // The per-event generic is erased so triggers for different events can share + // one array - the same trade-off `buildEventListener` makes. + return args as unknown as WorkflowEventTriggerDefinition; +}; + +/** + * ```ts + * buildWorkflowCronTrigger({ + * name: "nightly-reconciliation", + * schedule: "0 3 * * *", + * workflow: reconcileLedgerWorkflow, + * input: tick => ({ day: tick.toISOString().slice(0, 10) }), + * }); + * ``` + * + * The adapter starts the workflow with + * `idempotencyKey = cron:{name}:{tick}`, so a scheduler that fires the same + * minute twice produces one execution. + */ +export const buildWorkflowCronTrigger = (args: { + description?: string; + input?: (tick: Date) => z.input; + name: string; + schedule: string; + workflow: WorkflowDefinition; +}): WorkflowCronTriggerDefinition => { + assertTriggerName(args.name, "cron"); + + if (typeof args.schedule !== "string" || args.schedule.trim().length === 0) { + throw new WorkflowError( + WORKFLOW_ERROR_CODES.INVALID_TRIGGER, + `workflow cron trigger "${args.name}" needs a \`schedule\`.`, + ); + } + + return args; +}; diff --git a/packages/vitnode/src/api/workflows/types.ts b/packages/vitnode/src/api/workflows/types.ts new file mode 100644 index 000000000..29df4f58a --- /dev/null +++ b/packages/vitnode/src/api/workflows/types.ts @@ -0,0 +1,369 @@ +import type { Context } from "hono"; +import type { z } from "zod"; + +import type { EnvVitNode } from "../middlewares/global.middleware"; +import type { WORKFLOW_ACTOR_TYPES, WORKFLOW_TRIGGER_TYPES } from "./const"; +import type { WorkflowRetryPolicy, WorkflowRetryPolicyInput } from "./retry"; +import type { + WorkflowCompensationStatus, + WorkflowExecutionStatus, + WorkflowStepStatus, +} from "./state-machine"; + +export type WorkflowTriggerType = (typeof WORKFLOW_TRIGGER_TYPES)[number]; +export type WorkflowActorType = (typeof WORKFLOW_ACTOR_TYPES)[number]; + +/** + * A transaction handle, shaped exactly like `QueueModel.dispatch({ tx })`. + * + * Passing it is what makes "business row, execution, step rows and queue task + * commit together or not at all" true. Without it the execution row can commit + * while the row it is about rolls back, and the runner wakes up to orchestrate + * something that does not exist. + */ +export type WorkflowTransaction = Omit; + +/** + * Who asked for this workflow, recorded as metadata and nothing more. + * + * A queued step runs as system infrastructure: `c.get("admin")` and + * `c.get("user")` are null inside the runner's request and stay that way. The + * engine never reconstructs the original request's auth - a background job + * holding a forged session would make every permission check in the process + * lie about who is present. + */ +export interface WorkflowActor { + id?: null | number; + type: WorkflowActorType; +} + +/** + * What started the execution. Separate from the definition on purpose: one + * workflow is reachable from code, an event, a cron tick, the AdminCP, the API + * and a test, and none of those belong in the definition. + */ +export interface WorkflowTriggerRef { + /** + * The trigger's own identifier, when it has one worth keeping: + * `EventEnvelope.eventId` for an event, the tick key for a cron job. + */ + id?: null | string; + /** The event name, the cron job name, or a caller-chosen label. */ + name?: null | string; + type: WorkflowTriggerType; +} + +export interface WorkflowExecutionRef { + id: number; + pluginId: string; + workflowId: string; + workflowVersion: number; +} + +/** + * Reads the outputs of steps that already completed in this execution. + * + * `get` returns `unknown` because the value came back out of JSONB: after a + * restart the runner has a parsed row, not the object the previous step + * returned, and pretending otherwise would be a lie the type system cannot + * catch. `parse` is the supported way across that boundary. + */ +export interface WorkflowStepOutputs { + get: (stepId: string) => unknown; + has: (stepId: string) => boolean; + /** Validates a previous step's output, throwing a structured error if it no longer matches. */ + parse: (stepId: string, schema: z.ZodType) => TOutput; +} + +export interface WorkflowStepContext { + /** Metadata about who started the workflow. Never an authorization decision. */ + readonly actor: WorkflowActor; + /** 1 on the first run of this step, 2 on the first retry, and so on. */ + readonly attempt: number; + /** + * The runner's Hono context - use it for `c.get("db")`, `c.get("queue")`, + * `c.get("log")`. It is a background request: no user, no admin, no cookies. + */ + readonly c: Context; + readonly execution: WorkflowExecutionRef; + /** `workflow:{executionId}:{stepId}`. Stable across every attempt. */ + readonly idempotencyKey: string; + readonly input: TInput; + readonly outputs: WorkflowStepOutputs; + readonly step: { id: string; position: number }; + readonly trigger: WorkflowTriggerRef; +} + +export interface WorkflowCompensateContext extends Omit< + WorkflowStepContext, + "idempotencyKey" +> { + /** `workflow:{executionId}:{stepId}:compensate`. Never the step's own key. */ + readonly idempotencyKey: string; + /** What the step returned when it completed. Compensation only runs for completed steps. */ + readonly output: TOutput; +} + +export interface WorkflowStepDefinition { + /** + * Undo this step's side effect. + * + * Not a SQL rollback: the step already committed, and possibly charged a + * card. Compensation runs only for steps that *completed*, in reverse + * completion order, with its own retry budget and its own idempotency key. + */ + compensate?: ( + ctx: WorkflowCompensateContext, + ) => Promise | void; + description?: string; + id: string; + /** + * Validates what `run` returned before it is written to JSONB, and types + * `compensate`'s `output`. Omit it and the return value is stored as-is. + */ + output?: z.ZodType; + /** Business retry. Queue delivery retry is a separate, lower-level budget. */ + retry?: WorkflowRetryPolicyInput; + run: (ctx: WorkflowStepContext) => Promise | TOutput; +} + +/** + * A step as the registry stores it: the per-step output type is erased so one + * array can hold steps that return different things, and `retry` is resolved. + * + * The typed shape is checked at the `step({ ... })` call site, which is where + * the developer writes `run` and `compensate` - the same trade-off + * `buildEventListener` makes. + */ +export interface ResolvedWorkflowStep extends Omit< + WorkflowStepDefinition, + "retry" +> { + /** 0-based, assigned in declaration order and frozen at definition time. */ + position: number; + retry: WorkflowRetryPolicy; +} + +export interface WorkflowStepsBuilder { + step: ( + definition: WorkflowStepDefinition, + ) => WorkflowStepDefinition; +} + +export interface WorkflowDefinition< + TId extends string = string, + TInputSchema extends z.ZodType = z.ZodType, +> { + description?: string; + id: TId; + input: TInputSchema; + steps: ResolvedWorkflowStep>[]; + /** + * Bumped by hand whenever the *step list* changes meaning: a step added, + * removed, renamed or reordered. Executions store the version they started + * with and are only ever resolved against it. + */ + version: number; +} + +export type AnyWorkflowDefinition = WorkflowDefinition; + +/** A definition plus the plugin and module that registered it. */ +export interface RegisteredWorkflowDefinition { + definition: AnyWorkflowDefinition; + module: string; + pluginId: string; +} + +/** Identity of one definition. Never `workflowId` alone. */ +export interface WorkflowDefinitionRef { + pluginId: string; + version: number; + workflowId: string; +} + +export interface WorkflowStartOptions { + /** Defaults to the request's admin, then user, then `{ type: "system" }`. */ + actor?: WorkflowActor; + /** + * Collapses repeated starts into one execution, scoped by + * `pluginId + workflowId + workflowVersion`. + */ + idempotencyKey?: string; + trigger?: WorkflowTriggerRef; + /** Join the caller's transaction, exactly like `QueueModel.dispatch({ tx })`. */ + tx?: WorkflowTransaction; +} + +export interface WorkflowStartResult { + /** True when an execution with this idempotency key already existed. */ + deduplicated: boolean; + executionId: number; + status: WorkflowExecutionStatus; +} + +/** + * Everything `start()` decided before anything was written. + * + * Pure data, produced by `planWorkflowStart`. Splitting it from the write is + * what lets the SDK freeze *what* a start means while the persistence layer + * stays free to choose how the rows are inserted. + */ +export interface WorkflowStartPlan { + execution: { + actorId: null | number; + actorType: WorkflowActorType; + compensationStatus: WorkflowCompensationStatus; + idempotencyKey: null | string; + /** Parsed through the definition's schema, so it is exactly what steps receive. */ + input: Record; + module: string; + pluginId: string; + status: Extract; + triggerId: null | string; + triggerName: null | string; + triggerType: WorkflowTriggerType; + workflowId: string; + workflowVersion: number; + }; + /** + * The `@vitnode/core:workflow-step` task to dispatch inside the same + * transaction. `executionId` is only known after the insert, so the payload + * is completed by whoever writes the rows. + */ + queue: { + maxAttempts: number; + name: string; + pluginId: string; + stepId: string; + }; + steps: { + maxAttempts: number; + position: number; + status: Extract; + stepId: string; + }[]; +} + +export interface WorkflowExecutionRecord { + actorId: null | number; + actorType: WorkflowActorType; + cancellationRequestedAt: Date | null; + cancelledAt: Date | null; + compensationStatus: WorkflowCompensationStatus; + completedAt: Date | null; + createdAt: Date; + id: number; + idempotencyKey: null | string; + input: Record; + lastError: null | string; + module: string; + output: unknown; + pluginId: string; + startedAt: Date | null; + status: WorkflowExecutionStatus; + triggerId: null | string; + triggerName: null | string; + triggerType: WorkflowTriggerType; + updatedAt: Date; + workflowId: string; + workflowVersion: number; +} + +export interface WorkflowStepExecutionRecord { + attempts: number; + compensationAttempts: number; + compensationError: null | string; + compensationStatus: WorkflowCompensationStatus; + completedAt: Date | null; + executionId: number; + id: number; + lastError: null | string; + maxAttempts: number; + nextAttemptAt: Date | null; + output: unknown; + position: number; + startedAt: Date | null; + status: WorkflowStepStatus; + stepId: string; + updatedAt: Date; +} + +export interface WorkflowExecutionWithSteps { + execution: WorkflowExecutionRecord; + /** Ordered by `position`. */ + steps: WorkflowStepExecutionRecord[]; +} + +/** + * The persistence boundary between the SDK/runtime and the database. + * + * Frozen in Wave 0 so the runner can be written against it while the Drizzle + * implementation is written behind it. Adding a method is fine; changing one + * of these signatures is a contract change. + * + * Every method takes the Hono context rather than a handle, because the + * request's `c.get("db")` is the default handle and `tx` is the exception. + */ +export interface WorkflowStore { + /** + * Move a step from `pending` to `running` and count the attempt. + * + * Returns `undefined` when the row was not claimable - already running, + * already finished, or not due yet. That is the normal answer, not an error: + * the queue is at-least-once, so the same `workflow-step` task can be + * delivered twice and the second delivery has to be a no-op. + */ + claimStep: ( + c: Context, + args: { executionId: number; stepId: string }, + ) => Promise; + completeStep: ( + c: Context, + args: { executionId: number; output: unknown; stepId: string }, + ) => Promise; + /** + * Insert the execution, its step rows and the first `workflow-step` queue + * task in one unit of work. Returns the existing execution instead when the + * plan's idempotency key is already taken. + */ + createExecution: ( + c: Context, + plan: WorkflowStartPlan, + options?: { tx?: WorkflowTransaction }, + ) => Promise; + /** + * Record a failed attempt. `nextAttemptAt` comes from + * {@link nextWorkflowAttemptAt}: a date schedules a retry (`running -> + * pending`), `null` means the policy is exhausted (`running -> failed`). + */ + failStep: ( + c: Context, + args: { + error: string; + executionId: number; + nextAttemptAt: Date | null; + stepId: string; + }, + ) => Promise; + /** Terminal transition for the execution itself. */ + finishExecution: ( + c: Context, + args: { + executionId: number; + lastError?: null | string; + output?: unknown; + status: Exclude; + }, + ) => Promise; + loadExecution: ( + c: Context, + executionId: number, + ) => Promise; + /** Ask for cancellation. Never interrupts a step that is already running. */ + requestCancellation: (c: Context, executionId: number) => Promise; + /** Everything still `pending` becomes `skipped` - cancellation, or an earlier failure. */ + skipPendingSteps: (c: Context, executionId: number) => Promise; + /** `pending -> running`, stamping `startedAt` on the first step claim. */ + startExecution: (c: Context, executionId: number) => Promise; +} diff --git a/packages/vitnode/src/database/relations.ts b/packages/vitnode/src/database/relations.ts index 9a0648af2..cb5d15e69 100644 --- a/packages/vitnode/src/database/relations.ts +++ b/packages/vitnode/src/database/relations.ts @@ -14,6 +14,7 @@ import * as search from "./search"; import * as secrets from "./secrets"; import * as sessions from "./sessions"; import * as users from "./users"; +import * as workflows from "./workflows"; /** * Every table `@vitnode/core` ships, in one object. @@ -37,6 +38,7 @@ export const coreSchema = { ...secrets, ...sessions, ...users, + ...workflows, }; /** @@ -177,4 +179,15 @@ export const coreRelations = defineRelations(coreSchema, r => ({ to: r.core_users.id, }), }, + + core_workflow_executions: { + steps: r.many.core_workflow_step_executions(), + }, + + core_workflow_step_executions: { + execution: r.one.core_workflow_executions({ + from: r.core_workflow_step_executions.executionId, + to: r.core_workflow_executions.id, + }), + }, })); diff --git a/packages/vitnode/src/database/workflows.ts b/packages/vitnode/src/database/workflows.ts new file mode 100644 index 000000000..26235d0c3 --- /dev/null +++ b/packages/vitnode/src/database/workflows.ts @@ -0,0 +1,206 @@ +import { sql } from "drizzle-orm"; +import { camelCase, index, unique, uniqueIndex } from "drizzle-orm/pg-core"; + +import { + WORKFLOW_ACTOR_TYPES, + WORKFLOW_COMPENSATION_STATUSES, + WORKFLOW_EXECUTION_STATUSES, + WORKFLOW_STEP_STATUSES, + WORKFLOW_TRIGGER_TYPES, +} from "../api/workflows/const"; + +/** + * One durable run of one workflow. + * + * The row is the execution's memory: everything the runner needs to pick a + * queued step up - which plugin, which workflow, **which version**, what the + * input was - is here, because the worker that resolves it has no request, no + * plugin context and possibly no shared process with whoever started it. + * + * Workflow *definitions* are never stored. They are source code, imported by + * the plugin that owns them; only their identity and their runtime state live + * in Postgres. + */ +export const core_workflow_executions = camelCase.table.withRLS( + "core_workflow_executions", + t => ({ + id: t.serial().primaryKey(), + /** Owner of the definition, and the first third of its identity. */ + pluginId: t.varchar({ length: 100 }).notNull(), + /** The module that registered it - for the AdminCP, never for resolution. */ + module: t.varchar({ length: 100 }).notNull(), + workflowId: t.varchar({ length: 100 }).notNull(), + /** + * The version this execution started on, and the only one it will ever run. + * + * Deploying `place-order@2` does not move anything already in flight. If v1 + * is no longer registered the runner fails the execution with + * `WORKFLOW_DEFINITION_NOT_FOUND` and leaves every row alone, rather than + * running v2's steps against a plan that was made for v1. + */ + workflowVersion: t.integer().notNull(), + status: t + .varchar({ enum: WORKFLOW_EXECUTION_STATUSES, length: 20 }) + .notNull() + .default("pending"), + /** + * Rollback progress, tracked beside `status` rather than inside it. + * + * A combined vocabulary (`failed_compensating`, `failed_compensated`, ...) + * makes every "did this succeed" query enumerate compensation states it + * does not care about, and doubles in size each time a new one appears. + */ + compensationStatus: t + .varchar({ enum: WORKFLOW_COMPENSATION_STATUSES, length: 20 }) + .notNull() + .default("none"), + triggerType: t + .varchar({ enum: WORKFLOW_TRIGGER_TYPES, length: 20 }) + .notNull() + .default("manual"), + /** Event name, cron job name, or a caller-chosen label. */ + triggerName: t.varchar({ length: 255 }), + /** `EventEnvelope.eventId` for an event trigger, the tick key for cron. */ + triggerId: t.varchar({ length: 255 }), + /** + * Who asked for this - metadata, never authorization. + * + * No foreign key on purpose: the actor is a fact about the past, and it has + * to stay readable after the account is gone. Background steps run as + * system infrastructure regardless of what is recorded here. + */ + actorType: t + .varchar({ enum: WORKFLOW_ACTOR_TYPES, length: 20 }) + .notNull() + .default("system"), + actorId: t.integer(), + /** Genuinely schema-dynamic: one column serving every plugin's workflow. */ + input: t.jsonb().$type>().notNull().default({}), + /** The last step's output, once the execution completes. */ + output: t.jsonb().$type(), + /** + * Collapses repeated starts into one execution. + * + * Scoped to the definition, not global - see the partial unique index + * below. `event:{eventId}` for an event trigger, `cron:{name}:{tick}` for a + * cron one, whatever the caller passed otherwise. + */ + idempotencyKey: t.varchar({ length: 255 }), + lastError: t.text(), + /** + * When cancellation was *asked for*, which is not when it happened. + * + * The runner checks this between steps. A step already executing runs to + * completion - the engine cannot interrupt arbitrary JavaScript and does + * not claim to - and everything after it is skipped. + */ + cancellationRequestedAt: t.timestamp(), + createdAt: t.timestamp().notNull().defaultNow(), + startedAt: t.timestamp(), + completedAt: t.timestamp(), + cancelledAt: t.timestamp(), + updatedAt: t + .timestamp() + .notNull() + .defaultNow() + .$onUpdate(() => new Date()), + }), + t => [ + // At-least-once delivery means the same start can arrive twice; this is + // what makes the second one a no-op instead of a second execution. Partial, + // because an execution with no key is not de-duplicated at all and any + // number of them may exist. + uniqueIndex("core_workflow_executions_idempotency_unique") + .on(t.pluginId, t.workflowId, t.workflowVersion, t.idempotencyKey) + .where(sql`"idempotencyKey" is not null`), + // The AdminCP's list and the operator's "what is stuck" query. + index("core_workflow_executions_status_idx").on(t.status, t.createdAt), + // "Is anything still running on v1?" - the question a deploy has to answer + // before an old workflow version may be removed from the code. + index("core_workflow_executions_definition_idx").on( + t.pluginId, + t.workflowId, + t.workflowVersion, + t.status, + ), + ], +); + +/** + * One row per step of one execution, written up front when the execution is + * created rather than appearing as the workflow runs. + * + * Writing the whole plan at start is what makes the engine restart-safe: after + * a crash the runner reads state instead of re-deriving it, and an operator can + * see where a stuck execution stopped without replaying anything. + */ +export const core_workflow_step_executions = camelCase.table.withRLS( + "core_workflow_step_executions", + t => ({ + id: t.serial().primaryKey(), + executionId: t + .integer() + .notNull() + .references(() => core_workflow_executions.id, { + onDelete: "cascade", + onUpdate: "cascade", + }), + /** The step's id in the definition. Half of the execution-scoped identity. */ + stepId: t.varchar({ length: 100 }).notNull(), + /** 0-based declaration order, frozen when the execution was planned. */ + position: t.integer().notNull(), + status: t + .varchar({ enum: WORKFLOW_STEP_STATUSES, length: 20 }) + .notNull() + .default("pending"), + /** Runs of the step body so far, including the one in flight. */ + attempts: t.integer().notNull().default(0), + /** Copied from the step's resolved retry policy when the execution is planned. */ + maxAttempts: t.integer().notNull().default(3), + output: t.jsonb().$type(), + lastError: t.text(), + /** Set when a failed attempt is retryable; `null` once the policy is spent. */ + nextAttemptAt: t.timestamp(), + /** + * Compensation is tracked per step because it is resumable per step: a + * crash halfway through a rollback has to continue where it stopped, in + * reverse completion order, without undoing anything twice. + */ + compensationStatus: t + .varchar({ enum: WORKFLOW_COMPENSATION_STATUSES, length: 20 }) + .notNull() + .default("none"), + compensationAttempts: t.integer().notNull().default(0), + compensationError: t.text(), + startedAt: t.timestamp(), + completedAt: t.timestamp(), + updatedAt: t + .timestamp() + .notNull() + .defaultNow() + .$onUpdate(() => new Date()), + }), + t => [ + // The invariant the whole runner rests on: one row per step per execution, + // so "has this step already run" is a primary-key question and a duplicated + // queue delivery cannot create a second attempt row. + unique("core_workflow_step_executions_step_unique").on( + t.executionId, + t.stepId, + ), + // The runner's read: this execution's plan, in order. + index("core_workflow_step_executions_order_idx").on( + t.executionId, + t.position, + ), + // Retries that have come due. + index("core_workflow_step_executions_next_attempt_idx").on( + t.status, + t.nextAttemptAt, + ), + ], +); + +export type WorkflowExecutionRow = typeof core_workflow_executions.$inferSelect; +export type WorkflowStepExecutionRow = + typeof core_workflow_step_executions.$inferSelect; diff --git a/packages/vitnode/src/lib/api/resolve-stale-queue-lease.test.ts b/packages/vitnode/src/lib/api/resolve-stale-queue-lease.test.ts new file mode 100644 index 000000000..78c21ecaf --- /dev/null +++ b/packages/vitnode/src/lib/api/resolve-stale-queue-lease.test.ts @@ -0,0 +1,65 @@ +// @vitest-environment node +import { describe, expect, it } from "vitest"; + +import { + QUEUE_LEASE_EXPIRED_ERROR, + QUEUE_LEASE_TIMEOUT_MS, + queueLeaseCutoff, + resolveStaleQueueLease, +} from "./resolve-stale-queue-lease"; + +describe("queueLeaseCutoff", () => { + const now = new Date("2026-01-01T12:00:00.000Z"); + + it("looks one lease window back", () => { + expect(queueLeaseCutoff(now).toISOString()).toBe( + "2026-01-01T11:45:00.000Z", + ); + }); + + it("leaves a reservation younger than the lease alone", () => { + const reservedAt = new Date(now.getTime() - QUEUE_LEASE_TIMEOUT_MS + 1_000); + + expect(reservedAt.getTime()).toBeGreaterThan( + queueLeaseCutoff(now).getTime(), + ); + }); + + it("reclaims a reservation older than the lease", () => { + const reservedAt = new Date(now.getTime() - QUEUE_LEASE_TIMEOUT_MS - 1_000); + + expect(reservedAt.getTime()).toBeLessThan(queueLeaseCutoff(now).getTime()); + }); +}); + +describe("resolveStaleQueueLease", () => { + const now = new Date("2026-01-01T12:00:00.000Z"); + + it("puts a task with attempts left back on the queue immediately", () => { + expect( + resolveStaleQueueLease({ attempts: 1, maxAttempts: 3, now }), + ).toEqual({ + availableAt: now, + lastError: QUEUE_LEASE_EXPIRED_ERROR, + status: "pending", + }); + }); + + it("fails a task whose crashed run spent its last attempt", () => { + // The attempt was counted when the task was claimed, so a handler that + // reliably kills its worker cannot cycle forever. + expect( + resolveStaleQueueLease({ attempts: 3, maxAttempts: 3, now }), + ).toEqual({ + completedAt: now, + lastError: QUEUE_LEASE_EXPIRED_ERROR, + status: "failed", + }); + }); + + it("always says why, so an operator does not read an empty lastError", () => { + expect( + resolveStaleQueueLease({ attempts: 1, maxAttempts: 3, now }).lastError, + ).toContain("lease expired"); + }); +}); diff --git a/packages/vitnode/src/lib/api/resolve-stale-queue-lease.ts b/packages/vitnode/src/lib/api/resolve-stale-queue-lease.ts new file mode 100644 index 000000000..0b5fb50fd --- /dev/null +++ b/packages/vitnode/src/lib/api/resolve-stale-queue-lease.ts @@ -0,0 +1,63 @@ +import type { QueueTaskOutcome } from "./resolve-queue-task-outcome"; + +/** + * How long a claimed task may stay `processing` before another tick may take it + * back. + * + * The queue's claim is a lease, not a lock: `processQueueTasks` flips rows to + * `processing` and stamps `reservedAt`, then runs the handlers. If the process + * dies in between - a deploy, an OOM kill, a container reschedule - nothing + * ever writes the finishing update, and the row is invisible to every later + * tick, which only ever selects `pending`. Without recovery that task is lost + * permanently. + * + * Fifteen minutes rather than the cron's own minute: a batch of 25 tasks can + * legitimately take several minutes, and reclaiming a task that is still + * running would execute it twice for no reason. Long enough to be safe, short + * enough that a crashed worker's tasks resume within one deploy cycle. + */ +export const QUEUE_LEASE_TIMEOUT_MS = 15 * 60 * 1000; + +export const queueLeaseCutoff = ( + now: Date = new Date(), + leaseMs: number = QUEUE_LEASE_TIMEOUT_MS, +): Date => new Date(now.getTime() - leaseMs); + +export const QUEUE_LEASE_EXPIRED_ERROR = + "Worker stopped before the task finished (reservation lease expired)."; + +/** + * What to do with a task whose lease expired. + * + * The attempt was already counted when the task was claimed, so a crashed run + * has spent one - which is what stops a handler that reliably kills its process + * from being retried forever. With attempts left the task goes back to + * `pending` and is available immediately; with none it is `failed`, carrying an + * error that says what happened rather than an empty `lastError` an operator + * has to guess at. + * + * Generic queue behaviour, not workflow behaviour: any task benefits, and the + * Workflow Engine depends on it because a step that is stuck in `processing` + * forever is an execution that never advances and never fails - the one state a + * durable engine must not have. + */ +export const resolveStaleQueueLease = ({ + attempts, + maxAttempts, + now = new Date(), +}: { + attempts: number; + maxAttempts: number; + now?: Date; +}): QueueTaskOutcome => + attempts < maxAttempts + ? { + availableAt: now, + lastError: QUEUE_LEASE_EXPIRED_ERROR, + status: "pending", + } + : { + completedAt: now, + lastError: QUEUE_LEASE_EXPIRED_ERROR, + status: "failed", + };