diff --git a/.agents/rules/workflow-determinism.md b/.agents/rules/workflow-determinism.md index 7077848c..0e9f1d00 100644 --- a/.agents/rules/workflow-determinism.md +++ b/.agents/rules/workflow-determinism.md @@ -1,22 +1,37 @@ # Workflow Determinism -Workflow code runs inside Temporal's deterministic replay sandbox. Every time a workflow is rehydrated (worker restart, sticky-task reassignment, history replay), Temporal re-runs the implementation from the start and **must produce the exact same sequence of commands**. Any non-determinism — wall-clock reads, native randomness, direct I/O — will desync from history and crash the workflow with a non-determinism error. +Workflow code runs inside Temporal's deterministic replay sandbox. Every time a workflow is rehydrated (worker restart, sticky-task reassignment, history replay), Temporal re-runs the implementation from the start and **must produce the exact same sequence of commands**. A desync from history crashes the workflow with a non-determinism error. -This is THE most error-prone area in any Temporal codebase. Read it. +**The sandbox does more than people assume.** `@temporalio/workflow/lib/global-overrides.js` rewrites the common hazards before your code runs, and hard-blocks a couple more. So the guidance below is mostly about **semantics, not safety** — the patched APIs work, they just don't mean what their names suggest. Knowing which column a thing is in tells you whether a mistake shows up as a crash, a surprise, or silent corruption. -## Banned in workflow code +### Patched — safe, but the semantics differ -| Don't | Use instead | -| ---------------------------------- | ----------------------------------------------------------------------------------------------------- | -| `Date.now()` / `new Date()` | `workflowInfo().startTime` or `Date` from `@temporalio/workflow` (it's monkey-patched in the sandbox) | -| `Math.random()` | `uuid4()` from `@temporalio/workflow`, or do RNG inside an activity | -| `crypto.randomUUID()` / `crypto.*` | `uuid4()` from `@temporalio/workflow`, or activity | -| `setTimeout` / `setInterval` | `sleep(duration)` from `@temporalio/workflow` | -| `process.env.*` | Pass via `args` or read inside an activity | -| `fetch` / `http` / database / disk | Wrap in an activity — workflows must not touch I/O | -| `import.meta.*` / `__dirname` | Constant inputs; or read inside an activity | +These are rewritten in the sandbox. Using them will not break replay. Prefer the explicit primitive anyway, because the behavior is not what the name implies. -The rule of thumb: **if it can return a different value on a second call with the same inputs, it doesn't belong in workflow code.** Push it into an activity, where retries and non-determinism are explicitly handled. +| API | What it actually does | Prefer | +| ---------------------------- | ---------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------- | +| `Date.now()` / `new Date()` | Returns **workflow time** (`getActivator().now`), not wall clock. On replay it returns the historical value. | `workflowInfo().startTime` when you mean "when did this start" | +| `Math.random()` | A **seeded deterministic stream**. Replay-safe, but the sequence shifts when you change consuming code. | `uuid4()`, or do RNG in an activity | +| `setTimeout` / `setInterval` | Becomes a **durable Temporal timer** wrapped in a cancellation scope. Survives worker restarts; advances with workflow time. | `sleep(duration)` — same thing, honest name | + +### Blocked loudly — you cannot get this wrong silently + +| API | What happens | +| ------------------------------------ | ------------------------------------------------------------------------------- | +| `WeakRef` / `FinalizationRegistry` | Throws `DeterminismViolationError` — "v8 GC is non-deterministic" | +| `crypto.*`, `fetch`, `fs`, net, disk | Not present in the sandbox context; reference errors rather than nondeterminism | + +### Genuinely unprotected — this is where the real risk lives + +Nothing patches these. They are the reason this rule exists. + +| Hazard | Why it bites | Do instead | +| ----------------------------- | ------------------------------------------------------------------------------------------------------- | --------------------------------------- | +| `process.env.*` | Legitimately differs per worker and per deployment. Two workers replaying the same history can diverge. | Pass via `args`, or read in an activity | +| Module-level mutable state | Shared across executions in a reused VM context; ordering is not yours to control. | Keep workflow modules pure | +| `import.meta.*` / `__dirname` | Environment-dependent, and not something history records. | Constant inputs | + +The rule of thumb still holds — **if it can return a different value on a second call with the same inputs, it doesn't belong in workflow code** — but note the sandbox already enforces most of it. Spend your vigilance on the third table. ## Why activities can do anything @@ -30,7 +45,9 @@ Use `context.cancellableScope` / `context.nonCancellableScope` (`packages/worker ## Side-effect escape hatch -If you absolutely need non-determinism inside workflow code (e.g. logging at a checkpoint), use `LocalActivity` with `proxyLocalActivities` from `@temporalio/workflow` — same sandboxing rules but lower overhead than a full network round-trip. Even logging via `console.log` is fine in workflow code (`@temporalio/workflow` patches it through Temporal's logger), but `console.log({ now: Date.now() })` is not — the _value_ is non-deterministic. +If you need something the sandbox genuinely cannot provide — reading `process.env`, hitting a real clock, calling out — use `LocalActivity` with `proxyLocalActivities` from `@temporalio/workflow`: it runs outside the sandbox like a normal activity, but with lower overhead than a full network round-trip. + +Logging via `console.log` is fine in workflow code (`@temporalio/workflow` patches it through Temporal's logger). `console.log({ now: Date.now() })` is fine too — per the first table, that value is workflow time and is stable across replays. What is _not_ fine is `console.log({ env: process.env.REGION })`, because that value can differ between the worker that ran the workflow and the worker that replays it. ## Canonical examples diff --git a/.changeset/testing-test-rig-export.md b/.changeset/testing-test-rig-export.md new file mode 100644 index 00000000..616bc0ae --- /dev/null +++ b/.changeset/testing-test-rig-export.md @@ -0,0 +1,5 @@ +--- +"@temporal-contract/testing": minor +--- + +New `@temporal-contract/testing/test-rig` subpath: `testRig` builds the `TypedWorker` + `TypedClient` pair an in-process test needs against a `TestWorkflowEnvironment`, and registers an `onTestFinished` hook that replays every execution the client started — real replay-determinism coverage per test, with no separate replay pass to remember. Options include `replaySkipAllowlist`, a caller-supplied map of workflow-ID prefixes (with reasons) to skip when their execution is deliberately left non-terminal; it defaults to `{}` so a published rig never bakes in this repo's own fixture IDs. Also exports `isTerminalStatus`, `skipReasonFor`, and `extractStartedWorkflowId` for unit-testing the rig's own assumptions. diff --git a/AGENTS.md b/AGENTS.md index 990e91a3..7758d3b2 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -6,7 +6,7 @@ This file is the source of truth for agent guidance in this repo. `CLAUDE.md` an ## The 6 rules that prevent broken PRs -1. **Workflow code is deterministic.** No `Date.now()`, `Math.random()`, `setTimeout`, `crypto.randomUUID()`, native I/O, or `process.env` reads inside `declareWorkflow`'s `implementation`. Use `@temporalio/workflow` primitives (`sleep`, `uuid4`, the patched `Date`) or push the side effect into an activity. See [.agents/rules/workflow-determinism.md](.agents/rules/workflow-determinism.md). This is the #1 cause of broken Temporal workflows — read that file before touching workflow code. +1. **Workflow code is deterministic — and the sandbox enforces most of it for you.** `Date.now()`, `new Date()`, `Math.random()`, and `setTimeout` are **patched** in the sandbox and are replay-safe; `WeakRef`/`FinalizationRegistry` throw `DeterminismViolationError`; `crypto.*`, `fetch`, and native I/O aren't in the sandbox context at all. Prefer `@temporalio/workflow`'s `sleep` / `uuid4` / `workflowInfo()` anyway, because the patched APIs don't mean what their names suggest (`Date.now()` is _workflow_ time; `setTimeout` is a _durable timer_). The genuinely unprotected hazards inside `declareWorkflow`'s `implementation` are **`process.env` reads, module-level mutable state, and `import.meta`** — spend your vigilance there. See [.agents/rules/workflow-determinism.md](.agents/rules/workflow-determinism.md) before touching workflow code. 2. **Activities and the typed client return `AsyncResult` from `unthrown`.** Never throw — wrap technical errors in `ApplicationFailure` and surface them via `ErrAsync(...)` (or `fromPromise(promise, qualify)`, where `qualify` returns the modeled error `E`). `OkAsync(value)`/`ErrAsync(error)` are the canonical pre-lifted constructors (there is no lowercase `okAsync`/`errAsync`); lifting an existing sync `Result` with `.toAsync()` stays valid but is not used for direct construction in this codebase — prefer `OkAsync()` zero-arg over `OkAsync(undefined)`. The client uses unthrown's `Result` for sync returns. unthrown adds a third **`defect`** channel for _unanticipated_ failures — a thrown exception the code didn't model surfaces as a defect (inspectable via `result.isDefect()` / `result.cause`, re-thrown at the edge), not a typed `err`. Narrow before reaching `.value`/`.error`/`.cause` — both the `r.isOk()` method and the `isOk(r)` free function are type guards (same for `isErr`/`isDefect`); the codebase uses the methods. Error classes are built with `TaggedError("@temporal-contract/Name", { name: "Name" })<{ ...payload }>` — the `_tag` is package-namespaced to avoid collisions, while `options.name` keeps `Error.name` the bare class name for readable logs. The worker's `ValidationError` subclasses are the exception — they must stay `ApplicationFailure` for Temporal's terminal-failure semantics. There is no `neverthrow`, no `@swan-io/boxed`, and no `@temporal-contract/boxed` package — those were removed. 3. **No `any`.** Use `unknown` and narrow. Enforced by oxlint. 4. **`.js` extensions in every import.** TypeScript files import each other as `./foo.js`, never `./foo` or `./foo.ts`. Required by ESM module resolution. diff --git a/docs/superpowers/plans/2026-08-02-determinism-invariants.md b/docs/superpowers/plans/2026-08-02-determinism-invariants.md new file mode 100644 index 00000000..00945b08 --- /dev/null +++ b/docs/superpowers/plans/2026-08-02-determinism-invariants.md @@ -0,0 +1,913 @@ +# Determinism and Money-Safety Invariants Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Prove by effect the invariants the library could silently break — retry terminality, timeout forwarding, and replay determinism across every sandboxed construct. + +**Architecture:** A `testRig` helper replaces the worker+client construction already present in every in-process test. Its client records started workflow IDs; an `onTestFinished` hook fetches each execution's history and replays it against the bundle the worker was built from. Replay coverage therefore becomes automatic for present and future tests. Retry and timeout invariants are proven by reading what Temporal actually materialized, not by reading fields off the wire. + +**Tech Stack:** TypeScript (ESM), Vitest 4.1.10, `@temporalio/worker` 1.20.3, `@temporalio/activity`, unthrown 5. + +## Global Constraints + +Copied from `CLAUDE.md` and the spec. Every task's requirements implicitly include this section. + +- **Workflow code is deterministic — the sandbox enforces most of it.** `Date.now()`, `new Date()`, `Math.random()`, and `setTimeout` are patched and replay-safe; prefer `sleep` / `uuid4` / `workflowInfo()` from `@temporalio/workflow` because the patched APIs don't mean what their names suggest. `WeakRef`/`FinalizationRegistry` throw; `crypto.*` and native I/O aren't in the sandbox context. The genuinely unprotected hazards are **`process.env`, module-level mutable state, and `import.meta`** — those must stay out of `declareWorkflow`'s `implementation`. Activities are not sandboxed and may use anything. +- **`.js` extensions in every relative import.** ESM only. No CommonJS. +- **No `any`.** Use `unknown` and narrow. Enforced by oxlint. +- **Catalog versions.** New dependency versions go in `pnpm-workspace.yaml`'s `catalog:` block; per-package entries use `"catalog:"`. +- **Activities return `AsyncResult`** — `OkAsync` / `ErrAsync`, narrowed with `.isOk()` / `.isErr()` / `.isDefect()`. +- **Assert effects, never call shapes.** Specific outcomes with `toBe` / `toEqual`, never a substring any failure would satisfy. Any assert-empty needs a positive control. +- **A regression must fail an assertion, not hang** to the 120s timeout. Pass `workflowExecutionTimeout: "30 seconds"` in start option bags; fold failure detail into returned status strings rather than rethrowing. +- **Bound activity retries explicitly** (`retry: { maximumAttempts: 1 }`) unless the test is specifically about retry. +- **A real `DeterminismViolationError` is a finding, not an obstacle.** If replay surfaces one in shipped code, STOP and report it. Do not adjust the test to pass. + +## Key API Facts (verified against the installed SDK — do not re-derive) + +- `Worker.runReplayHistory(options: ReplayWorkerOptions, history: History | unknown, workflowId?: string): Promise` — from `@temporalio/worker`. `ReplayWorkerOptions` is `Omit`, so **`workflowBundle` and `workflowsPath` both pass through**. It needs no server. +- History is fetched with `testEnv.client.workflow.getHandle(workflowId).fetchHistory()`. +- `handle.describe()` returns `{ status: { name: WorkflowExecutionStatusName } }` where the union is `'UNSPECIFIED' | 'RUNNING' | 'COMPLETED' | 'FAILED' | 'CANCELLED' | 'TERMINATED' | 'CONTINUED_AS_NEW' | 'TIMED_OUT' | 'PAUSED' | 'UNKNOWN'`. **Note the double-L `CANCELLED`.** +- `onTestFinished(fn)` is exported from `vitest` and **may be called from inside a test body**. This is what lets `testRig` register its own teardown. +- Activity `Context.current().info` exposes `attempt: number`, `scheduleToCloseTimeoutMs: number`, `startToCloseTimeoutMs: number`, and `heartbeatTimeoutMs?: number` (optional — only present when declared). +- `ContractClient` has a **private constructor**; wrap it with a `Proxy`, do not subclass. +- The three start methods are `startWorkflow`, `executeWorkflow`, and `signalWithStart`. + +--- + +### Task 1: The `testRig` replay-harvest helper + +**Files:** + +- Create: `packages/testing/src/test-rig.ts` +- Create: `packages/testing/src/test-rig.spec.ts` +- Modify: `packages/testing/package.json` (add `src/test-rig.ts` to the `build` script's tsdown entry list, and a `./test-rig` subpath export) +- Modify: `packages/testing/typedoc.json` (add the entry point) + +**Interfaces:** + +- Consumes: `bundleFor` / `withTaskQueue` / `nextTaskQueueId` from `@temporal-contract/testing/workflow-bundle` (callers still use these; the rig does not). +- Produces: + - `testRig(testEnv, options): Promise<{ worker: TypedWorker; client: ContractClient }>` where `options` is `{ contract, bundle, activities? }`. + - `REPLAY_SKIP_ALLOWLIST: Record` — workflow-ID **prefix** → reason. + - `isTerminalStatus(name: string): boolean` — exported for its own unit test. + - `skipReasonFor(workflowId: string, allowlist: Record): string | undefined` — + exported so the allowlist-matching rule is unit-testable without a server. + +- [ ] **Step 1: Write the failing unit tests for the pure parts** + +Create `packages/testing/src/test-rig.spec.ts`: + +```ts +import { describe, expect, it } from "vitest"; + +import { isTerminalStatus, REPLAY_SKIP_ALLOWLIST, skipReasonFor } from "./test-rig.js"; + +describe("isTerminalStatus", () => { + it("treats every finished status as terminal", () => { + for (const name of ["COMPLETED", "FAILED", "CANCELLED", "TERMINATED", "TIMED_OUT"]) { + expect(isTerminalStatus(name)).toBe(true); + } + }); + + it("treats CONTINUED_AS_NEW as terminal — that run's history is complete and replayable", () => { + expect(isTerminalStatus("CONTINUED_AS_NEW")).toBe(true); + }); + + it("treats unfinished statuses as non-terminal", () => { + for (const name of ["RUNNING", "PAUSED", "UNSPECIFIED", "UNKNOWN"]) { + expect(isTerminalStatus(name)).toBe(false); + } + }); +}); + +describe("skipReasonFor", () => { + it("matches an allowlist entry by workflow-ID prefix", () => { + expect(skipReasonFor("probe-edge-cases-1", { "probe-edge-cases": "blocks forever" })).toBe( + "blocks forever", + ); + }); + + it("returns undefined for an unlisted id, so the caller can fail", () => { + expect( + skipReasonFor("some-other-id", { "probe-edge-cases": "blocks forever" }), + ).toBeUndefined(); + }); + + it("ships an allowlist whose every entry carries a non-empty reason", () => { + for (const [prefix, reason] of Object.entries(REPLAY_SKIP_ALLOWLIST)) { + expect(reason, `allowlist entry "${prefix}" needs a reason`).not.toBe(""); + } + }); +}); +``` + +- [ ] **Step 2: Run to verify it fails** + +Run: `cd packages/testing && pnpm vitest run src/test-rig.spec.ts` +Expected: FAIL — `Cannot find module './test-rig.js'` + +- [ ] **Step 3: Write the implementation** + +Create `packages/testing/src/test-rig.ts`: + +```ts +import type { ContractClient } from "@temporal-contract/client"; +import { TypedClient } from "@temporal-contract/client"; +import type { ContractDefinition } from "@temporal-contract/contract"; +// `ActivitiesHandler` lives on the /activity subpath — worker.ts imports it +// but does not re-export it. `TypedWorker` is both a type and a value, so one +// non-type import covers both uses. +import type { ActivitiesHandler } from "@temporal-contract/worker/activity"; +import { TypedWorker } from "@temporal-contract/worker/worker"; +import type { TestWorkflowEnvironment } from "@temporalio/testing"; +import { Worker } from "@temporalio/worker"; +import type { WorkflowBundleWithSourceMap } from "@temporalio/worker"; +import { onTestFinished } from "vitest"; + +/** + * Workflow-ID prefixes whose executions are deliberately left non-terminal, so + * their histories cannot be replayed. Every entry needs a reason. + * + * This list may only ever shrink. A silently-skipped execution would report + * replay coverage it does not have — exactly the rot this rig exists to + * prevent — so an unlisted non-terminal execution fails the test instead. + */ +export const REPLAY_SKIP_ALLOWLIST: Record = {}; + +/** Statuses whose history is complete and therefore replayable. */ +const TERMINAL_STATUSES = new Set([ + "COMPLETED", + "FAILED", + "CANCELLED", + "TERMINATED", + "TIMED_OUT", + // The run ended; the next run is a separate execution with its own history. + "CONTINUED_AS_NEW", +]); + +export function isTerminalStatus(name: string): boolean { + return TERMINAL_STATUSES.has(name); +} + +export function skipReasonFor( + workflowId: string, + allowlist: Record, +): string | undefined { + for (const [prefix, reason] of Object.entries(allowlist)) { + if (workflowId.startsWith(prefix)) return reason; + } + return undefined; +} + +/** The three `ContractClient` methods that can start an execution. */ +const START_METHODS = new Set(["startWorkflow", "executeWorkflow", "signalWithStart"]); + +type RigOptions = { + readonly contract: TContract; + readonly bundle: WorkflowBundleWithSourceMap; + readonly activities?: ActivitiesHandler; +}; + +/** + * Build the worker + client pair every in-process test needs, and register an + * `onTestFinished` hook that replays the history of every execution the client + * started. + * + * The rig deliberately does NOT scope the task queue — callers keep calling + * `withTaskQueue` themselves. A same-workflow continue-as-new must land on the + * contract's static queue, because the contract is closed over inside the + * bundled workflow module and a test-side copy can never reach it. + */ +export async function testRig( + testEnv: TestWorkflowEnvironment, + options: RigOptions, +): Promise<{ worker: TypedWorker; client: ContractClient }> { + const { contract, bundle, activities } = options; + + const worker = await TypedWorker.create({ + contract, + connection: testEnv.nativeConnection, + workflowBundle: bundle, + // Spread conditionally: `TypedWorker.create` distinguishes an absent + // `activities` key from `activities: undefined` (a workflow-only worker + // must not register an activity poller). + ...(activities !== undefined ? { activities } : {}), + }).get(); + + const typedClient = await TypedClient.create({ client: testEnv.client }).get(); + const bound = typedClient.for(contract); + + const startedIds: string[] = []; + + const client = new Proxy(bound, { + get(target, property, receiver) { + const value = Reflect.get(target, property, receiver) as unknown; + if (typeof property !== "string" || !START_METHODS.has(property)) return value; + if (typeof value !== "function") return value; + + return (...args: readonly unknown[]) => { + const bag = args[1]; + const workflowId = + typeof bag === "object" && bag !== null && "workflowId" in bag + ? (bag as { workflowId?: unknown }).workflowId + : undefined; + if (typeof workflowId === "string") startedIds.push(workflowId); + return (value as (...rest: readonly unknown[]) => unknown).apply(target, args); + }; + }, + }) as ContractClient; + + onTestFinished(async () => { + for (const workflowId of startedIds) { + const handle = testEnv.client.workflow.getHandle(workflowId); + const described = await handle.describe(); + + if (!isTerminalStatus(described.status.name)) { + const reason = skipReasonFor(workflowId, REPLAY_SKIP_ALLOWLIST); + if (reason === undefined) { + throw new Error( + `Workflow "${workflowId}" ended ${described.status.name}, so its history cannot be ` + + `replayed and this test proves nothing about replay determinism for it. Either make ` + + `the execution terminal, or add a REPLAY_SKIP_ALLOWLIST entry in ` + + `packages/testing/src/test-rig.ts with a reason.`, + ); + } + continue; + } + + const history = await handle.fetchHistory(); + await Worker.runReplayHistory({ workflowBundle: bundle }, history, workflowId); + } + }); + + return { worker, client }; +} +``` + +- [ ] **Step 4: Run the unit tests to verify they pass** + +Run: `cd packages/testing && pnpm vitest run src/test-rig.spec.ts` +Expected: PASS (6 tests) + +- [ ] **Step 5: Wire the subpath export** + +In `packages/testing/package.json`, append `src/test-rig.ts` to the `build` script's tsdown entry list, and add to `"exports"` (alphabetical, matching the existing block): + +```json + "./test-rig": { + "types": "./dist/test-rig.d.mts", + "import": "./dist/test-rig.mjs" + }, +``` + +In `packages/testing/typedoc.json`, add `"src/test-rig.ts"` to `entryPoints`. + +- [ ] **Step 6: Verify the package builds and typechecks** + +Run: `pnpm turbo run build typecheck lint --filter=@temporal-contract/testing` +Expected: all pass; `packages/testing/dist/test-rig.mjs` exists. + +- [ ] **Step 7: Commit** + +```bash +git add packages/testing/src/test-rig.ts packages/testing/src/test-rig.spec.ts \ + packages/testing/package.json packages/testing/typedoc.json +git commit -m "test(testing): add the testRig replay-harvest helper" +``` + +--- + +### Task 2: Prove the rig on `replay.inprocess.spec.ts` + +This file already replays manually, so it is the one place with a known-good expectation to check the rig against. + +**Files:** + +- Modify: `packages/worker/src/__tests__/replay.inprocess.spec.ts` + +**Interfaces:** + +- Consumes: `testRig` from `@temporal-contract/testing/test-rig` (Task 1). +- Produces: nothing later tasks depend on. + +- [ ] **Step 1: Replace the manual construction and manual replay loop** + +Delete the trailing `for (const workflowId of [happyId, declinedId]) { … runReplayHistory … }` block and the `TypedWorker.create` / `TypedClient.create` lines, replacing them with: + +```ts +const { worker, client } = await testRig(testEnv, { + contract: inprocessContract, + bundle: await bundleFor(fixturePath(import.meta.url, "inprocess.workflows")), + activities, +}); +``` + +Keep every `expect` in the test body unchanged. + +- [ ] **Step 2: Run and confirm still green** + +Run: `cd packages/worker && pnpm vitest run --project integration-inprocess src/__tests__/replay.inprocess.spec.ts` +Expected: PASS. The replay now happens in the rig's teardown rather than inline. + +- [ ] **Step 3: Prove the rig's replay actually runs** + +The danger is a teardown that silently does nothing. Temporarily break it: in `test-rig.ts`, change `Worker.runReplayHistory({ workflowBundle: bundle }, history, workflowId)` to replay against a **different** workflow module — e.g. `{ workflowsPath: fixturePath(import.meta.url, "handlers.workflows") }` hardcoded — and confirm the test now FAILS with a replay error. Restore afterwards. + +Quote the verbatim failure in your report. A teardown never observed failing is not known to run. + +- [ ] **Step 4: Commit** + +```bash +git add packages/worker/src/__tests__/replay.inprocess.spec.ts +git commit -m "test(worker): move replay.inprocess.spec.ts onto the rig" +``` + +--- + +### Task 3: Migrate `handlers` and `activity-options` to the rig + +The two largest suites, covering signals, queries, updates, and the activity-options merge layers. + +**Files:** + +- Modify: `packages/worker/src/__tests__/handlers.inprocess.spec.ts` +- Modify: `packages/worker/src/__tests__/activity-options.inprocess.spec.ts` + +**Interfaces:** + +- Consumes: `testRig` (Task 1). +- Produces: nothing later tasks depend on. + +- [ ] **Step 1: Rewrite each test's construction** + +In every test, replace this pair: + +```ts +const worker = await TypedWorker.create({ + contract, + connection: testEnv.nativeConnection, + workflowBundle: bundle, + activities, +}).get(); +const typedClient = await TypedClient.create({ client: testEnv.client }).get(); +const client = typedClient.for(contract); +``` + +with: + +```ts +const { worker, client } = await testRig(testEnv, { contract, bundle, activities }); +``` + +Leave `withTaskQueue` / `nextTaskQueueId` / `bundleFor` calls exactly as they are — the rig does not scope the queue. + +For workflow-only tests (no `activities` variable in scope), omit the key entirely rather than passing `activities: undefined`. + +**Do not change any assertion.** This task is mechanical; a changed expectation is a defect. + +- [ ] **Step 2: Run and triage** + +Run: `cd packages/worker && pnpm vitest run --project integration-inprocess src/__tests__/handlers.inprocess.spec.ts src/__tests__/activity-options.inprocess.spec.ts` + +Three outcomes, and they must be distinguished in your report: + +1. **PASS** — the rig replayed every history cleanly. +2. **FAIL: non-terminal execution** — the rig's own error message names the workflow ID. `handlers.workflows.ts`'s `probeEdgeCases` and `transformWorkflow` block on `condition(() => false)` and are expected here. Add a `REPLAY_SKIP_ALLOWLIST` entry keyed by the workflow-ID prefix with a reason naming why it never terminates. +3. **FAIL: `DeterminismViolationError` or `ReplayError`** — **STOP.** This is a real determinism bug in shipped code. Report it with the workflow, the history, and the error. Do not allowlist it, do not adjust the test. + +- [ ] **Step 3: Run the whole in-process tier to catch cross-file effects** + +Run: `cd packages/worker && pnpm vitest run --project integration-inprocess` +Expected: PASS, with the same test count as before the migration. + +- [ ] **Step 4: Commit** + +```bash +git add packages/worker/src/__tests__/handlers.inprocess.spec.ts \ + packages/worker/src/__tests__/activity-options.inprocess.spec.ts \ + packages/testing/src/test-rig.ts +git commit -m "test(worker): move handlers and activity-options onto the rig" +``` + +--- + +### Task 4: Migrate `cancellation` and `continue-as-new` to the rig + +The two trickiest: cancellation produces `CANCELLED` terminal states, and continue-as-new produces `CONTINUED_AS_NEW` plus a chain of follow-on runs. + +**Files:** + +- Modify: `packages/worker/src/__tests__/cancellation.inprocess.spec.ts` +- Modify: `packages/worker/src/__tests__/continue-as-new.inprocess.spec.ts` + +**Interfaces:** + +- Consumes: `testRig` (Task 1). +- Produces: nothing later tasks depend on. + +- [ ] **Step 1: Rewrite each test's construction** + +Same mechanical replacement as Task 3 — replace the `TypedWorker.create` + `TypedClient.create` + `.for(contract)` trio with: + +```ts +const { worker, client } = await testRig(testEnv, { contract, bundle, activities }); +``` + +`continue-as-new.inprocess.spec.ts` shares the static queue for six tests and documents why in a `SHARED STATIC QUEUE CAVEAT` block. **Leave that alone** — the rig does not touch the queue, so the caveat still holds and the tests still work. + +- [ ] **Step 2: Run and triage** + +Run: `cd packages/worker && pnpm vitest run --project integration-inprocess src/__tests__/cancellation.inprocess.spec.ts src/__tests__/continue-as-new.inprocess.spec.ts` + +Two specific things to expect: + +- **Cancelled executions replay fine** — `CANCELLED` is terminal and its history is complete. If one does not replay, that is outcome 3 below. +- **Continue-as-new**: `getHandle(workflowId)` with no run ID resolves to the _latest_ run in the chain (verified in workstream 1 against `workflow-client.js:149`). So the recorded ID replays the final run's history, not the first. That is correct and sufficient — the whole chain ran the same workflow code. Note this in your report so a later reader does not mistake it for a gap. + +Same three outcomes as Task 3. A `DeterminismViolationError` means **STOP and report**. + +- [ ] **Step 3: Run the whole in-process tier** + +Run: `cd packages/worker && pnpm vitest run --project integration-inprocess` +Expected: PASS, same test count as before. + +- [ ] **Step 4: Commit** + +```bash +git add packages/worker/src/__tests__/cancellation.inprocess.spec.ts \ + packages/worker/src/__tests__/continue-as-new.inprocess.spec.ts \ + packages/testing/src/test-rig.ts +git commit -m "test(worker): move cancellation and continue-as-new onto the rig" +``` + +--- + +### Task 5: Migrate the remaining in-process specs + +**Files:** + +- Modify: `packages/worker/src/__tests__/child-wire.inprocess.spec.ts` +- Modify: `packages/worker/src/__tests__/rehydration.inprocess.spec.ts` +- Modify: `packages/worker/src/__tests__/time-skipping.inprocess.spec.ts` +- Modify: `packages/worker/src/__tests__/registration.inprocess.spec.ts` + +**Interfaces:** + +- Consumes: `testRig` (Task 1). +- Produces: nothing later tasks depend on. + +- [ ] **Step 1: Rewrite construction in the three straightforward files** + +`child-wire`, `rehydration`, and `time-skipping` take the same mechanical replacement as Task 3. + +- [ ] **Step 2: Handle `registration.inprocess.spec.ts` carefully — it must NOT use the rig for worker creation** + +`TypedWorker.create` skips `verifyWorkflowRegistration` whenever `workflowsPath` is absent, and it is always absent for a prebuilt bundle. `registration.inprocess.spec.ts` passes `workflowsPath` deliberately, and its file header documents this. + +Several of its tests also assert on the **creation Result itself** (that creation defects), so there is no worker to hand back. + +Therefore: **leave worker creation in that file exactly as it is.** Use the rig only where a test successfully creates a worker and starts a workflow, and only for the client half. If that is awkward, leave the file entirely unmigrated and record it — replay coverage of the registration fixtures is not worth breaking the one constraint the whole file exists to test. + +State which choice you made and why in your report. + +- [ ] **Step 3: Run and triage** + +Run: `cd packages/worker && pnpm vitest run --project integration-inprocess` + +Same three outcomes as Task 3. `DeterminismViolationError` means **STOP and report**. + +- [ ] **Step 4: Confirm the allowlist is honest** + +Run: `grep -A 3 "REPLAY_SKIP_ALLOWLIST" packages/testing/src/test-rig.ts` + +Every entry must name a workflow that genuinely never terminates, with a reason that says why. If an entry exists only because replay was failing for another cause, remove it and fix the cause. + +- [ ] **Step 5: Commit** + +```bash +git add packages/worker/src/__tests__/ packages/testing/src/test-rig.ts +git commit -m "test(worker): move the remaining in-process specs onto the rig" +``` + +--- + +### Task 6: Prove `nonRetryable` by attempt count + +The invariant that reframed this workstream: today `nonRetryable` is asserted only as a field on the wire failure. + +**Files:** + +- Create: `packages/worker/src/__tests__/retry.contract.ts` +- Create: `packages/worker/src/__tests__/retry.workflows.ts` +- Create: `packages/worker/src/__tests__/retry.inprocess.spec.ts` + +**Interfaces:** + +- Consumes: `testRig` (Task 1), `bundleFor` / `withTaskQueue` / `nextTaskQueueId` / `fixturePath` from `@temporal-contract/testing/workflow-bundle`. +- Produces: `retryContract`. + +- [ ] **Step 1: Write the contract with two errors differing only in retryability** + +Create `packages/worker/src/__tests__/retry.contract.ts`: + +```ts +import { defineActivity, defineContract, defineWorkflow } from "@temporal-contract/contract"; +import { z } from "zod"; + +// Composition-first: resources defined individually, then composed. + +/** + * Two declared errors identical but for `nonRetryable`. The activity chooses + * which to raise from its input, so a single fixture proves both directions + * and the only variable between them is the flag under test. + * + * `maximumAttempts: 3` bounds the retryable case: it must retry (proving the + * flag reached Temporal) without retrying forever if the flag regresses. + */ +const flaky = defineActivity({ + input: z.object({ mode: z.enum(["terminal", "retryable"]) }), + output: z.object({ attempts: z.number() }), + errors: { + TerminalFailure: { data: z.object({ at: z.number() }), nonRetryable: true }, + RetryableFailure: { data: z.object({ at: z.number() }), nonRetryable: false }, + }, + activityOptions: { + startToCloseTimeout: "10 seconds", + retry: { maximumAttempts: 3, backoffCoefficient: 1, initialInterval: "1 second" }, + }, +}); + +const runsFlaky = defineWorkflow({ + input: z.object({ mode: z.enum(["terminal", "retryable"]) }), + output: z.object({ outcome: z.string(), attempts: z.number() }), + activities: { flaky }, +}); + +export const retryContract = defineContract({ + taskQueue: "retry-tests", + workflows: { runsFlaky }, +}); +``` + +- [ ] **Step 2: Write the workflow** + +Create `packages/worker/src/__tests__/retry.workflows.ts`: + +```ts +import { ContractError, declareWorkflow } from "../workflow.js"; +import { retryContract } from "./retry.contract.js"; + +export const runsFlaky = declareWorkflow({ + workflowName: "runsFlaky", + contract: retryContract, + implementation: async (context, args) => { + const result = await context.activities.flaky({ mode: args.mode }); + + // Fold the failure into a returned status rather than rethrowing: a + // rethrown defect becomes a Workflow-Task retry loop that time-skipping + // cannot fast-forward past, turning a regression into a 120s hang. + if (result.isDefect()) return { outcome: `defect:${String(result.cause)}`, attempts: -1 }; + if (result.isErr()) { + const error = result.error; + if (error instanceof ContractError) { + // `data.at` carries the attempt number the activity failed on. + const at = (error.data as { at: number }).at; + return { outcome: `err:${error.errorName}`, attempts: at }; + } + return { outcome: `err:${error.name}`, attempts: -1 }; + } + return { outcome: "ok", attempts: result.value.attempts }; + }, +}); +``` + +- [ ] **Step 3: Write the failing spec** + +Create `packages/worker/src/__tests__/retry.inprocess.spec.ts`: + +```ts +import { it } from "@temporal-contract/testing/time-skipping"; +import { testRig } from "@temporal-contract/testing/test-rig"; +import { + bundleFor, + fixturePath, + nextTaskQueueId, + withTaskQueue, +} from "@temporal-contract/testing/workflow-bundle"; +import { Context } from "@temporalio/activity"; +import { describe, expect } from "vitest"; +import { ErrAsync } from "unthrown"; + +import { declareActivitiesHandler } from "../activity.js"; +import { retryContract } from "./retry.contract.js"; + +const WORKFLOW_EXECUTION_TIMEOUT = "30 seconds"; + +describe("nonRetryable is a behavior, not a field", () => { + it("stops after exactly one attempt when the declared error is nonRetryable", async ({ + testEnv, + }) => { + const contract = withTaskQueue(retryContract, nextTaskQueueId("retry-terminal")); + const bundle = await bundleFor(fixturePath(import.meta.url, "retry.workflows")); + + const activities = declareActivitiesHandler({ + contract, + activities: { + runsFlaky: { + // Always fails. `Context.current().info.attempt` is Temporal's own + // attempt counter, so the assertion reads what the server did. + flaky: (_input, { errors }) => + ErrAsync(errors.TerminalFailure({ at: Context.current().info.attempt })), + }, + }, + }); + + const { worker, client } = await testRig(testEnv, { contract, bundle, activities }); + + const result = await worker.raw.runUntil( + client + .executeWorkflow("runsFlaky", { + workflowId: "retry-terminal", + args: { mode: "terminal" }, + workflowExecutionTimeout: WORKFLOW_EXECUTION_TIMEOUT, + }) + .getOrThrow(), + ); + + // The whole point: attempt 1 and no more. A regression that dropped + // `nonRetryable` on the wire would let Temporal retry to 3. + expect(result).toEqual({ outcome: "err:TerminalFailure", attempts: 1 }); + }); + + it("retries when the declared error is retryable", async ({ testEnv }) => { + const contract = withTaskQueue(retryContract, nextTaskQueueId("retry-retryable")); + const bundle = await bundleFor(fixturePath(import.meta.url, "retry.workflows")); + + const activities = declareActivitiesHandler({ + contract, + activities: { + runsFlaky: { + flaky: (_input, { errors }) => + ErrAsync(errors.RetryableFailure({ at: Context.current().info.attempt })), + }, + }, + }); + + const { worker, client } = await testRig(testEnv, { contract, bundle, activities }); + + const result = await worker.raw.runUntil( + client + .executeWorkflow("runsFlaky", { + workflowId: "retry-retryable", + args: { mode: "retryable" }, + workflowExecutionTimeout: WORKFLOW_EXECUTION_TIMEOUT, + }) + .getOrThrow(), + ); + + // Exhausts the declared `maximumAttempts: 3`, so the surfaced failure is + // from attempt 3 — proving Temporal really retried. + expect(result).toEqual({ outcome: "err:RetryableFailure", attempts: 3 }); + }); +}); +``` + +- [ ] **Step 4: Run** + +Run: `cd packages/worker && pnpm vitest run --project integration-inprocess src/__tests__/retry.inprocess.spec.ts` +Expected: PASS (2 tests). + +If the retryable case reports `attempts: 1`, that is a **real finding** — it would mean `nonRetryable: false` is not reaching Temporal. Report it; do not adjust the expectation. + +- [ ] **Step 5: Reality-check both assertions against production source** + +Break `packages/worker/src/contract-errors.ts`'s `nonRetryable: definition.nonRetryable ?? false` — hardcode it to `false`, confirm the terminal test now reports `attempts: 3` and fails; then hardcode `true` and confirm the retryable test reports `attempts: 1` and fails. Restore. + +Quote both verbatim failures. Production source must be byte-identical afterwards. + +- [ ] **Step 6: Add the knip entry and commit** + +`knip.json` already globs `src/__tests__/*.workflows.ts`, so no change is needed — verify with `pnpm knip`. + +```bash +git add packages/worker/src/__tests__/retry.contract.ts \ + packages/worker/src/__tests__/retry.workflows.ts \ + packages/worker/src/__tests__/retry.inprocess.spec.ts +git commit -m "test(worker): prove nonRetryable by attempt count, not by field" +``` + +--- + +### Task 7: Prove timeout forwarding via `Context.current().info` + +**Files:** + +- Create: `packages/worker/src/__tests__/timeouts.contract.ts` +- Create: `packages/worker/src/__tests__/timeouts.workflows.ts` +- Create: `packages/worker/src/__tests__/timeouts.inprocess.spec.ts` + +**Interfaces:** + +- Consumes: `testRig` (Task 1). +- Produces: `timeoutsContract`. + +- [ ] **Step 1: Write a contract exercising all three merge layers** + +Create `packages/worker/src/__tests__/timeouts.contract.ts`: + +```ts +import { defineActivity, defineContract, defineWorkflow } from "@temporal-contract/contract"; +import { z } from "zod"; + +/** + * The activity reports back the timeouts Temporal actually materialized for + * its task, so each layer of the merge is proven by effect rather than by + * inspecting the options we passed in. + * + * `heartbeatTimeout` is the one this repo asserted nowhere before — a lost + * heartbeat timeout means a dead activity is never retried. + */ +const reportsTimeouts = defineActivity({ + input: z.object({}), + output: z.object({ + startToCloseMs: z.number(), + scheduleToCloseMs: z.number(), + heartbeatMs: z.number(), + }), + // Contract layer: heartbeat declared here and nowhere else. + activityOptions: { + heartbeatTimeout: "7 seconds", + retry: { maximumAttempts: 1 }, + }, +}); + +const reportsLayered = defineWorkflow({ + input: z.object({}), + output: z.object({ + startToCloseMs: z.number(), + scheduleToCloseMs: z.number(), + heartbeatMs: z.number(), + }), + activities: { reportsTimeouts }, +}); + +export const timeoutsContract = defineContract({ + taskQueue: "timeouts-tests", + workflows: { reportsLayered }, +}); +``` + +- [ ] **Step 2: Write the workflow supplying the other two layers** + +Create `packages/worker/src/__tests__/timeouts.workflows.ts`: + +```ts +import { declareWorkflow } from "../workflow.js"; +import { timeoutsContract } from "./timeouts.contract.js"; + +export const reportsLayered = declareWorkflow({ + workflowName: "reportsLayered", + contract: timeoutsContract, + // Workflow-wide layer. + activityOptions: { scheduleToCloseTimeout: "20 seconds" }, + // Per-activity layer — most specific, must win for startToClose. + activityOptionsByName: { reportsTimeouts: { startToCloseTimeout: "9 seconds" } }, + implementation: async (context) => { + const result = await context.activities.reportsTimeouts({}); + + if (result.isDefect()) throw result.cause; + if (result.isErr()) return { startToCloseMs: -1, scheduleToCloseMs: -1, heartbeatMs: -1 }; + return result.value; + }, +}); +``` + +- [ ] **Step 3: Write the spec** + +Create `packages/worker/src/__tests__/timeouts.inprocess.spec.ts`: + +```ts +import { it } from "@temporal-contract/testing/time-skipping"; +import { testRig } from "@temporal-contract/testing/test-rig"; +import { + bundleFor, + fixturePath, + nextTaskQueueId, + withTaskQueue, +} from "@temporal-contract/testing/workflow-bundle"; +import { Context } from "@temporalio/activity"; +import { describe, expect } from "vitest"; +import { OkAsync } from "unthrown"; + +import { declareActivitiesHandler } from "../activity.js"; +import { timeoutsContract } from "./timeouts.contract.js"; + +const WORKFLOW_EXECUTION_TIMEOUT = "30 seconds"; + +describe("activity timeouts reach Temporal through every merge layer", () => { + it("materializes the value each layer contributed", async ({ testEnv }) => { + const contract = withTaskQueue(timeoutsContract, nextTaskQueueId("timeouts")); + const bundle = await bundleFor(fixturePath(import.meta.url, "timeouts.workflows")); + + const activities = declareActivitiesHandler({ + contract, + activities: { + reportsLayered: { + // Reads what Temporal actually scheduled the task with, rather than + // what we passed in — an effect, not a call shape. + reportsTimeouts: () => { + const info = Context.current().info; + return OkAsync({ + startToCloseMs: info.startToCloseTimeoutMs, + scheduleToCloseMs: info.scheduleToCloseTimeoutMs, + // `heartbeatTimeoutMs` is optional on ActivityInfo — 0 + // distinguishes "declared but lost in the merge" from "never + // declared", so the assertion below fails loudly either way. + heartbeatMs: info.heartbeatTimeoutMs ?? 0, + }); + }, + }, + }, + }); + + const { worker, client } = await testRig(testEnv, { contract, bundle, activities }); + + const result = await worker.raw.runUntil( + client + .executeWorkflow("reportsLayered", { + workflowId: "timeouts-layered", + args: {}, + workflowExecutionTimeout: WORKFLOW_EXECUTION_TIMEOUT, + }) + .getOrThrow(), + ); + + expect(result).toEqual({ + startToCloseMs: 9_000, // activityOptionsByName wins — most specific layer + scheduleToCloseMs: 20_000, // declareWorkflow's workflow-wide default + heartbeatMs: 7_000, // contract-level, the layer with no competitor + }); + }); +}); +``` + +- [ ] **Step 4: Run** + +Run: `cd packages/worker && pnpm vitest run --project integration-inprocess src/__tests__/timeouts.inprocess.spec.ts` +Expected: PASS. + +`heartbeatMs: 0` would mean the contract-level heartbeat is lost in the merge — a **real finding**. Report it; do not adjust the expectation. + +- [ ] **Step 5: Reality-check each of the three values** + +Break the merge in `packages/worker/src/internal.ts` — drop each layer in turn from the spread — and confirm the corresponding field changes and the test fails. Restore. Quote all three verbatim failures. + +- [ ] **Step 6: Commit** + +```bash +git add packages/worker/src/__tests__/timeouts.contract.ts \ + packages/worker/src/__tests__/timeouts.workflows.ts \ + packages/worker/src/__tests__/timeouts.inprocess.spec.ts +git commit -m "test(worker): prove timeout forwarding through every merge layer" +``` + +--- + +### Task 8: Final verification + +- [ ] **Step 1: Full suite** + +Run: `pnpm turbo run typecheck lint test` — report exact task counts. +Run: `cd packages/worker && pnpm vitest run --project unit` — report exact counts. +Run: `cd packages/worker && pnpm vitest run --project integration-inprocess` — report exact counts. +Run: `cd packages/testing && pnpm vitest run` — report exact counts. +Run: `pnpm knip` — must stay clean. + +- [ ] **Step 2: Docker integration, serially** + +Run: `pnpm turbo run test:integration --concurrency=1` + +The script now runs the two tiers sequentially (fixed in workstream 1). If a test fails, re-run that file alone to distinguish a genuine failure from contention. + +- [ ] **Step 3: Measure wall-clock** + +Run: `cd packages/worker && time pnpm vitest run --project integration-inprocess` + +The pre-workstream-2 baseline is ~39s for unit+in-process combined. Report the new number and the delta. Replay needs no server, but it is not free. + +- [ ] **Step 4: Audit the skip allowlist** + +Report every `REPLAY_SKIP_ALLOWLIST` entry with its reason, and confirm each names a workflow that genuinely never reaches a terminal state. An entry that exists to silence a replay failure is a defect, not an exemption. + +- [ ] **Step 5: Report replay coverage** + +State how many executions were replayed across the in-process tier and how many were skipped. This is the number that says whether the harvest worked. + +--- + +## Deferred (explicitly out of scope) + +- Idempotency surface (reuse/conflict policy, safe defaults) — workstream 4. +- Cancellation semantics — proven in workstream 1. +- Raising the Stryker `break` threshold — the baseline is one nightly old. +- The `withTestWorker` boilerplate refactor flagged in workstream 1's final review — `testRig` supersedes part of it, but consolidating the remaining ~500 lines of standup is a separate cleanup. diff --git a/docs/superpowers/specs/2026-08-02-determinism-invariants-design.md b/docs/superpowers/specs/2026-08-02-determinism-invariants-design.md new file mode 100644 index 00000000..6c59167b --- /dev/null +++ b/docs/superpowers/specs/2026-08-02-determinism-invariants-design.md @@ -0,0 +1,191 @@ +# Determinism and money-safety invariants + +**Date:** 2026-08-02 +**Status:** Approved +**Scope:** Workstream 2 of 4 in the production-hardening effort + +## Context + +`temporal-contract` is in production use where real money depends on it. The +hardening driver is preventive — no incident has occurred. + +The four workstreams: + +1. Mock-free test architecture — **shipped 2026-08-02 (PR #359)** +2. **Determinism & money-safety invariants** (this spec) +3. API/type strength — making misuse unrepresentable +4. Pattern enforcement — forcing correct usage on end users + +Workstream 1 established that the suite asserts real effects against a real +time-skipping server, and produced a mutation baseline of 61.45%. This +workstream uses that foundation to prove the invariants the library could +silently break. + +### What "money-safety" means here + +The library does not move money — it is a contract layer. Its money-safety role +is therefore **defensive**: Temporal offers guarantees, and the library sits +between the user and those guarantees. If the library corrupts one, the user's +workflow misbehaves in a way their own tests would not catch. + +Idempotency surface (contract-level reuse/conflict policy, safe defaults) was +considered and **deliberately excluded**: `workflowIdReusePolicy` and +`workflowIdConflictPolicy` already pass through untouched +(`TypedWorkflowStartOptions` omits only `taskQueue`, `args`, and the two +search-attribute fields), so this is API design rather than invariant proving, +and it belongs with workstream 4. + +### Measured starting state + +| Invariant | State | +| ------------------------ | ---------------------------------------------------------------------- | +| `nonRetryable` behavior | Asserted **only as a field value** on the `ApplicationFailure` | +| Timeout behavior | `startToCloseTimeout` proven by effect; `heartbeatTimeout` **nowhere** | +| Replay determinism | **2 paths** of ~8 sandboxed constructs | +| Idempotency pass-through | Works; no guidance (out of scope, see above) | + +The `nonRetryable` gap is the one to lead with. Every existing assertion reads +`raw.nonRetryable === true` off the wire failure. That is a call-shape +assertion in the precise sense workstream 1 set out to eliminate — it would +survive the flag being dropped between the wire and Temporal's retry decision. +It persisted because those assertions were _added_ during workstream 1, as +fixes restoring a dropped property, and restoring a property is not the same as +proving a behavior. + +## Invariants to prove + +### Retry semantics + +- A contract error declared `nonRetryable: true` results in **exactly one** + activity attempt. +- A contract error declared `nonRetryable: false` results in **more than one** + attempt. +- Attempt counts are read from the activity itself (`Context.current().info.attempt`) + or a counter the activity increments — not inferred from elapsed time. + +### Timeout semantics + +- `startToCloseTimeout`, `scheduleToCloseTimeout`, and `heartbeatTimeout` reach + Temporal as declared through all three option-merge layers + (`declareWorkflow`'s `activityOptions` → contract-level `activityOptions` → + `activityOptionsByName`). +- Asserted by reading `Context.current().info` inside the activity, which + exposes `startToCloseTimeoutMs`, `scheduleToCloseTimeoutMs`, and + `heartbeatTimeoutMs` — the values Temporal actually materialized for the + scheduled task. + +This mechanism was identified in workstream 1's final review as a lever the +whole workstream missed: it is an effect, not a call shape, and it removes the +multi-second real sleeps currently used to prove timeouts by waiting for them. + +### Replay determinism + +Every construct that runs unthrown pipelines or async Standard Schema +validation inside Temporal's sandbox must replay without a +`DeterminismViolationError`: workflow entry/exit, activity calls, signals, +queries, updates, child workflows, continue-as-new, and cancellation. + +## Architecture + +### The replay harvest rig + +The in-process tier already runs ~56 tests producing real histories across +every construct. Replaying a history needs no server — only the workflow +bundle. Harvesting those histories therefore buys near-total replay coverage, +and — the durable part — **every future test gets replay coverage +automatically**, so it cannot rot the way an enumerated list does. + +The obstacle: an `afterEach` cannot know which bundle to replay against, +because tests hold the bundle locally. The fix moves the seam to where the +bundle already is. + +```ts +// before — two lines already present in every test +const worker = await TypedWorker.create({ contract, connection, workflowBundle: bundle }).get(); +const typedClient = await TypedClient.create({ client: testEnv.client }).get(); + +// after — one line, same information +const { worker, client } = await testRig(testEnv, { contract, bundle, activities }); +``` + +`testRig` lives in `@temporal-contract/testing` and returns: + +- a `TypedWorker` built exactly as before, and +- a `ContractClient` that **records every `workflowId` it starts**, paired with + the bundle the worker was built from. + +Two signature details, both load-bearing: + +- **`activities` is optional.** Workflow-only workers exist and must keep + working — `TypedWorker.create` treats an absent `activities` key differently + from `activities: undefined` (exactOptionalPropertyTypes discipline), so the + rig must spread it conditionally rather than pass it through unconditionally. +- **The rig takes the contract as given and does not scope the task queue.** + Callers keep calling `withTaskQueue(contract, nextTaskQueueId(...))` + themselves. This is not laziness: workstream 1 established that a + same-workflow continue-as-new _must_ land on the contract's static queue, + because the contract is closed over inside the bundled workflow module and a + test-side copy can never reach it. Six tests deliberately share the static + queue for that reason. A rig that scoped unconditionally would break them. + +An `afterEach` registered by the fixture fetches each recorded execution's +history and replays it with `Worker.runReplayHistory`. + +Client construction is already uniform — 48 of ~52 in-process tests use a +byte-identical `TypedClient.create({ client: testEnv.client }).get()` line — so +this is a mechanical rewrite, not a redesign. + +**Recording at the client, not listing from the server.** The alternative — +enumerating executions per task queue via visibility — would avoid touching +tests, but depends on the time-skipping test server supporting workflow +listing, which is unverified. Recording is independent of that. + +### Non-terminal executions + +Several tests deliberately leave workflows running (`condition(() => false)` as +a probe that never completes). Replaying a partial history may error rather +than pass. + +The rig skips executions that are not in a terminal state — **and every skip +must be explicitly opted out by workflow-ID prefix with a stated reason**, in a +shrink-only allowlist mirroring `no-sdk-mocks.spec.ts`. An unlisted skip fails +the test. + +A silently-skipped execution is exactly the coverage rot this workstream exists +to prevent: it would report replay coverage it does not have. The allowlist +makes each exemption a decision someone made on purpose. + +## Risks + +| Risk | Mitigation | +| ------------------------------------------------- | ---------------------------------------------------------------------------------------------------------- | +| Non-terminal executions silently skipped | Shrink-only allowlist keyed by workflow-ID prefix; unlisted skips fail | +| Wall-clock growth on the slowest tier | Replay needs no server. Measure before/after; if material, the lever is `poolOptions.forks.isolate: false` | +| The rig rewrite silently drops a test's assertion | Rewrite is mechanical and reviewed per file; the existing suite must stay green throughout | +| A replay failure is a _real_ determinism bug | Stop and report it. Do not adjust the test to pass — that is the bug this workstream exists to find | + +That last row is the one to take seriously. If harvesting surfaces a genuine +`DeterminismViolationError` in shipped code, that is a **finding, not an +obstacle**, and it outranks the rest of the workstream. + +## Out of scope + +- Idempotency surface (reuse/conflict policy, safe defaults) — belongs to + workstream 4. +- Cancellation semantics — proven in workstream 1 + (`cancellation.inprocess.spec.ts`, including the swallowed-cancellation + hazard and its `rethrowCancellation` fix). +- Raising the Stryker `break` threshold — the baseline is one nightly old. + +## Success criteria + +1. `nonRetryable: true` and `false` are each proven by **attempt count**, not + by a field read. +2. `startToCloseTimeout`, `scheduleToCloseTimeout`, and `heartbeatTimeout` are + each proven through the option-merge layers via `Context.current().info`. +3. Every in-process test's history is replayed, or its skip is explicitly + allowlisted with a reason. +4. The rig makes replay coverage automatic for future tests — adding a test + requires no replay-specific code. +5. The existing suite stays green, and in-process wall-clock growth is measured + and reported. diff --git a/packages/testing/package.json b/packages/testing/package.json index 66e3f0b3..a98a9642 100644 --- a/packages/testing/package.json +++ b/packages/testing/package.json @@ -44,6 +44,10 @@ "import": "./dist/global-setup.mjs" }, "./package.json": "./package.json", + "./test-rig": { + "types": "./dist/test-rig.d.mts", + "import": "./dist/test-rig.mjs" + }, "./time-skipping": { "types": "./dist/time-skipping.d.mts", "import": "./dist/time-skipping.mjs" @@ -54,7 +58,7 @@ } }, "scripts": { - "build": "tsdown src/global-setup.ts src/extension.ts src/time-skipping.ts src/contract.ts src/activity.ts src/workflow-bundle.ts --format esm --dts --clean", + "build": "tsdown src/global-setup.ts src/extension.ts src/time-skipping.ts src/contract.ts src/activity.ts src/workflow-bundle.ts src/test-rig.ts --format esm --dts --clean", "build:docs": "typedoc", "check:package": "publint --strict && attw --pack . --profile esm-only", "test": "vitest run --project unit", diff --git a/packages/testing/src/inprocess-specs-use-rig.spec.ts b/packages/testing/src/inprocess-specs-use-rig.spec.ts new file mode 100644 index 00000000..b4c4df49 --- /dev/null +++ b/packages/testing/src/inprocess-specs-use-rig.spec.ts @@ -0,0 +1,166 @@ +import { readFile, readdir } from "node:fs/promises"; +import { join, relative, sep } from "node:path"; +import { fileURLToPath } from "node:url"; + +import { describe, expect, it } from "vitest"; + +const WORKSPACE_ROOT = fileURLToPath(new URL("../../../", import.meta.url)); + +/** + * Workspace-relative path in POSIX form, whatever the host separator is. See + * `no-sdk-mocks.spec.ts`'s identical helper for the Windows rationale. + */ +function workspaceRelative(absolutePath: string): string { + return relative(WORKSPACE_ROOT, absolutePath).split(sep).join("/"); +} + +/** + * `*.inprocess.spec.ts` files permitted to contain a test that does not call + * `testRig(`. Every entry needs a reason. + * + * This list may only ever shrink. A hand-rolled `TypedWorker.create` + + * `TypedClient.create` pair (the pattern this rejects) silently gets no + * replay coverage — exactly the rot `testRig`'s `onTestFinished` hook exists + * to prevent — so an unlisted test without `testRig(` fails the guard + * instead of quietly shipping uncovered. + */ +const ALLOWLIST: Record = { + "packages/worker/src/__tests__/registration.inprocess.spec.ts": + "every test here must pass workflowsPath directly to TypedWorker.create — testRig's " + + "RigOptions only accepts a prebuilt bundle, which always skips verifyWorkflowRegistration " + + "(see the option's JSDoc), silently exempting the very check this file exists to exercise.", + "packages/worker/src/__tests__/time-skipping.inprocess.spec.ts": + "4 of its 5 tests build the worker/client pair by hand: 2 assert on the TypedWorker.create " + + "/ TypedClient.create Result itself (Ok/Defect), which testRig's `.get()`-unwrapping " + + "helper hides; 2 need `interceptors` / arbitrary WorkerOptions passthrough that RigOptions " + + "does not expose.", +}; + +async function inprocessSpecFiles(dir: string): Promise { + const entries = await readdir(dir, { withFileTypes: true }); + const found: string[] = []; + for (const entry of entries) { + const full = join(dir, entry.name); + if (entry.isDirectory()) { + if (entry.name === "node_modules" || entry.name === "dist" || entry.name.startsWith(".")) + continue; + found.push(...(await inprocessSpecFiles(full))); + } else if (entry.name.endsWith(".inprocess.spec.ts")) { + found.push(full); + } + } + return found; +} + +/** + * Matches a top-level test declaration in any of Vitest's spellings — `it(`, + * `test(`, and every modifier chain (`it.each`, `it.concurrent`, `it.skip`, + * `test.only`, …). + * + * Deliberately broader than `^\s*it\(`. A narrower pattern makes this guard + * fail *open*: a file written entirely with `test(` or `it.each(` yields zero + * blocks, the loop below never runs, and the file passes while enforcing + * nothing. The corpus-level block assertion is the second net for that. + * + * `\b` keeps `itemCount` / `testEnv` from matching. + */ +const TEST_START = /^\s*(?:it|test)\b/; + +/** + * Strip comments before looking for `testRig(`. Without this, a block that + * merely *mentions* the rig in prose — "// migrated off testRig(...)" — + * satisfies the check without calling it, which is the same fail-open shape + * the pattern above guards against. + */ +function stripComments(source: string): string { + return source.replace(/\/\*[\s\S]*?\*\//g, "").replace(/(^|[^:])\/\/.*$/gm, "$1"); +} + +/** + * Split a spec file's source into one chunk per `it(` block, each running + * from its `it(` line to the line before the next one (or EOF). Line-based + * rather than brace-balanced: every file in this corpus opens tests at + * `^\s*it\(`, and slicing on that is enough to attribute a `testRig(` call + * (or its absence) to the right test without a full parser. + */ +function testBlocks(source: string): string[] { + const lines = source.split("\n"); + const starts = lines.reduce((acc, line, index) => { + if (TEST_START.test(line)) acc.push(index); + return acc; + }, []); + return starts.map((start, index) => { + const end = index + 1 < starts.length ? starts[index + 1] : lines.length; + return lines.slice(start, end).join("\n"); + }); +} + +/** Pulls the `it("...")` description out of a test block, for readable offender messages. */ +function testDescription(block: string): string { + // Mirrors TEST_START's breadth — a `test(` or `it.each(` offender should be + // named in the failure, not reported as "(description not found)". + const match = /(?:it|test)\b[^(]*\(\s*["'`]([^"'`]*)["'`]/.exec(block); + return match?.[1] ?? "(description not found)"; +} + +describe("every in-process test uses the rig", () => { + it("keeps replay coverage from silently regressing to zero for a new test", async () => { + const files = await inprocessSpecFiles(WORKSPACE_ROOT); + + // Positive control on the directory walk — see `no-sdk-mocks.spec.ts`'s + // identical rationale: an empty (or near-empty) `files` would make the + // loop below pass vacuously. The workspace has well over 10 + // `*.inprocess.spec.ts` files today. + expect(files.length).toBeGreaterThan(10); + + // Second, block-level positive control. The file count above proves the + // corpus walk found files; it does NOT prove `testBlocks` found tests + // inside them. If TEST_START ever stops matching this repo's spelling, + // every file yields zero blocks and the offender loop passes while + // checking nothing — the precise fail-open this guard exists to prevent. + let totalBlocks = 0; + + const offenders: string[] = []; + + for (const file of files) { + const rel = workspaceRelative(file); + if (rel in ALLOWLIST) continue; + + const source = await readFile(file, "utf8"); + const blocks = testBlocks(source); + totalBlocks += blocks.length; + for (const block of blocks) { + if (!stripComments(block).includes("testRig(")) { + offenders.push(`${rel}: "${testDescription(block)}"`); + } + } + } + + // The tier has ~59 tests; anything near zero means TEST_START stopped + // matching, not that the tests vanished. + expect(totalBlocks).toBeGreaterThan(40); + + expect( + offenders, + `These in-process tests don't call testRig(, so their executions get no replay-determinism ` + + `coverage. Either use testRig (see any other *.inprocess.spec.ts) or, if the rig genuinely ` + + `can't cover this case, add a reason-carrying entry to this file's ALLOWLIST.`, + ).toEqual([]); + }); + + it("has no stale allowlist entries", async () => { + const stale: string[] = []; + for (const rel of Object.keys(ALLOWLIST)) { + const source = await readFile(join(WORKSPACE_ROOT, rel), "utf8").catch(() => ""); + const stillOffRig = testBlocks(source).some( + (block) => !stripComments(block).includes("testRig("), + ); + if (!stillOffRig) stale.push(rel); + } + + expect( + stale, + "Allowlisted files where every test now calls testRig( — delete these entries.", + ).toEqual([]); + }); +}); diff --git a/packages/testing/src/test-rig.spec.ts b/packages/testing/src/test-rig.spec.ts new file mode 100644 index 00000000..38bd4fe9 --- /dev/null +++ b/packages/testing/src/test-rig.spec.ts @@ -0,0 +1,130 @@ +import { ContractClient } from "@temporal-contract/client"; +import { describe, expect, it } from "vitest"; + +import { + extractStartedWorkflowId, + isTerminalStatus, + skipReasonFor, + START_METHODS, +} from "./test-rig.js"; + +describe("isTerminalStatus", () => { + it("treats every finished status as terminal", () => { + for (const name of ["COMPLETED", "FAILED", "CANCELLED", "TERMINATED", "TIMED_OUT"]) { + expect(isTerminalStatus(name)).toBe(true); + } + }); + + it("treats CONTINUED_AS_NEW as terminal — that run's history is complete and replayable", () => { + expect(isTerminalStatus("CONTINUED_AS_NEW")).toBe(true); + }); + + it("treats unfinished statuses as non-terminal", () => { + for (const name of ["RUNNING", "PAUSED", "UNSPECIFIED", "UNKNOWN"]) { + expect(isTerminalStatus(name)).toBe(false); + } + }); +}); + +describe("skipReasonFor", () => { + it("matches an allowlist entry by workflow-ID prefix", () => { + expect(skipReasonFor("probe-edge-cases-1", { "probe-edge-cases": "blocks forever" })).toBe( + "blocks forever", + ); + }); + + it("returns undefined for an unlisted id, so the caller can fail", () => { + expect( + skipReasonFor("some-other-id", { "probe-edge-cases": "blocks forever" }), + ).toBeUndefined(); + }); + + it("returns undefined against an empty allowlist — testRig's own default", () => { + expect(skipReasonFor("anything", {})).toBeUndefined(); + }); +}); + +describe("START_METHODS", () => { + it("names exactly ContractClient's start-capable public methods", () => { + // `createTypedHandle` is `private` in the TypeScript source, but + // `private` is compile-time only — it still shows up in runtime + // reflection below, so it's named and excluded explicitly rather than + // silently swallowed by some naming convention that could just as + // easily hide a real public method in the future. + const PRIVATE_HELPER = "createTypedHandle"; + // `getHandle` is ContractClient's one public method that does NOT start + // an execution — everything else public is in START_METHODS. + const NON_START_METHOD = "getHandle"; + + const publicMethods = Object.getOwnPropertyNames(ContractClient.prototype).filter((name) => { + if (name === "constructor" || name === PRIVATE_HELPER) return false; + // `getOwnPropertyDescriptor` (rather than direct property access) + // avoids invoking `taskQueue`'s getter on the bare prototype, which + // has no bound instance state and would throw. + const descriptor = Object.getOwnPropertyDescriptor(ContractClient.prototype, name); + return typeof descriptor?.value === "function"; + }); + + expect(new Set(publicMethods)).toEqual(new Set([...START_METHODS, NON_START_METHOD])); + }); +}); + +describe("extractStartedWorkflowId", () => { + it("extracts the workflowId from the options bag at args[1]", () => { + expect( + extractStartedWorkflowId("startWorkflow", ["myWorkflow", { workflowId: "order-123" }]), + ).toBe("order-123"); + }); + + it("throws naming the method and the received bag when workflowId is missing", () => { + expect(() => extractStartedWorkflowId("startWorkflow", ["myWorkflow", {}])).toThrow( + /startWorkflow.*workflowId/s, + ); + }); + + it("throws when the second argument isn't an object at all", () => { + expect(() => extractStartedWorkflowId("executeWorkflow", ["myWorkflow", undefined])).toThrow( + /executeWorkflow.*workflowId/s, + ); + }); + + it("throws when workflowId is present but not a string", () => { + expect(() => + extractStartedWorkflowId("signalWithStart", ["myWorkflow", { workflowId: 123 }]), + ).toThrow(/signalWithStart.*workflowId/s); + }); +}); + +describe("skipReasonFor — overlapping prefixes", () => { + it("returns the longest match, not the first declared", () => { + // Declaration order puts the broader prefix first; a first-match + // implementation would return "broad" and shadow the narrower entry. + const allowlist = { order: "broad", "order-cancel": "narrow" }; + expect(skipReasonFor("order-cancel-1", allowlist)).toBe("narrow"); + }); + + it("is order-independent", () => { + const a = { order: "broad", "order-cancel": "narrow" }; + const b = { "order-cancel": "narrow", order: "broad" }; + expect(skipReasonFor("order-cancel-1", a)).toBe(skipReasonFor("order-cancel-1", b)); + }); +}); + +describe("extractStartedWorkflowId — unserializable options bag", () => { + it("still reports the guard error when the bag cannot be JSON-stringified", () => { + // A BigInt makes JSON.stringify throw. The guard's own error must survive + // that, or an unrelated TypeError replaces the actionable diagnostic. + const bag = { workflowId: 1n as unknown as string, nested: 2n }; + expect(() => extractStartedWorkflowId("startWorkflow", ["wf", bag])).toThrowError( + /testRig expected "startWorkflow"/, + ); + }); + + it("does not let a circular bag mask the guard error", () => { + const bag: Record = {}; + bag["self"] = bag; + expect(() => extractStartedWorkflowId("executeWorkflow", ["wf", bag])).toThrowError( + /testRig expected "executeWorkflow"/, + ); + }); +}); diff --git a/packages/testing/src/test-rig.ts b/packages/testing/src/test-rig.ts new file mode 100644 index 00000000..5b9a7558 --- /dev/null +++ b/packages/testing/src/test-rig.ts @@ -0,0 +1,282 @@ +import type { ContractClient } from "@temporal-contract/client"; +import { TypedClient } from "@temporal-contract/client"; +import type { ContractDefinition } from "@temporal-contract/contract"; +// `ActivitiesHandler` lives on the /activity subpath — worker.ts imports it +// but does not re-export it. `TypedWorker` is both a type and a value, so one +// non-type import covers both uses. +import type { ActivitiesHandler } from "@temporal-contract/worker/activity"; +import { TypedWorker } from "@temporal-contract/worker/worker"; +import type { TestWorkflowEnvironment } from "@temporalio/testing"; +import { Worker } from "@temporalio/worker"; +import type { History, WorkflowBundleWithSourceMap } from "@temporalio/worker"; +import { onTestFinished } from "vitest"; + +/** Statuses whose history is complete and therefore replayable. */ +const TERMINAL_STATUSES = new Set([ + "COMPLETED", + "FAILED", + "CANCELLED", + "TERMINATED", + "TIMED_OUT", + // The run ended; the next run is a separate execution with its own history. + "CONTINUED_AS_NEW", +]); + +/** + * Whether a workflow-execution status names a finished run, and is therefore + * safe to fetch and replay. An unscoped `handle.describe()` (no `runId`) + * always resolves to the *latest* run in a chain, so in practice it can + * never itself report `CONTINUED_AS_NEW` — that status only ever shows up if + * a caller describes a specific older run directly. It's kept in the set + * anyway because it genuinely is a finished, replayable state for whichever + * run it's read from. + */ +export function isTerminalStatus(name: string): boolean { + return TERMINAL_STATUSES.has(name); +} + +/** + * Look up the caller-supplied `replaySkipAllowlist` reason (see + * {@link RigOptions}) for a non-terminal execution, matching by workflow-ID + * *prefix* so one entry covers every workflow ID `nextTaskQueueId`-style + * counters generate from a shared base (e.g. `"probe-edge-cases"` matches + * `"probe-edge-cases-1"`, `"probe-edge-cases-2"`, ...). Returns `undefined` + * for anything unlisted, so the caller can fail loudly instead of silently + * under-reporting replay coverage. + */ +export function skipReasonFor( + workflowId: string, + allowlist: Readonly>, +): string | undefined { + // Longest match wins, not first. `Object.entries` order would otherwise make + // overlapping prefixes ("order" and "order-cancel") resolve to whichever was + // declared first — so the same workflow ID could pick up a different reason + // purely from key ordering, and a deliberately narrower entry could be + // shadowed by a broader one. + let best: { prefix: string; reason: string } | undefined; + for (const [prefix, reason] of Object.entries(allowlist)) { + if (!workflowId.startsWith(prefix)) continue; + if (best === undefined || prefix.length > best.prefix.length) best = { prefix, reason }; + } + return best?.reason; +} + +/** + * The three `ContractClient` methods that can start an execution. Pinned + * against `ContractClient`'s actual method surface by a unit test in + * `test-rig.spec.ts` — exported so that test can import it; guarded again at + * runtime inside {@link testRig} itself, since a unit test only catches drift + * when someone remembers to run it. + */ +export const START_METHODS = new Set(["startWorkflow", "executeWorkflow", "signalWithStart"]); + +/** + * `JSON.stringify` for a diagnostic message, but never at the cost of the + * diagnostic. A bag containing a BigInt (or a circular reference) makes + * `JSON.stringify` throw, which would replace the guard's actionable error + * with an unrelated TypeError — hiding exactly the problem it exists to + * surface. Falls back to a coarse description. + */ +function describeBag(bag: unknown): string { + try { + return JSON.stringify(bag) ?? String(bag); + } catch { + return `[unserializable ${typeof bag}]`; + } +} +/** + * Pull the `workflowId` out of a start method's options bag (its second + * argument, per every `START_METHODS` signature) so the rig knows which + * execution to replay later. + * + * This is the rig's single load-bearing assumption about another package's + * call shape — held with no runtime enforcement anywhere else. Throwing here + * when the assumption doesn't hold is deliberate: the alternative is + * `startedIds` silently staying empty, `onTestFinished` iterating nothing, + * and the test passing green while proving zero replay coverage — exactly + * the failure mode this whole rig exists to prevent. Pure and exported so + * the guard is unit-testable without a server. + */ +export function extractStartedWorkflowId(methodName: string, args: readonly unknown[]): string { + const bag = args[1]; + const workflowId = + typeof bag === "object" && bag !== null && "workflowId" in bag + ? (bag as { workflowId?: unknown }).workflowId + : undefined; + if (typeof workflowId !== "string") { + // oxlint-disable-next-line unthrown/no-throw -- test-harness assertion: guards the rig's one load-bearing assumption about ContractClient's call shape; see this function's JSDoc + throw new Error( + `testRig expected "${methodName}"'s second argument to carry a string "workflowId" ` + + `(every Temporal WorkflowOptions requires one) but received: ${describeBag(bag)}. ` + + `Without it, this execution's history can never be harvested for replay.`, + ); + } + return workflowId; +} + +/** + * Duck-types `@temporalio/common`'s `WorkflowNotFoundError` by `error.name` + * rather than `instanceof`, so this module doesn't need `@temporalio/common` + * as a direct dependency — `@temporalio/client`'s decorator-based error + * classes (`SymbolBasedInstanceOfError`) set `name` on the prototype + * unconditionally, so the check is as reliable as an `instanceof` would be. + */ +function isWorkflowNotFoundError(error: unknown): boolean { + return error instanceof Error && error.name === "WorkflowNotFoundError"; +} + +type RigOptions = { + readonly contract: TContract; + readonly bundle: WorkflowBundleWithSourceMap; + readonly activities?: ActivitiesHandler; + /** + * Workflow-ID prefixes (matched via {@link skipReasonFor}) whose executions + * are deliberately left non-terminal, so their histories cannot be + * replayed. Every entry needs a reason. Defaults to `{}` — a published rig + * cannot know a consuming repo's fixture IDs, so nothing is skipped unless + * the caller opts a workflow ID in explicitly. + * + * This list may only ever shrink per caller. A silently-skipped execution + * would report replay coverage it does not have — exactly the rot this rig + * exists to prevent — so an unlisted non-terminal execution fails the test + * instead. + */ + readonly replaySkipAllowlist?: Readonly>; +}; + +/** + * Replay every run in a continue-as-new/retry/cron chain by walking + * *backward* from the latest run, replaying newest to oldest as each prior + * run's id is discovered. + * + * `getHandle(workflowId)` with no `runId` binds to the newest run, so its + * history alone omits every earlier run — including the ones that actually + * contain the continue-as-new command, the determinism surface most worth + * replaying. Each run's `WorkflowExecutionStarted` event records the prior + * run's id in `continuedExecutionRunId` (populated for continue-as-new, + * retry, and cron alike), so walking that pointer back to its origin and + * replaying each run visited is the only way to cover the whole chain. + * Earlier runs need no `describe()` / terminal check of their own: a run + * reachable this way already closed — that's *why* the next run exists. + */ +async function replayChain( + client: TestWorkflowEnvironment["client"], + bundle: WorkflowBundleWithSourceMap, + workflowId: string, +): Promise { + let runId: string | undefined = undefined; + for (;;) { + const history: History = await client.workflow.getHandle(workflowId, runId).fetchHistory(); + await Worker.runReplayHistory({ workflowBundle: bundle }, history, workflowId); + + const previousRunId = + history.events?.[0]?.workflowExecutionStartedEventAttributes?.continuedExecutionRunId; + if (previousRunId === null || previousRunId === undefined || previousRunId === "") return; + runId = previousRunId; + } +} + +/** + * Build the worker + client pair every in-process test needs, and register an + * `onTestFinished` hook that replays the history of every execution the client + * started. + * + * The rig deliberately does NOT scope the task queue — callers keep calling + * `withTaskQueue` themselves. A same-workflow continue-as-new must land on the + * contract's static queue, because the contract is closed over inside the + * bundled workflow module and a test-side copy can never reach it. + * + * @public consumed from sibling packages' suites via the + * `@temporal-contract/testing/test-rig` subpath; the tsconfig `paths` + * indirection hides that usage from knip. + */ +export async function testRig( + testEnv: TestWorkflowEnvironment, + options: RigOptions, +): Promise<{ worker: TypedWorker; client: ContractClient }> { + const { contract, bundle, activities, replaySkipAllowlist = {} } = options; + + const worker = await TypedWorker.create({ + contract, + connection: testEnv.nativeConnection, + workflowBundle: bundle, + // Spread conditionally: `TypedWorker.create` distinguishes an absent + // `activities` key from `activities: undefined` (a workflow-only worker + // must not register an activity poller). + ...(activities !== undefined ? { activities } : {}), + }).get(); + + const typedClient = await TypedClient.create({ client: testEnv.client }).get(); + const bound = typedClient.for(contract); + + // Guards the Proxy's own load-bearing assumption below: every name in + // `START_METHODS` must resolve to an actual method on `bound`. A rename + // (or a fourth start method `START_METHODS` doesn't know about yet) would + // otherwise make the Proxy silently stop intercepting that method — + // `startedIds` stays empty, `onTestFinished` iterates nothing, and the + // whole tier goes green proving zero replay coverage. Thrown eagerly, at + // rig setup, rather than left to surface as a quiet coverage gap later. + for (const methodName of START_METHODS) { + if (typeof (bound as unknown as Record)[methodName] !== "function") { + // oxlint-disable-next-line unthrown/no-throw -- test-harness assertion: guards the rig's load-bearing assumption that every START_METHODS name is a real ContractClient method + throw new Error( + `testRig's START_METHODS names "${methodName}", but ContractClient has no such method. ` + + `Either the method was renamed or removed (update START_METHODS in ` + + `packages/testing/src/test-rig.ts to match), or this is a typo.`, + ); + } + } + + // A `Set`: a test that calls e.g. `signalWithStart` more than once against + // the same workflow ID must not queue the same replay twice. + const startedIds = new Set(); + + const client = new Proxy(bound, { + get(target, property, receiver) { + const value = Reflect.get(target, property, receiver) as unknown; + if (typeof property !== "string" || !START_METHODS.has(property)) return value; + if (typeof value !== "function") return value; + const methodName = property; + + return (...args: readonly unknown[]) => { + startedIds.add(extractStartedWorkflowId(methodName, args)); + return Reflect.apply(value as (...rest: readonly unknown[]) => unknown, target, args); + }; + }, + }) as ContractClient; + + onTestFinished(async () => { + for (const workflowId of startedIds) { + const handle = testEnv.client.workflow.getHandle(workflowId); + let described; + try { + described = await handle.describe(); + } catch (error) { + // A start call recorded this id, but the server never dispatched + // it — e.g. it failed contract validation before the RPC went out. + // Nothing was ever created, so there's nothing to replay. + if (isWorkflowNotFoundError(error)) continue; + // oxlint-disable-next-line unthrown/no-throw -- sanctioned re-raise: an unrecognized describe() failure must keep riding its original error, not be swallowed by this WorkflowNotFoundError-specific catch + throw error; + } + + if (!isTerminalStatus(described.status.name)) { + const reason = skipReasonFor(workflowId, replaySkipAllowlist); + if (reason === undefined) { + // oxlint-disable-next-line unthrown/no-throw -- test-harness assertion: onTestFinished has no Result seam, and Vitest surfaces test failures via throw + throw new Error( + `Workflow "${workflowId}" ended ${described.status.name}, so its history cannot be ` + + `replayed and this test proves nothing about replay determinism for it. Either make ` + + `the execution terminal, or add an entry with a reason to the "replaySkipAllowlist" ` + + `passed to this testRig(...) call.`, + ); + } + continue; + } + + await replayChain(testEnv.client, bundle, workflowId); + } + }); + + return { worker, client }; +} diff --git a/packages/testing/typedoc.json b/packages/testing/typedoc.json index ad763ab7..9aa2b76c 100644 --- a/packages/testing/typedoc.json +++ b/packages/testing/typedoc.json @@ -5,6 +5,7 @@ "src/contract.ts", "src/extension.ts", "src/global-setup.ts", + "src/test-rig.ts", "src/time-skipping.ts", "src/workflow-bundle.ts" ], diff --git a/packages/worker/src/__tests__/activity-options.inprocess.spec.ts b/packages/worker/src/__tests__/activity-options.inprocess.spec.ts index 0498fe95..bc91242f 100644 --- a/packages/worker/src/__tests__/activity-options.inprocess.spec.ts +++ b/packages/worker/src/__tests__/activity-options.inprocess.spec.ts @@ -1,4 +1,4 @@ -import { TypedClient } from "@temporal-contract/client"; +import { testRig } from "@temporal-contract/testing/test-rig"; import { it } from "@temporal-contract/testing/time-skipping"; import { bundleFor, @@ -11,7 +11,6 @@ import { ErrAsync, fromSafePromise, OkAsync } from "unthrown"; import { describe, expect } from "vitest"; import { declareActivitiesHandler } from "../activity.js"; -import { TypedWorker } from "../worker.js"; import { activityOptionsContract, layeredOptionsContract } from "./activity-options.contract.js"; /** @@ -48,15 +47,7 @@ describe("contract-level activityOptions reach Temporal", () => { }, }); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - activities, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle, activities }); const outcome = await worker.raw.runUntil(async () => { // `.getOrThrow()`, not `.get()`: both calls carry a real (non-`never`) @@ -115,15 +106,7 @@ describe("activityOptions merge precedence across layers", () => { }, }); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - activities, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle, activities }); const outcome = await worker.raw.runUntil(async () => { const handle = await client diff --git a/packages/worker/src/__tests__/cancellation.inprocess.spec.ts b/packages/worker/src/__tests__/cancellation.inprocess.spec.ts index 7db5a1ab..ed700bb3 100644 --- a/packages/worker/src/__tests__/cancellation.inprocess.spec.ts +++ b/packages/worker/src/__tests__/cancellation.inprocess.spec.ts @@ -1,8 +1,5 @@ -import { - TypedClient, - WORKFLOW_CANCELLED_ERROR_TAG, - WorkflowCancelledError, -} from "@temporal-contract/client"; +import { WORKFLOW_CANCELLED_ERROR_TAG, WorkflowCancelledError } from "@temporal-contract/client"; +import { testRig } from "@temporal-contract/testing/test-rig"; import { it } from "@temporal-contract/testing/time-skipping"; import { bundleFor, @@ -16,7 +13,6 @@ import { fromSafePromise } from "unthrown"; import { describe, expect } from "vitest"; import { ACTIVITY_CANCELLED_ERROR_TAG, declareActivitiesHandler } from "../activity.js"; -import { TypedWorker } from "../worker.js"; import { cancellationContract } from "./cancellation.contract.js"; import { inprocessContract } from "./inprocess.contract.js"; @@ -92,14 +88,7 @@ describe("cancellation against a real server", () => { const contract = withTaskQueue(inprocessContract, nextTaskQueueId("cancellation")); const bundle = await bundleFor(fixturePath(import.meta.url, "inprocess.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); const outcome = await worker.raw.runUntil(async () => { const handle = await client @@ -142,15 +131,7 @@ describe("cancellation against a real server", () => { activities: { slowActivity: cancellableSleep }, }); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - activities, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle, activities }); const outcome = await worker.raw.runUntil(async () => { const handle = await client @@ -182,15 +163,7 @@ describe("cancellation against a real server", () => { activities: { slowActivity: cancellableSleep }, }); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - activities, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle, activities }); const outcome = await worker.raw.runUntil(async () => { const handle = await client @@ -219,15 +192,7 @@ describe("cancellation against a real server", () => { activities: { slowActivity: cancellableSleep }, }); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - activities, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle, activities }); const outcome = await worker.raw.runUntil(async () => { const handle = await client @@ -262,14 +227,7 @@ describe("cancellableScope / nonCancellableScope mechanics against a real server const contract = withTaskQueue(cancellationContract, nextTaskQueueId("cancellation")); const bundle = await bundleFor(fixturePath(import.meta.url, "cancellation.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); const outcome = await worker.raw.runUntil(async () => client.executeWorkflow("scopeMechanics", { @@ -288,14 +246,7 @@ describe("cancellableScope / nonCancellableScope mechanics against a real server const contract = withTaskQueue(cancellationContract, nextTaskQueueId("cancellation")); const bundle = await bundleFor(fixturePath(import.meta.url, "cancellation.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); const outcome = await worker.raw.runUntil(async () => client.executeWorkflow("scopeMechanics", { @@ -320,14 +271,7 @@ describe("cancellableScope / nonCancellableScope mechanics against a real server const contract = withTaskQueue(cancellationContract, nextTaskQueueId("cancellation")); const bundle = await bundleFor(fixturePath(import.meta.url, "cancellation.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); const outcome = await worker.raw.runUntil(async () => client.executeWorkflow("scopeMechanics", { @@ -346,14 +290,7 @@ describe("cancellableScope / nonCancellableScope mechanics against a real server const contract = withTaskQueue(cancellationContract, nextTaskQueueId("cancellation")); const bundle = await bundleFor(fixturePath(import.meta.url, "cancellation.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); const outcome = await worker.raw.runUntil(async () => client.executeWorkflow("scopeMechanics", { @@ -375,14 +312,7 @@ describe("cancellableScope / nonCancellableScope mechanics against a real server const contract = withTaskQueue(cancellationContract, nextTaskQueueId("cancellation")); const bundle = await bundleFor(fixturePath(import.meta.url, "cancellation.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); const outcome = await worker.raw.runUntil(async () => client.executeWorkflow("scopeMechanics", { diff --git a/packages/worker/src/__tests__/child-wire.inprocess.spec.ts b/packages/worker/src/__tests__/child-wire.inprocess.spec.ts index 151f3fc0..60ee491e 100644 --- a/packages/worker/src/__tests__/child-wire.inprocess.spec.ts +++ b/packages/worker/src/__tests__/child-wire.inprocess.spec.ts @@ -1,4 +1,5 @@ -import { TypedClient, WorkflowFailedError } from "@temporal-contract/client"; +import { WorkflowFailedError } from "@temporal-contract/client"; +import { testRig } from "@temporal-contract/testing/test-rig"; import { it } from "@temporal-contract/testing/time-skipping"; import { bundleFor, @@ -9,7 +10,6 @@ import { import { ApplicationFailure } from "@temporalio/client"; import { describe, expect } from "vitest"; -import { TypedWorker } from "../worker.js"; import { childWireContract } from "./child-wire.contract.js"; /** @@ -48,14 +48,7 @@ describe("workflow entry point — wire format against a real server", () => { const contract = withTaskQueue(childWireContract, nextTaskQueueId("child-wire")); const bundle = await bundleFor(fixturePath(import.meta.url, "child-wire.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); const { raw, parsed } = await worker.raw.runUntil(async () => { const handle = await client @@ -96,14 +89,7 @@ describe("workflow entry point — wire format against a real server", () => { const contract = withTaskQueue(childWireContract, nextTaskQueueId("child-wire")); const bundle = await bundleFor(fixturePath(import.meta.url, "child-wire.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); const result = await worker.raw.runUntil(async () => { const handle = await client @@ -155,14 +141,7 @@ describe("child-workflow boundary — wire format against a real server", () => const contract = childWireContract; const bundle = await bundleFor(fixturePath(import.meta.url, "child-wire.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); const result = await worker.raw.runUntil(async () => { const handle = await client @@ -193,14 +172,7 @@ describe("child-workflow boundary — wire format against a real server", () => const contract = childWireContract; const bundle = await bundleFor(fixturePath(import.meta.url, "child-wire.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); const { parentResult, actualRunId } = await worker.raw.runUntil(async () => { const handle = await client @@ -250,14 +222,7 @@ describe("child-workflow boundary — wire format against a real server", () => const contract = childWireContract; const bundle = await bundleFor(fixturePath(import.meta.url, "child-wire.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); const { result, signalNames } = await worker.raw.runUntil(async () => { const handle = await client @@ -305,14 +270,7 @@ describe("child-workflow boundary — wire format against a real server", () => const contract = childWireContract; const bundle = await bundleFor(fixturePath(import.meta.url, "child-wire.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); const { result, signalNames } = await worker.raw.runUntil(async () => { const handle = await client diff --git a/packages/worker/src/__tests__/continue-as-new.inprocess.spec.ts b/packages/worker/src/__tests__/continue-as-new.inprocess.spec.ts index 253da42e..0eeb5ee8 100644 --- a/packages/worker/src/__tests__/continue-as-new.inprocess.spec.ts +++ b/packages/worker/src/__tests__/continue-as-new.inprocess.spec.ts @@ -1,4 +1,5 @@ -import { TypedClient, WorkflowFailedError } from "@temporal-contract/client"; +import { WorkflowFailedError } from "@temporal-contract/client"; +import { testRig } from "@temporal-contract/testing/test-rig"; import { it } from "@temporal-contract/testing/time-skipping"; import { bundleFor, @@ -9,7 +10,6 @@ import { import { ApplicationFailure } from "@temporalio/client"; import { describe, expect } from "vitest"; -import { TypedWorker } from "../worker.js"; import { continueAsNewContract } from "./continue-as-new.contract.js"; /** @@ -67,14 +67,7 @@ describe("continue-as-new against a real server", () => { const contract = continueAsNewContract; const bundle = await bundleFor(fixturePath(import.meta.url, "continue-as-new.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); const { total, memo } = await worker.raw.runUntil(async () => { const handle = await client @@ -107,14 +100,7 @@ describe("continue-as-new against a real server", () => { const contract = continueAsNewContract; const bundle = await bundleFor(fixturePath(import.meta.url, "continue-as-new.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); const total = await worker.raw.runUntil(async () => { const handle = await client @@ -146,14 +132,7 @@ describe("continue-as-new against a real server", () => { const contract = continueAsNewContract; const bundle = await bundleFor(fixturePath(import.meta.url, "continue-as-new.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); const { result, firstExecutionRunId, latestRunId } = await worker.raw.runUntil(async () => { const handle = await client @@ -205,14 +184,7 @@ describe("continue-as-new against a real server", () => { const contract = continueAsNewContract; const bundle = await bundleFor(fixturePath(import.meta.url, "continue-as-new.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); const result = await worker.raw.runUntil(async () => { const handle = await client @@ -241,14 +213,7 @@ describe("continue-as-new against a real server", () => { const contract = withTaskQueue(continueAsNewContract, queueId); const bundle = await bundleFor(fixturePath(import.meta.url, "continue-as-new.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); const raw = await worker.raw.runUntil(async () => { const handle = await client @@ -281,14 +246,7 @@ describe("continue-as-new against a real server", () => { const contract = withTaskQueue(continueAsNewContract, queueId); const bundle = await bundleFor(fixturePath(import.meta.url, "continue-as-new.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); const { result, firstExecutionRunId, latestRunId } = await worker.raw.runUntil(async () => { const handle = await client @@ -339,14 +297,7 @@ describe("continue-as-new against a real server", () => { const contract = withTaskQueue(continueAsNewContract, queueId); const bundle = await bundleFor(fixturePath(import.meta.url, "continue-as-new.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); const result = await worker.raw.runUntil(async () => { const handle = await client @@ -378,14 +329,7 @@ describe("continue-as-new against a real server", () => { const contract = continueAsNewContract; const bundle = await bundleFor(fixturePath(import.meta.url, "continue-as-new.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); const result = await worker.raw.runUntil(async () => { const handle = await client @@ -417,14 +361,7 @@ describe("continue-as-new against a real server", () => { const contract = continueAsNewContract; const bundle = await bundleFor(fixturePath(import.meta.url, "continue-as-new.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); const result = await worker.raw.runUntil(async () => { const handle = await client diff --git a/packages/worker/src/__tests__/handlers-replay-skip-allowlist.ts b/packages/worker/src/__tests__/handlers-replay-skip-allowlist.ts new file mode 100644 index 00000000..24c46a20 --- /dev/null +++ b/packages/worker/src/__tests__/handlers-replay-skip-allowlist.ts @@ -0,0 +1,38 @@ +/** + * `testRig`'s `replaySkipAllowlist` for `handlers.inprocess.spec.ts` — + * workflow-ID prefixes whose executions are deliberately left non-terminal, + * so their histories cannot be replayed. Every entry needs a reason. + * + * `@temporal-contract/testing/test-rig` no longer bakes fixture IDs from this + * repo into the published rig (an external consumer's workflow IDs could + * collide with them and be silently skipped) — `replaySkipAllowlist` is a + * caller-supplied `RigOptions` field instead, defaulting to `{}`. This is + * that caller's list, kept as a fixture next to the spec that owns the + * workflows it names; it is not exported from the package. + * + * This list may only ever shrink. A silently-skipped execution would report + * replay coverage it does not have — exactly the rot `testRig` exists to + * prevent — so an unlisted non-terminal execution fails the test instead. + */ +export const HANDLERS_REPLAY_SKIP_ALLOWLIST: Readonly> = { + "handlers-probe-edge-cases": + "handlers.workflows.ts's probeEdgeCases blocks on condition(() => false) and is never " + + "signaled to finish — the spec only issues queries against it, so its execution is " + + "deliberately left running forever.", + // Three exact IDs (not a shared prefix): each is a static literal, not + // counter-suffixed, so a narrower key here means a future + // "handlers-wire-*" workflow that hangs by accident isn't silently + // swept into this entry too. + "handlers-wire-signal": + "handlers.workflows.ts's transformWorkflow blocks on condition(() => false) and is never " + + "signaled to finish — this spec only signals against it directly, so its execution is " + + "deliberately left running forever.", + "handlers-wire-query": + "handlers.workflows.ts's transformWorkflow blocks on condition(() => false) and is never " + + "signaled to finish — this spec only queries it directly, so its execution is " + + "deliberately left running forever.", + "handlers-wire-update": + "handlers.workflows.ts's transformWorkflow blocks on condition(() => false) and is never " + + "signaled to finish — this spec only updates it directly, so its execution is " + + "deliberately left running forever.", +}; diff --git a/packages/worker/src/__tests__/handlers.inprocess.spec.ts b/packages/worker/src/__tests__/handlers.inprocess.spec.ts index 7eda545d..b9afb2e8 100644 --- a/packages/worker/src/__tests__/handlers.inprocess.spec.ts +++ b/packages/worker/src/__tests__/handlers.inprocess.spec.ts @@ -1,4 +1,5 @@ -import { TypedClient, WorkflowFailedError } from "@temporal-contract/client"; +import { WorkflowFailedError } from "@temporal-contract/client"; +import { testRig } from "@temporal-contract/testing/test-rig"; import { it } from "@temporal-contract/testing/time-skipping"; import { bundleFor, @@ -15,7 +16,7 @@ import { import { Runtime, type Logger } from "@temporalio/worker"; import { describe, expect } from "vitest"; -import { TypedWorker } from "../worker.js"; +import { HANDLERS_REPLAY_SKIP_ALLOWLIST } from "./handlers-replay-skip-allowlist.js"; import { handlersContract } from "./handlers.contract.js"; /** @@ -96,14 +97,7 @@ describe("handler binding against a real server", () => { const { logs: capturedLogs, restore: restoreLogger } = captureWorkflowLogs(); try { - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); const total = await worker.raw.runUntil(async () => { const handle = await client @@ -153,14 +147,7 @@ describe("handler binding against a real server", () => { const contract = withTaskQueue(handlersContract, nextTaskQueueId("handlers")); const bundle = await bundleFor(fixturePath(import.meta.url, "handlers.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); await worker.raw.runUntil(async () => { const handle = await client @@ -185,14 +172,7 @@ describe("handler binding against a real server", () => { const contract = withTaskQueue(handlersContract, nextTaskQueueId("handlers")); const bundle = await bundleFor(fixturePath(import.meta.url, "handlers.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); await worker.raw.runUntil(async () => { const handle = await client @@ -233,14 +213,7 @@ describe("handler binding against a real server", () => { const contract = withTaskQueue(handlersContract, nextTaskQueueId("handlers")); const bundle = await bundleFor(fixturePath(import.meta.url, "handlers.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); await worker.raw.runUntil(async () => { const handle = await client @@ -274,14 +247,7 @@ describe("handler binding against a real server", () => { const contract = withTaskQueue(handlersContract, nextTaskQueueId("handlers")); const bundle = await bundleFor(fixturePath(import.meta.url, "handlers.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); await worker.raw.runUntil(async () => { const handle = await client @@ -304,14 +270,7 @@ describe("handler binding against a real server", () => { const contract = withTaskQueue(handlersContract, nextTaskQueueId("handlers")); const bundle = await bundleFor(fixturePath(import.meta.url, "handlers.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); await worker.raw.runUntil(async () => { const handle = await client @@ -383,14 +342,7 @@ describe("handler binding against a real server", () => { const contract = withTaskQueue(handlersContract, nextTaskQueueId("handlers")); const bundle = await bundleFor(fixturePath(import.meta.url, "handlers.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); await worker.raw.runUntil(async () => { const handle = await client @@ -428,14 +380,7 @@ describe("handler binding against a real server", () => { const contract = withTaskQueue(handlersContract, nextTaskQueueId("handlers")); const bundle = await bundleFor(fixturePath(import.meta.url, "handlers.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); await worker.raw.runUntil(async () => { const handle = await client @@ -464,14 +409,7 @@ describe("handler binding against a real server", () => { const contract = withTaskQueue(handlersContract, nextTaskQueueId("handlers")); const bundle = await bundleFor(fixturePath(import.meta.url, "handlers.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); await worker.raw.runUntil(async () => { const handle = await client @@ -508,14 +446,7 @@ describe("handler binding against a real server", () => { const contract = withTaskQueue(handlersContract, nextTaskQueueId("handlers")); const bundle = await bundleFor(fixturePath(import.meta.url, "handlers.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); await worker.raw.runUntil(async () => { const handle = await client @@ -548,14 +479,7 @@ describe("handler binding against a real server", () => { const contract = withTaskQueue(handlersContract, nextTaskQueueId("handlers")); const bundle = await bundleFor(fixturePath(import.meta.url, "handlers.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle }); await worker.raw.runUntil(async () => { const handle = await client @@ -590,14 +514,13 @@ describe("handler binding against a real server", () => { const contract = withTaskQueue(handlersContract, nextTaskQueueId("handlers")); const bundle = await bundleFor(fixturePath(import.meta.url, "handlers.workflows")); - const worker = await TypedWorker.create({ + const { worker, client } = await testRig(testEnv, { contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + bundle, + // `probeEdgeCases` deliberately never finishes — see + // `handlers-replay-skip-allowlist.ts`. + replaySkipAllowlist: HANDLERS_REPLAY_SKIP_ALLOWLIST, + }); await worker.raw.runUntil(async () => { const handle = await client @@ -652,14 +575,13 @@ describe("handler binding against a real server", () => { const contract = withTaskQueue(handlersContract, nextTaskQueueId("handlers")); const bundle = await bundleFor(fixturePath(import.meta.url, "handlers.workflows")); - const worker = await TypedWorker.create({ + const { worker, client } = await testRig(testEnv, { contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + bundle, + // `transformWorkflow` deliberately never finishes — see + // `handlers-replay-skip-allowlist.ts`. + replaySkipAllowlist: HANDLERS_REPLAY_SKIP_ALLOWLIST, + }); await worker.raw.runUntil(async () => { const handle = await client @@ -682,14 +604,13 @@ describe("handler binding against a real server", () => { const contract = withTaskQueue(handlersContract, nextTaskQueueId("handlers")); const bundle = await bundleFor(fixturePath(import.meta.url, "handlers.workflows")); - const worker = await TypedWorker.create({ + const { worker, client } = await testRig(testEnv, { contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + bundle, + // `transformWorkflow` deliberately never finishes — see + // `handlers-replay-skip-allowlist.ts`. + replaySkipAllowlist: HANDLERS_REPLAY_SKIP_ALLOWLIST, + }); await worker.raw.runUntil(async () => { const handle = await client @@ -719,14 +640,13 @@ describe("handler binding against a real server", () => { const contract = withTaskQueue(handlersContract, nextTaskQueueId("handlers")); const bundle = await bundleFor(fixturePath(import.meta.url, "handlers.workflows")); - const worker = await TypedWorker.create({ + const { worker, client } = await testRig(testEnv, { contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + bundle, + // `transformWorkflow` deliberately never finishes — see + // `handlers-replay-skip-allowlist.ts`. + replaySkipAllowlist: HANDLERS_REPLAY_SKIP_ALLOWLIST, + }); await worker.raw.runUntil(async () => { const handle = await client diff --git a/packages/worker/src/__tests__/rehydration.inprocess.spec.ts b/packages/worker/src/__tests__/rehydration.inprocess.spec.ts index 4408826c..00a30a09 100644 --- a/packages/worker/src/__tests__/rehydration.inprocess.spec.ts +++ b/packages/worker/src/__tests__/rehydration.inprocess.spec.ts @@ -1,5 +1,6 @@ import { ContractError, TypedClient, WorkflowFailedError } from "@temporal-contract/client"; import { onRehydrationMiss, type RehydrationMiss } from "@temporal-contract/contract/errors"; +import { testRig } from "@temporal-contract/testing/test-rig"; import { it } from "@temporal-contract/testing/time-skipping"; import { bundleFor, @@ -40,7 +41,6 @@ import { OkAsync, ErrAsync } from "unthrown"; import { describe, expect } from "vitest"; import { ApplicationFailure, declareActivitiesHandler } from "../activity.js"; -import { TypedWorker } from "../worker.js"; import { rehydrationClientContract, rehydrationWorkerContract } from "./rehydration.contract.js"; const activities = declareActivitiesHandler({ @@ -78,15 +78,7 @@ describe("rehydration at the e2e boundary", () => { const contract = withTaskQueue(rehydrationWorkerContract, nextTaskQueueId("rehydration")); const bundle = await bundleFor(fixturePath(import.meta.url, "rehydration.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - activities, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle, activities }); await worker.raw.runUntil(async () => { // Control: the typed constructor's failure carries the marker and DOES @@ -115,15 +107,17 @@ describe("rehydration at the e2e boundary", () => { const clientContract = withTaskQueue(rehydrationClientContract, queueId); const bundle = await bundleFor(fixturePath(import.meta.url, "rehydration.workflows")); - const worker = await TypedWorker.create({ + // Two contract views of the same worker (worker-side and client-side + // schema skew), so `testRig` — which binds one client to one contract — + // only covers the worker-side half; `skewedClient` is a second, + // independent `TypedClient` bound to `clientContract` on top of it. + const { worker, client: workerSideClient } = await testRig(testEnv, { contract: workerContract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, + bundle, activities, - }).get(); + }); const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const workerSideClient = typedClient.for(workerContract); const skewedClient = typedClient.for(clientContract); const misses: RehydrationMiss[] = []; @@ -189,15 +183,7 @@ describe("declareWorkflow — contract-error conversion", () => { const contract = withTaskQueue(rehydrationWorkerContract, nextTaskQueueId("rehydration")); const bundle = await bundleFor(fixturePath(import.meta.url, "rehydration.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - activities, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle, activities }); await worker.raw.runUntil(async () => { const result = await client.executeWorkflow("quote", { @@ -226,15 +212,7 @@ describe("declareWorkflow — contract-error conversion", () => { const contract = withTaskQueue(rehydrationWorkerContract, nextTaskQueueId("rehydration")); const bundle = await bundleFor(fixturePath(import.meta.url, "rehydration.workflows")); - const worker = await TypedWorker.create({ - contract, - connection: testEnv.nativeConnection, - workflowBundle: bundle, - activities, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(contract); + const { worker, client } = await testRig(testEnv, { contract, bundle, activities }); await worker.raw.runUntil(async () => { const result = await client.executeWorkflow("quote", { diff --git a/packages/worker/src/__tests__/replay.inprocess.spec.ts b/packages/worker/src/__tests__/replay.inprocess.spec.ts index c79bc065..a8f15fbc 100644 --- a/packages/worker/src/__tests__/replay.inprocess.spec.ts +++ b/packages/worker/src/__tests__/replay.inprocess.spec.ts @@ -1,7 +1,6 @@ -import { TypedClient } from "@temporal-contract/client"; +import { testRig } from "@temporal-contract/testing/test-rig"; import { it } from "@temporal-contract/testing/time-skipping"; -import { fixturePath } from "@temporal-contract/testing/workflow-bundle"; -import { Worker } from "@temporalio/worker"; +import { bundleFor, fixturePath } from "@temporal-contract/testing/workflow-bundle"; import { OkAsync, ErrAsync } from "unthrown"; /** * Replay-determinism coverage for the Result/AsyncResult machinery inside @@ -17,14 +16,13 @@ import { OkAsync, ErrAsync } from "unthrown"; * * This spec runs the full pipeline to completion — one happy path and one * typed-activity-error path (ContractError → ApplicationFailure wire shape → - * rehydration inside the workflow) — fetches each execution's history, and - * replays it with `Worker.runReplayHistory`, which rejects on any - * determinism violation. + * rehydration inside the workflow). `testRig`'s `onTestFinished` hook then + * fetches each started execution's history and replays it with + * `Worker.runReplayHistory`, which rejects on any determinism violation. */ import { describe, expect } from "vitest"; import { declareActivitiesHandler } from "../activity.js"; -import { TypedWorker } from "../worker.js"; import { inprocessContract } from "./inprocess.contract.js"; const activities = declareActivitiesHandler({ @@ -45,15 +43,11 @@ describe("declareWorkflow replay determinism", () => { it("replays histories produced by Result-shaped workflows without determinism violations", async ({ testEnv, }) => { - const worker = await TypedWorker.create({ + const { worker, client } = await testRig(testEnv, { contract: inprocessContract, - connection: testEnv.nativeConnection, - workflowsPath: fixturePath(import.meta.url, "inprocess.workflows"), + bundle: await bundleFor(fixturePath(import.meta.url, "inprocess.workflows")), activities, - }).get(); - - const typedClient = await TypedClient.create({ client: testEnv.client }).get(); - const client = typedClient.for(inprocessContract); + }); const happyId = "replay-happy"; const declinedId = "replay-declined"; @@ -77,19 +71,5 @@ describe("declareWorkflow replay determinism", () => { }); expect(declined).toBeOkWith({ status: "declined:negative-amount" }); }); - - // Replay both histories against the same workflow code. - // `runReplayHistory` rejects with a DeterminismViolationError (or - // ReplayError) if the unthrown machinery diverges on replay. - for (const workflowId of [happyId, declinedId]) { - const history = await testEnv.client.workflow.getHandle(workflowId).fetchHistory(); - await expect( - Worker.runReplayHistory( - { workflowsPath: fixturePath(import.meta.url, "inprocess.workflows") }, - history, - workflowId, - ), - ).resolves.toBeUndefined(); - } }); }); diff --git a/packages/worker/src/__tests__/retry.contract.ts b/packages/worker/src/__tests__/retry.contract.ts new file mode 100644 index 00000000..7b0d08e6 --- /dev/null +++ b/packages/worker/src/__tests__/retry.contract.ts @@ -0,0 +1,38 @@ +import { defineActivity, defineContract, defineWorkflow } from "@temporal-contract/contract"; +import { z } from "zod"; + +// Composition-first: resources defined individually, then composed. + +/** + * Two declared errors identical but for `nonRetryable`. Each `retry.inprocess.spec.ts` + * test supplies its own activity handler that hardcodes which one to raise — + * `mode` only threads client → workflow → activity input, unread by the + * handler itself — so a single fixture proves both directions and the only + * variable between them is the flag under test. + * + * `maximumAttempts: 3` bounds the retryable case: it must retry (proving the + * flag reached Temporal) without retrying forever if the flag regresses. + */ +const flaky = defineActivity({ + input: z.object({ mode: z.enum(["terminal", "retryable"]) }), + output: z.object({ attempts: z.number() }), + errors: { + TerminalFailure: { data: z.object({ at: z.number() }), nonRetryable: true }, + RetryableFailure: { data: z.object({ at: z.number() }), nonRetryable: false }, + }, + activityOptions: { + startToCloseTimeout: "10 seconds", + retry: { maximumAttempts: 3, backoffCoefficient: 1, initialInterval: "1 second" }, + }, +}); + +const runsFlaky = defineWorkflow({ + input: z.object({ mode: z.enum(["terminal", "retryable"]) }), + output: z.object({ outcome: z.string(), attempts: z.number() }), + activities: { flaky }, +}); + +export const retryContract = defineContract({ + taskQueue: "retry-tests", + workflows: { runsFlaky }, +}); diff --git a/packages/worker/src/__tests__/retry.inprocess.spec.ts b/packages/worker/src/__tests__/retry.inprocess.spec.ts new file mode 100644 index 00000000..5798b709 --- /dev/null +++ b/packages/worker/src/__tests__/retry.inprocess.spec.ts @@ -0,0 +1,84 @@ +import { testRig } from "@temporal-contract/testing/test-rig"; +import { it } from "@temporal-contract/testing/time-skipping"; +import { + bundleFor, + fixturePath, + nextTaskQueueId, + withTaskQueue, +} from "@temporal-contract/testing/workflow-bundle"; +import { Context } from "@temporalio/activity"; +import { ErrAsync } from "unthrown"; +import { describe, expect } from "vitest"; + +import { declareActivitiesHandler } from "../activity.js"; +import { retryContract } from "./retry.contract.js"; + +const WORKFLOW_EXECUTION_TIMEOUT = "30 seconds"; + +describe("nonRetryable is a behavior, not a field", () => { + it("stops after exactly one attempt when the declared error is nonRetryable", async ({ + testEnv, + }) => { + const contract = withTaskQueue(retryContract, nextTaskQueueId("retry-terminal")); + const bundle = await bundleFor(fixturePath(import.meta.url, "retry.workflows")); + + const activities = declareActivitiesHandler({ + contract, + activities: { + runsFlaky: { + // Always fails. `Context.current().info.attempt` is Temporal's own + // attempt counter, so the assertion reads what the server did. + flaky: (_input, { errors }) => + ErrAsync(errors.TerminalFailure({ at: Context.current().info.attempt })), + }, + }, + }); + + const { worker, client } = await testRig(testEnv, { contract, bundle, activities }); + + const result = await worker.raw.runUntil( + client + .executeWorkflow("runsFlaky", { + workflowId: "retry-terminal", + args: { mode: "terminal" }, + workflowExecutionTimeout: WORKFLOW_EXECUTION_TIMEOUT, + }) + .getOrThrow(), + ); + + // The whole point: attempt 1 and no more. A regression that dropped + // `nonRetryable` on the wire would let Temporal retry to 3. + expect(result).toEqual({ outcome: "err:TerminalFailure", attempts: 1 }); + }); + + it("retries when the declared error is retryable", async ({ testEnv }) => { + const contract = withTaskQueue(retryContract, nextTaskQueueId("retry-retryable")); + const bundle = await bundleFor(fixturePath(import.meta.url, "retry.workflows")); + + const activities = declareActivitiesHandler({ + contract, + activities: { + runsFlaky: { + flaky: (_input, { errors }) => + ErrAsync(errors.RetryableFailure({ at: Context.current().info.attempt })), + }, + }, + }); + + const { worker, client } = await testRig(testEnv, { contract, bundle, activities }); + + const result = await worker.raw.runUntil( + client + .executeWorkflow("runsFlaky", { + workflowId: "retry-retryable", + args: { mode: "retryable" }, + workflowExecutionTimeout: WORKFLOW_EXECUTION_TIMEOUT, + }) + .getOrThrow(), + ); + + // Exhausts the declared `maximumAttempts: 3`, so the surfaced failure is + // from attempt 3 — proving Temporal really retried. + expect(result).toEqual({ outcome: "err:RetryableFailure", attempts: 3 }); + }); +}); diff --git a/packages/worker/src/__tests__/retry.workflows.ts b/packages/worker/src/__tests__/retry.workflows.ts new file mode 100644 index 00000000..5e8b425f --- /dev/null +++ b/packages/worker/src/__tests__/retry.workflows.ts @@ -0,0 +1,25 @@ +import { ContractError, declareWorkflow } from "../workflow.js"; +import { retryContract } from "./retry.contract.js"; + +export const runsFlaky = declareWorkflow({ + workflowName: "runsFlaky", + contract: retryContract, + implementation: async (context, args) => { + const result = await context.activities.flaky({ mode: args.mode }); + + // Fold the failure into a returned status rather than rethrowing: a + // rethrown defect becomes a Workflow-Task retry loop that time-skipping + // cannot fast-forward past, turning a regression into a 120s hang. + if (result.isDefect()) return { outcome: `defect:${String(result.cause)}`, attempts: -1 }; + if (result.isErr()) { + const error = result.error; + if (error instanceof ContractError) { + // `data.at` carries the attempt number the activity failed on. + const at = (error.data as { at: number }).at; + return { outcome: `err:${error.errorName}`, attempts: at }; + } + return { outcome: `err:${error.name}`, attempts: -1 }; + } + return { outcome: "ok", attempts: result.value.attempts }; + }, +}); diff --git a/packages/worker/src/__tests__/time-skipping.inprocess.spec.ts b/packages/worker/src/__tests__/time-skipping.inprocess.spec.ts index f9fa83e6..9edd2dba 100644 --- a/packages/worker/src/__tests__/time-skipping.inprocess.spec.ts +++ b/packages/worker/src/__tests__/time-skipping.inprocess.spec.ts @@ -1,6 +1,7 @@ import { ContractError, TypedClient, type ClientInterceptor } from "@temporal-contract/client"; +import { testRig } from "@temporal-contract/testing/test-rig"; import { it } from "@temporal-contract/testing/time-skipping"; -import { fixturePath } from "@temporal-contract/testing/workflow-bundle"; +import { bundleFor, fixturePath } from "@temporal-contract/testing/workflow-bundle"; import { OkAsync, ErrAsync } from "unthrown"; /** * Full contract-pipeline coverage against the time-skipping @@ -123,23 +124,20 @@ describe("time-skipping TestWorkflowEnvironment", () => { ]); }); + // The one test below that uses testRig — the other four in this file build + // TypedWorker.create / TypedClient.create by hand instead: two assert on + // the creation Result itself (Ok/Defect), which testRig's `.get()`-unwrapping + // helper hides, and two need `interceptors` / arbitrary WorkerOptions + // passthrough that RigOptions doesn't expose. it("a workflow that re-raises cancellation via rethrowCancellation ends Cancelled", async ({ testEnv, }) => { - const workerResult = await TypedWorker.create({ + const bundle = await bundleFor(fixturePath(import.meta.url, "inprocess.workflows")); + const { worker, client } = await testRig(testEnv, { contract: inprocessContract, - connection: testEnv.nativeConnection, - workflowsPath: fixturePath(import.meta.url, "inprocess.workflows"), + bundle, activities, }); - expect(workerResult).toBeOk(); - if (!workerResult.isOk()) return; - const worker = workerResult.value; - - const clientResult = await TypedClient.create({ client: testEnv.client }); - expect(clientResult).toBeOk(); - if (!clientResult.isOk()) return; - const client = clientResult.value.for(inprocessContract); await worker.raw.runUntil(async () => { const workflowId = "inprocess-cancelled-outcome"; diff --git a/packages/worker/src/__tests__/timeouts.contract.ts b/packages/worker/src/__tests__/timeouts.contract.ts new file mode 100644 index 00000000..e6e69dee --- /dev/null +++ b/packages/worker/src/__tests__/timeouts.contract.ts @@ -0,0 +1,39 @@ +import { defineActivity, defineContract, defineWorkflow } from "@temporal-contract/contract"; +import { z } from "zod"; + +/** + * The activity reports back the timeouts Temporal actually materialized for + * its task, so each layer of the merge is proven by effect rather than by + * inspecting the options we passed in. + * + * `heartbeatTimeout` is the one this repo asserted nowhere before — a lost + * heartbeat timeout means a dead activity is never retried. + */ +const reportsTimeouts = defineActivity({ + input: z.object({}), + output: z.object({ + startToCloseMs: z.number(), + scheduleToCloseMs: z.number(), + heartbeatMs: z.number(), + }), + // Contract layer: heartbeat declared here and nowhere else. + activityOptions: { + heartbeatTimeout: "7 seconds", + retry: { maximumAttempts: 1 }, + }, +}); + +const reportsLayered = defineWorkflow({ + input: z.object({}), + output: z.object({ + startToCloseMs: z.number(), + scheduleToCloseMs: z.number(), + heartbeatMs: z.number(), + }), + activities: { reportsTimeouts }, +}); + +export const timeoutsContract = defineContract({ + taskQueue: "timeouts-tests", + workflows: { reportsLayered }, +}); diff --git a/packages/worker/src/__tests__/timeouts.inprocess.spec.ts b/packages/worker/src/__tests__/timeouts.inprocess.spec.ts new file mode 100644 index 00000000..566b58eb --- /dev/null +++ b/packages/worker/src/__tests__/timeouts.inprocess.spec.ts @@ -0,0 +1,62 @@ +import { testRig } from "@temporal-contract/testing/test-rig"; +import { it } from "@temporal-contract/testing/time-skipping"; +import { + bundleFor, + fixturePath, + nextTaskQueueId, + withTaskQueue, +} from "@temporal-contract/testing/workflow-bundle"; +import { Context } from "@temporalio/activity"; +import { OkAsync } from "unthrown"; +import { describe, expect } from "vitest"; + +import { declareActivitiesHandler } from "../activity.js"; +import { timeoutsContract } from "./timeouts.contract.js"; + +const WORKFLOW_EXECUTION_TIMEOUT = "30 seconds"; + +describe("activity timeouts reach Temporal through every merge layer", () => { + it("materializes the value each layer contributed", async ({ testEnv }) => { + const contract = withTaskQueue(timeoutsContract, nextTaskQueueId("timeouts")); + const bundle = await bundleFor(fixturePath(import.meta.url, "timeouts.workflows")); + + const activities = declareActivitiesHandler({ + contract, + activities: { + reportsLayered: { + // Reads what Temporal actually scheduled the task with, rather than + // what we passed in — an effect, not a call shape. + reportsTimeouts: () => { + const info = Context.current().info; + return OkAsync({ + startToCloseMs: info.startToCloseTimeoutMs, + scheduleToCloseMs: info.scheduleToCloseTimeoutMs, + // `heartbeatTimeoutMs` is optional on ActivityInfo — 0 + // distinguishes "declared but lost in the merge" from "never + // declared", so the assertion below fails loudly either way. + heartbeatMs: info.heartbeatTimeoutMs ?? 0, + }); + }, + }, + }, + }); + + const { worker, client } = await testRig(testEnv, { contract, bundle, activities }); + + const result = await worker.raw.runUntil( + client + .executeWorkflow("reportsLayered", { + workflowId: "timeouts-layered", + args: {}, + workflowExecutionTimeout: WORKFLOW_EXECUTION_TIMEOUT, + }) + .getOrThrow(), + ); + + expect(result).toEqual({ + startToCloseMs: 9_000, // contributed by activityOptionsByName + scheduleToCloseMs: 20_000, // declareWorkflow's workflow-wide default + heartbeatMs: 7_000, // contract-level, the layer with no competitor + }); + }); +}); diff --git a/packages/worker/src/__tests__/timeouts.workflows.ts b/packages/worker/src/__tests__/timeouts.workflows.ts new file mode 100644 index 00000000..a15ae1ac --- /dev/null +++ b/packages/worker/src/__tests__/timeouts.workflows.ts @@ -0,0 +1,22 @@ +import { declareWorkflow } from "../workflow.js"; +import { timeoutsContract } from "./timeouts.contract.js"; + +export const reportsLayered = declareWorkflow({ + workflowName: "reportsLayered", + contract: timeoutsContract, + // Workflow-wide layer. + activityOptions: { scheduleToCloseTimeout: "20 seconds" }, + // Per-activity layer — contributes startToClose. The three keys asserted + // in timeouts.inprocess.spec.ts are disjoint across all three layers, so + // this fixture proves each layer's value reaches Temporal (forwarding), + // not precedence between layers — that's covered separately by + // activity-options.contract.ts. + activityOptionsByName: { reportsTimeouts: { startToCloseTimeout: "9 seconds" } }, + implementation: async (context) => { + // `reportsTimeouts` declares no contract `errors`, so the workflow-side + // proxy is the plain-throwing wrapper (matching Temporal's native + // behavior) rather than an AsyncResult — there is no `.isErr()`/ + // `.isDefect()` to narrow here. + return await context.activities.reportsTimeouts({}); + }, +}); diff --git a/packages/worker/tsconfig.json b/packages/worker/tsconfig.json index b3a55006..458535db 100644 --- a/packages/worker/tsconfig.json +++ b/packages/worker/tsconfig.json @@ -2,7 +2,43 @@ "extends": "@btravstack/tsconfig/base.json", "compilerOptions": { "outDir": "./dist", - "rootDir": "./src" + // `rootDir` spans the packages dir: the `paths` below intentionally pull + // sibling sources into the program, which a src-scoped rootDir would + // reject (TS6059). The config never emits (base sets `noEmit`), so this + // only satisfies the compiler's emit-layout validation. Mirrors + // `packages/testing/tsconfig.json`'s identical setup. + "rootDir": "..", + // `@temporal-contract/client`, `@temporal-contract/contract`, and + // `@temporal-contract/worker/activity` + `/worker` are `peerDependencies` + // (not `devDependencies`) of `@temporal-contract/testing` — see the long + // comment on `workspaceAliases` in `vitest.config.ts` for why. That + // alias only fixes Vite/Vitest's *runtime* module resolution; `tsc` + // doesn't read `vitest.config.ts` at all. Left unfixed here, `tsc` + // silently fails to resolve these bare specifiers when they're imported + // from inside `@temporal-contract/testing`'s built `test-rig.d.mts` + // (Node module resolution walks up from *that file's* directory, which + // has no route to peer-only packages) — and because `skipLibCheck` is + // `true` (below, inherited from the base config), that failure never + // surfaces as a diagnostic. It just silently collapses `testRig`'s + // `ContractClient` and `TypedWorker` types to `any`, which + // then only becomes visible far downstream, as a `noImplicitAny` error + // on some unrelated callback parameter (see the + // `events.filter((event) => ...)` calls in `handlers.inprocess.spec.ts` + // for a concrete example) or, worse, as NO error at all when the `any` + // flows into a zero-argument callback (`worker.raw.runUntil(async () => + // {...})`), silently un-typechecking every property access on its + // result. These four `paths` entries are `tsc`'s equivalent of that Vite + // alias, mapped to sibling *source* exactly like + // `packages/testing/tsconfig.json` already does for the same peer-dep + // cycle — not to built `dist`, which would coincidentally cover 3 of the + // 4 specifiers actually imported by `test-rig.d.mts` and miss the + // fourth. + "paths": { + "@temporal-contract/client": ["../client/src/index.ts"], + "@temporal-contract/contract": ["../contract/src/index.ts"], + "@temporal-contract/worker/activity": ["./src/activity.ts"], + "@temporal-contract/worker/worker": ["./src/worker.ts"] + } }, "include": ["src/**/*"], "exclude": ["node_modules", "dist"] diff --git a/packages/worker/vitest.config.ts b/packages/worker/vitest.config.ts index 061ad987..6ab7216b 100644 --- a/packages/worker/vitest.config.ts +++ b/packages/worker/vitest.config.ts @@ -1,5 +1,59 @@ +import { fileURLToPath } from "node:url"; + import { defineConfig } from "vitest/config"; +const sibling = (path: string) => fileURLToPath(new URL(path, import.meta.url)); + +// `@temporal-contract/testing`'s built `test-rig.mjs` imports `TypedClient` +// from `@temporal-contract/client` and `TypedWorker` from +// `@temporal-contract/worker/worker` at the top level. Both are +// `peerDependencies` of `testing`, not `devDependencies` — a devDependency +// would create a real cycle (`client` already devDepends on `testing`), +// which would break turbo's package-graph ordering. Peer specs resolve fine +// for a real published consumer (whose own node_modules sits above the dist +// file), but inside this workspace pnpm symlinks `@temporal-contract/testing` +// straight to `packages/testing`, and Node resolves bare specifiers from +// that real path — which has no route to `client`/`worker` in its own +// `node_modules`. Tried `dependenciesMeta.injected: true` on `testing` in +// this package's `package.json` first (hard-links `testing` into this +// package's `node_modules` instead), but it made `pnpm install` fail: +// resolving `testing`'s peer on `@temporal-contract/worker` (this very +// package, which — being a peer, not a workspace dependency — doesn't +// self-reference) fell through to the registry and tripped +// `minimumReleaseAge`'s supply-chain-maturity gate +// (`[ERR_PNPM_NO_MATURE_MATCHING_VERSION]`, naming +// `@temporal-contract/contract@8.0.0-beta.4` and +// `@temporal-contract/worker@8.0.0-beta.4`). Aliasing to source here instead +// mirrors the *resolve.alias* half of the technique +// `packages/testing/vitest.config.ts` already uses to resolve its own +// peers — `server.deps.inline` below has no counterpart there: testing's +// config aliases bare specifiers imported from its own *source* .ts files, +// which Vite/esbuild transforms and resolves natively, alias included. Here +// the entry point is a prebuilt `.mjs` under `@temporal-contract/testing`, +// which Vitest externalizes to Node's native loader by default — bypassing +// Vite's resolver, and therefore this alias, entirely. `server.deps.inline` +// forces Vitest to route that package through Vite instead, which is the +// only reason it's needed here. +// +// Deliberately scoped to the `integration-inprocess` project only (not +// `integration`, the Docker tier) — nothing in that tier consumes `testRig` +// yet, and applying it there would silently swap 10 unrelated spec files +// from exercising `@temporal-contract/client`'s built `dist` output to its +// source, retiring the only place that dist ever runs under test. If a +// future migration needs `testRig` from the Docker tier too, widen this +// deliberately, not as a side effect of some other change. +// +// Only the two specifiers `test-rig.ts` actually imports today are aliased. +// `packages/testing/vitest.config.ts` also aliases `@temporal-contract/contract`, +// `/contract/errors`, `/contract/internal`, `/worker/activity`, and +// `/worker/workflow` for its own broader needs — add the matching entry +// here if and when a migrated spec's dependency chain actually needs it, +// rather than pre-aliasing unused specifiers. +const workspaceAliases = [ + { find: /^@temporal-contract\/client$/, replacement: sibling("../client/src/index.ts") }, + { find: /^@temporal-contract\/worker\/worker$/, replacement: sibling("./src/worker.ts") }, +]; + export default defineConfig({ test: { reporters: ["default"], @@ -33,12 +87,14 @@ export default defineConfig({ }, }, { + resolve: { alias: workspaceAliases }, // In-process integration via the time-skipping // TestWorkflowEnvironment — no Docker; @temporalio/testing downloads // and caches the test-server binary on first run (hence the generous // timeout). test: { name: "integration-inprocess", + server: { deps: { inline: [/@temporal-contract\/testing/] } }, include: ["src/**/__tests__/*.inprocess.spec.ts"], testTimeout: 120_000, setupFiles: ["./src/vitest.setup.ts"],