From 1552136193af4aa85786f2d3577ef271213058aa Mon Sep 17 00:00:00 2001 From: psteinroe Date: Tue, 20 Oct 2026 15:00:24 +0000 Subject: [PATCH] fix(waits): enforce wait deadlines, fail empty races and memoize child timeouts - result waits store no outcome after their deadline (C11) - waitForAny without an open branch raises instead of suspending forever (C12) - a caught invokeAndWait timeout is memoized and no longer fails later child calls (C23) - conductor.waitForResult cancels a blocked read on timeout (O4-08) - durations take DurationInput everywhere; timeout: 0 means zero (D5c) --- AGENTS.md | 8 +- docs/content/api/conductor.md | 4 +- docs/content/api/task-context.md | 21 +- docs/content/task-execution/batching.md | 2 +- docs/content/task-execution/timeouts.md | 4 +- migrations/0000000001_setup.sql | 63 ++- packages/pgconductor-js/src/conductor.ts | 28 +- .../pgconductor-js/src/database-client.ts | 27 +- packages/pgconductor-js/src/generated/sql.ts | 63 ++- packages/pgconductor-js/src/query-builder.ts | 11 +- packages/pgconductor-js/src/task-context.ts | 69 ++- .../tests/integration/wait-deadlines.test.ts | 509 ++++++++++++++++++ .../tests/integration/wait-for-result.test.ts | 3 +- .../tests/mocks/database-client.mock.ts | 2 +- 14 files changed, 709 insertions(+), 105 deletions(-) create mode 100644 packages/pgconductor-js/tests/integration/wait-deadlines.test.ts diff --git a/AGENTS.md b/AGENTS.md index 10fe655..da555bb 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -241,6 +241,10 @@ Tables live in the `pgconductor` schema with a `_private_` prefix (`_private_exe - Used when `ctx.invokeAndWait()` is called without timeout - Parent waits forever until child completes +**Durations and Child Timeouts** +- Every duration argument (`ctx.sleep()`, wait and child timeouts, `conductor.waitForResult()`) is a `DurationInput` read with `parseDuration()`: numbers are milliseconds, and `0` means zero, never infinity. Config fields named `...Ms` stay numbers +- An `invokeAndWait` step stores `{status: 'completed', result}` (settlement) or `{status: 'timed_out'}` (`clearWaitingState`, which memoizes the timeout when the parent resumes at the step's `waiting_step_key` without a result), so a replay after a caught timeout throws it again + **Dedupe Keys** - The unique index on `(task_key, dedupe_key, queue)` covers only rows with `singleton_on is null`, so its conflict targets and lookups by `dedupe_key` need `singleton_on is null` to use it - Throttled and debounced rows set `singleton_on` to their slot and are unique per `(task_key, singleton_on, dedupe_key, queue)` while non-terminal. Ordinary invokes never update or supersede them, and throttle and debounce never supersede @@ -262,7 +266,9 @@ Tables live in the `pgconductor` schema with a `_private_` prefix (`_private_exe **Result Waits** - `{ execution: id }` branches of `ctx.waitForAny()` are `execution_wait` subscriptions with `awaited_execution_id` set and no event key, so race, timeout and cleanup work as for event waits. `ctx.waitForResult()` is a `waitForAny()` with one such branch -- `_private_register_event_race` locks awaited executions, all at once, before the waiter (the order settlement uses), stores the outcome of a settled one at once, and otherwise appends `{execution_id, queue, step_key}` to the awaited row's `result_waiters` +- `_private_register_event_race` locks awaited executions, all at once, before the waiter (the order settlement uses), stores the outcome of a settled one at once, and otherwise appends `{execution_id, queue, step_key}` to the awaited row's `result_waiters`. Only the first call registers a branch: one with a subscription gets its outcome from the path that ends the execution +- Deadlines compare against the time the outcome settles: `_private_resolve_result_waits` stores no outcome after its subscription's `expires_at`, so a wait whose waiter resumes late still times out +- A race keeps its deadline on its branch subscriptions, so `_private_register_event_race` raises when it has neither a winner nor an open subscription (no branches, or only branches closed by an earlier race) - Every path that ends an execution (settlement's `resolved_result_waits`, `cancel_execution()`, `invoke()` superseding a running execution, `clearWaitingState`, and removing pending schedule executions) reads `result_waiters` from the locked or deleted row, like `signal_count`, so a wait registered after its snapshot is not missed. It passes the waits with their outcomes to `_private_resolve_result_waits`, which stores `{status: 'resolved', outcome}` under each wait's step key and wakes a waiter suspended at a wait, like dispatch does. A new path that ends or removes executions must do the same; `drop_queue()` does not - `_private_resolve_result_waits` wakes suspended waiters with its own update, which runs before the data-modifying CTEs of the calling statement. A statement that also updates or deletes a suspended execution must exclude it from the waits it passes, or the later update of that row fails or is lost - A race that settles removes its entries from the `result_waiters` of awaited executions still running, so timeouts and lost races do not accumulate entries diff --git a/docs/content/api/conductor.md b/docs/content/api/conductor.md index 782bce8..29b30be 100644 --- a/docs/content/api/conductor.md +++ b/docs/content/api/conductor.md @@ -283,14 +283,14 @@ const result = await conductor.waitForResult( { name: "send-email" }, // optional, types the result executionId, { - timeout?: number, // milliseconds, rejects with a TimeoutError + timeout?: DurationInput, // milliseconds or a string such as "30s"; rejects with a TimeoutError pollIntervalMs?: number, // default: 1000 signal?: AbortSignal, // rejects with the abort reason }, ); ``` -Resolves with the return value of a completed execution. Rejects with the error message if the execution failed or was cancelled, and rejects if the execution does not exist. The result is only available while the execution exists: with `removeOnComplete: true` the execution is removed when it completes, so `waitForResult` rejects with a not found error. +Resolves with the return value of a completed execution. The timeout also cancels a read in flight, so a blocked database cannot delay the rejection, and `timeout: 0` rejects at once. Rejects with the error message if the execution failed or was cancelled, and rejects if the execution does not exist. The result is only available while the execution exists: with `removeOnComplete: true` the execution is removed when it completes, so `waitForResult` rejects with a not found error. Inside a task, use [`ctx.waitForResult()`](task-context.md#ctxwaitforresult) instead: it suspends the task rather than polling and returns failures and cancellations as values. diff --git a/docs/content/api/task-context.md b/docs/content/api/task-context.md index 31dfbb1..c983a43 100644 --- a/docs/content/api/task-context.md +++ b/docs/content/api/task-context.md @@ -46,7 +46,7 @@ Pause execution for a duration: ```typescript await ctx.sleep( stepName: string, - durationMs: number + duration: DurationInput // milliseconds, or a string such as "15s", "15m", "1h" or "1d" ): Promise ``` @@ -59,7 +59,7 @@ const task = conductor.createTask( async (event, ctx) => { await ctx.step("step1", () => doWork()); - await ctx.sleep("pause", 3600000); // 1 hour + await ctx.sleep("pause", "1h"); await ctx.step("step2", () => moreWork()); } @@ -228,7 +228,11 @@ for (let n = 0; ; n++) { matching event, settlement or signal. - Losing executions keep running; the wait never cancels them. - The timeout starts when `waitForAny()` is first called. When it wins, the result is - `{ key: "timeout" }` instead of an error, so `timeout` cannot be used as a name. + `{ key: "timeout" }` instead of an error, so `timeout` cannot be used as a name. An event or + outcome after the timeout does not win, even when the task resumes later. +- At least one branch must be open: a race whose branches are all closed by an earlier + `waitForAny()` or already timed out, or one without branches, throws + `waitForAny "" has no open branch`. - When the wait settles, all of the subscriptions are removed. Calling `wait()` on one that did not win throws, unless its event arrived before the wait settled. - The result is memoized by the step key, so retries and resumes return the same winner. @@ -267,7 +271,7 @@ const result = await ctx.invokeAndWait( taskRef: { name: string, queue?: string }, payload: TPayload, options?: { - timeout?: number, + timeout?: DurationInput, // milliseconds, or a string such as "30s" group?: string, metadata?: Metadata | ((metadata: Metadata | undefined) => Metadata), } @@ -285,7 +289,7 @@ const parent = conductor.createTask( "call-child", { name: "child-task" }, { input: event.payload.value }, - { timeout: 30000 } // 30 second timeout + { timeout: "30s" } ); return { final: childResult.output + 10 }; @@ -297,7 +301,9 @@ const parent = conductor.createTask( - Creates child execution - Parent hangs up and waits - Returns child's result, memoized by the step name -- Throws if the timeout elapses first; the child is cancelled +- Throws if the timeout elapses first; the child is cancelled. The timeout is memoized by the + step name, so retries and resumes throw it again instead of invoking the child again. + `timeout: 0` times out at once; omit `timeout` to wait indefinitely - If the child fails permanently or is cancelled, the parent fails permanently with `Child execution failed: ` without running its handler again, so a `catch` around `invokeAndWait()` does not see it. Use `ctx.invoke()` and @@ -399,7 +405,8 @@ if (outcome.status === "completed") { `failed` carries the last error and `cancelled` the cancellation reason. - An execution that already settled returns at once without suspending. - The timeout starts at the first call and resolves to `{ status: "timeout" }`. The execution - keeps running. + keeps running. An outcome after the timeout resolves to `{ status: "timeout" }` too, even when + the task resumes later. - The outcome is memoized by the step key, so retries and resumes return it even after the execution has been removed by retention. It is a `ctx.waitForAny()` with one branch named `result`, so it also uses the step key `:result`. diff --git a/docs/content/task-execution/batching.md b/docs/content/task-execution/batching.md index cc4a8f1..8bd3425 100644 --- a/docs/content/task-execution/batching.md +++ b/docs/content/task-execution/batching.md @@ -135,7 +135,7 @@ const batchTask = conductor.createTask( - `logger` - Logging methods - `signal` - AbortSignal for cancellation. Cancelling any execution of a running batch aborts it; the cancelled executions fail and the others run again without using up an attempt -- `sleep(id, ms)` - Reschedule all executions in the batch. On replay it returns once every execution in the batch has slept at `id`. Since a replayed batch can contain new peers, executions that already slept wait again until the whole batch has. +- `sleep(id, duration)` - Reschedule all executions in the batch. On replay it returns once every execution in the batch has slept at `id`. Since a replayed batch can contain new peers, executions that already slept wait again until the whole batch has. **BatchTaskContext does NOT have:** diff --git a/docs/content/task-execution/timeouts.md b/docs/content/task-execution/timeouts.md index 35de69b..6317f74 100644 --- a/docs/content/task-execution/timeouts.md +++ b/docs/content/task-execution/timeouts.md @@ -68,11 +68,13 @@ When a timeout occurs: 2. The child task is cancelled: a pending child never runs, and a running child's `ctx.signal` is aborted 3. If the parent does not catch the error, the attempt fails and moves to retries +The timeout is memoized by the step name: when the handler replays after catching it, `ctx.invokeAndWait()` throws it again without invoking the child again. + To let the child keep running, start it with `ctx.invoke()` and wait for it with `ctx.waitForResult()`, whose timeout leaves the execution running. ## Infinite Timeout -Omit the timeout option to wait indefinitely: +The timeout is a duration: milliseconds, or a string such as `"30s"` or `"1h"`. `timeout: 0` times out at once. Omit the timeout option to wait indefinitely: ```typescript // Wait forever for child to complete diff --git a/migrations/0000000001_setup.sql b/migrations/0000000001_setup.sql index cbc8407..021f804 100644 --- a/migrations/0000000001_setup.sql +++ b/migrations/0000000001_setup.sql @@ -1744,7 +1744,8 @@ $function$; -- Stores the outcomes of settled executions for the waits in p_waits, read from their -- result_waiters ([{execution_id, queue, step_key, outcome}]), and wakes waiters suspended --- at a wait, like dispatch does. Every path that ends an execution calls it with the +-- at a wait, like dispatch does. An outcome after the deadline of its wait is not stored, +-- so the wait times out. Every path that ends an execution calls it with the -- result_waiters of its locked or deleted row, and only when there are waiters, so -- settlement stays cheap to plan. create or replace function pgconductor._private_resolve_result_waits(p_waits jsonb) @@ -1778,6 +1779,12 @@ begin jsonb_build_object('status', 'resolved', 'outcome', w.outcome) from waits w join locked_waiters l on l.id = w.execution_id and l.queue = w.queue + join pgconductor._private_custom_event_subscriptions subscription + on subscription.kind = 'execution_wait' + and subscription.execution_id = w.execution_id + and subscription.step_key = w.step_key + where subscription.expires_at is null + or subscription.expires_at >= pgconductor._private_current_time() on conflict (execution_id, key) do nothing returning execution_id, queue, key ), deleted_waits as ( @@ -1802,15 +1809,16 @@ $function$; -- Registers a waitForAny step over subscriptions created by -- _private_register_event_wait, referenced by their step keys. The branch whose -- event was stored first wins (then branch order). An { execution } branch is --- registered here: the outcome of a settled execution is stored at once, otherwise --- the wait gets a subscription and is added to the execution's result_waiters, from --- which settlement and cancel_execution store the outcome. A { signal } branch is +-- registered by the first call: the outcome of a settled execution is stored at once, +-- otherwise the wait gets a subscription and is added to the execution's result_waiters, +-- from which settlement and cancel_execution store the outcome. A { signal } branch is -- registered here too: an undelivered signal with its name, or the one whose handler is -- running (p_delivering_signal_id), is stored at once, otherwise the wait gets a -- subscription from which _private_store_signal stores the next one. Otherwise the timeout is set -- on all branch subscriptions and the execution is released until dispatch wakes --- it or the timeout passes. Settling removes the branch subscriptions and closes --- the branches, so subscribing again on a resume is a no-op. +-- it or the timeout passes. A race without a subscription has nothing to wait on and +-- raises. Settling removes the branch subscriptions and closes the branches, so +-- subscribing again on a resume is a no-op. create or replace function pgconductor._private_register_event_race( p_execution_id uuid, p_queue text, @@ -1834,9 +1842,11 @@ declare from jsonb_array_elements(p_branches) branch where branch ? 'executionId' ); + v_new_branches jsonb; v_unknown_id uuid; v_result jsonb; v_expires_at timestamptz; + v_subscriptions bigint; v_now timestamptz := pgconductor._private_current_time(); begin -- Settlement locks an execution before its waiters, so awaited executions are locked first, @@ -1872,9 +1882,10 @@ begin end if; if cardinality(v_awaited_ids) > 0 then - -- A branch with a stored step already received its outcome - select (branch->>'executionId')::uuid - into v_unknown_id + -- A branch with a stored step already received its outcome, and one with a subscription + -- receives it from the path that ends the execution, which checks the deadline + select coalesce(jsonb_agg(branch), '[]'::jsonb) + into v_new_branches from jsonb_array_elements(p_branches) branch where branch ? 'executionId' and not exists ( @@ -1882,11 +1893,21 @@ begin where step.execution_id = p_execution_id and step.key = branch->>'stepKey' ) and not exists ( - select 1 from pgconductor._private_executions awaited - where awaited.id = (branch->>'executionId')::uuid - and (branch->>'taskKey' is null - or (awaited.task_key = branch->>'taskKey' and awaited.queue = branch->>'queue')) - ) + select 1 from pgconductor._private_custom_event_subscriptions subscription + where subscription.kind = 'execution_wait' + and subscription.execution_id = p_execution_id + and subscription.step_key = branch->>'stepKey' + ); + + select (branch->>'executionId')::uuid + into v_unknown_id + from jsonb_array_elements(v_new_branches) branch + where not exists ( + select 1 from pgconductor._private_executions awaited + where awaited.id = (branch->>'executionId')::uuid + and (branch->>'taskKey' is null + or (awaited.task_key = branch->>'taskKey' and awaited.queue = branch->>'queue')) + ) limit 1; if found then @@ -1903,15 +1924,11 @@ begin then jsonb_build_object('status', 'cancelled', 'reason', awaited.last_error) else jsonb_build_object('status', 'failed', 'error', awaited.last_error) end as outcome - from jsonb_array_elements(p_branches) branch + from jsonb_array_elements(v_new_branches) branch join pgconductor._private_executions awaited on awaited.id = (branch->>'executionId')::uuid and (branch->>'taskKey' is null or (awaited.task_key = branch->>'taskKey' and awaited.queue = branch->>'queue')) - where not exists ( - select 1 from pgconductor._private_steps step - where step.execution_id = p_execution_id and step.key = branch->>'stepKey' - ) ), settled as ( insert into pgconductor._private_steps (execution_id, queue, key, result) select p_execution_id, p_queue, b.step_key, @@ -1992,14 +2009,16 @@ begin if v_result is null then -- The first call sets the timeout, so later calls find it set. - select max(subscription.expires_at) - into v_expires_at + select max(subscription.expires_at), count(*) + into v_expires_at, v_subscriptions from pgconductor._private_custom_event_subscriptions subscription where subscription.kind = 'execution_wait' and subscription.execution_id = p_execution_id and subscription.step_key = any(v_step_keys); - if v_expires_at is null then + if v_subscriptions = 0 then + raise exception 'waitForAny "%" has no open branch', p_step_key; + elsif v_expires_at is null then v_expires_at := v_now + p_timeout_ms * interval '1 millisecond'; update pgconductor._private_custom_event_subscriptions subscription diff --git a/packages/pgconductor-js/src/conductor.ts b/packages/pgconductor-js/src/conductor.ts index 7bdd9fa..805d4b6 100644 --- a/packages/pgconductor-js/src/conductor.ts +++ b/packages/pgconductor-js/src/conductor.ts @@ -39,6 +39,7 @@ import { Worker, type WorkerConfig } from "./worker.js"; import { DefaultLogger, type Logger } from "./lib/logger.js"; import type { AddedContext, Middleware } from "./middleware.js"; import { waitFor } from "./lib/wait-for.js"; +import { parseDuration, type DurationInput } from "./lib/duration.js"; import { SchemaManager } from "./schema-manager.js"; import { Telemetry } from "./telemetry.js"; import { validateSchema } from "./lib/validate-schema.js"; @@ -58,8 +59,8 @@ import { } from "./schemas.js"; export type WaitForResultOptions = { - /** Reject with a `TimeoutError` after this many milliseconds */ - timeout?: number; + /** Reject with a `TimeoutError` after this long; numbers are milliseconds */ + timeout?: DurationInput; /** Delay between polls (default: 1000) */ pollIntervalMs?: number; signal?: AbortSignal; @@ -544,18 +545,25 @@ export class Conductor< : [taskOrId, idOrOptions as string, options]; const stop = AbortSignal.any([ ...(signal ? [signal] : []), - ...(timeout ? [AbortSignal.timeout(timeout)] : []), + ...(timeout === undefined ? [] : [AbortSignal.timeout(parseDuration(timeout))]), ]); while (true) { stop.throwIfAborted(); - const execution = await this.db.getExecution( - { - executionId, - task: task && { key: task.name, queue: task.queue || "default" }, - }, - { signal: stop }, - ); + // The read is cancelled when stop fires, and a result read after that is not returned + const execution = await this.db + .getExecution( + { + executionId, + task: task && { key: task.name, queue: task.queue || "default" }, + }, + { signal: stop }, + ) + .catch((error: unknown) => { + stop.throwIfAborted(); + throw error; + }); + stop.throwIfAborted(); if (!execution) { throw new Error(`Execution ${executionId} not found`); } diff --git a/packages/pgconductor-js/src/database-client.ts b/packages/pgconductor-js/src/database-client.ts index 468d0c1..4974d93 100644 --- a/packages/pgconductor-js/src/database-client.ts +++ b/packages/pgconductor-js/src/database-client.ts @@ -635,14 +635,24 @@ export class DatabaseClient { return result; } + /** Cancels the read when the signal aborts, so a read blocked on a lock does not outlive it */ async getExecution( args: GetExecutionArgs, opts?: QueryMethodOptions, ): Promise { - const rows = await this.query(() => this.builder.buildGetExecution(args), { - label: "getExecution", - ...opts, - }); + const rows = await this.query( + async () => { + const query = this.builder.buildGetExecution(args); + const cancel = () => query.cancel(); + opts?.signal?.addEventListener("abort", cancel); + try { + return await query; + } finally { + opts?.signal?.removeEventListener("abort", cancel); + } + }, + { label: "getExecution", ...opts }, + ); return rows[0] || null; } @@ -760,11 +770,16 @@ export class DatabaseClient { }); } - async clearWaitingState(args: ClearWaitingStateArgs, opts?: QueryMethodOptions): Promise { - await this.query(() => this.builder.buildClearWaitingState(args), { + /** Returns false when the claim no longer holds the execution */ + async clearWaitingState( + args: ClearWaitingStateArgs, + opts?: QueryMethodOptions, + ): Promise { + const rows = await this.query(() => this.builder.buildClearWaitingState(args), { label: "clearWaitingState", ...opts, }); + return rows.length > 0; } async registerEventWait( diff --git a/packages/pgconductor-js/src/generated/sql.ts b/packages/pgconductor-js/src/generated/sql.ts index 3c154f6..9625bcb 100644 --- a/packages/pgconductor-js/src/generated/sql.ts +++ b/packages/pgconductor-js/src/generated/sql.ts @@ -1758,7 +1758,8 @@ $function$; -- Stores the outcomes of settled executions for the waits in p_waits, read from their -- result_waiters ([{execution_id, queue, step_key, outcome}]), and wakes waiters suspended --- at a wait, like dispatch does. Every path that ends an execution calls it with the +-- at a wait, like dispatch does. An outcome after the deadline of its wait is not stored, +-- so the wait times out. Every path that ends an execution calls it with the -- result_waiters of its locked or deleted row, and only when there are waiters, so -- settlement stays cheap to plan. create or replace function pgconductor._private_resolve_result_waits(p_waits jsonb) @@ -1792,6 +1793,12 @@ begin jsonb_build_object('status', 'resolved', 'outcome', w.outcome) from waits w join locked_waiters l on l.id = w.execution_id and l.queue = w.queue + join pgconductor._private_custom_event_subscriptions subscription + on subscription.kind = 'execution_wait' + and subscription.execution_id = w.execution_id + and subscription.step_key = w.step_key + where subscription.expires_at is null + or subscription.expires_at >= pgconductor._private_current_time() on conflict (execution_id, key) do nothing returning execution_id, queue, key ), deleted_waits as ( @@ -1816,15 +1823,16 @@ $function$; -- Registers a waitForAny step over subscriptions created by -- _private_register_event_wait, referenced by their step keys. The branch whose -- event was stored first wins (then branch order). An { execution } branch is --- registered here: the outcome of a settled execution is stored at once, otherwise --- the wait gets a subscription and is added to the execution's result_waiters, from --- which settlement and cancel_execution store the outcome. A { signal } branch is +-- registered by the first call: the outcome of a settled execution is stored at once, +-- otherwise the wait gets a subscription and is added to the execution's result_waiters, +-- from which settlement and cancel_execution store the outcome. A { signal } branch is -- registered here too: an undelivered signal with its name, or the one whose handler is -- running (p_delivering_signal_id), is stored at once, otherwise the wait gets a -- subscription from which _private_store_signal stores the next one. Otherwise the timeout is set -- on all branch subscriptions and the execution is released until dispatch wakes --- it or the timeout passes. Settling removes the branch subscriptions and closes --- the branches, so subscribing again on a resume is a no-op. +-- it or the timeout passes. A race without a subscription has nothing to wait on and +-- raises. Settling removes the branch subscriptions and closes the branches, so +-- subscribing again on a resume is a no-op. create or replace function pgconductor._private_register_event_race( p_execution_id uuid, p_queue text, @@ -1848,9 +1856,11 @@ declare from jsonb_array_elements(p_branches) branch where branch ? 'executionId' ); + v_new_branches jsonb; v_unknown_id uuid; v_result jsonb; v_expires_at timestamptz; + v_subscriptions bigint; v_now timestamptz := pgconductor._private_current_time(); begin -- Settlement locks an execution before its waiters, so awaited executions are locked first, @@ -1886,9 +1896,10 @@ begin end if; if cardinality(v_awaited_ids) > 0 then - -- A branch with a stored step already received its outcome - select (branch->>'executionId')::uuid - into v_unknown_id + -- A branch with a stored step already received its outcome, and one with a subscription + -- receives it from the path that ends the execution, which checks the deadline + select coalesce(jsonb_agg(branch), '[]'::jsonb) + into v_new_branches from jsonb_array_elements(p_branches) branch where branch ? 'executionId' and not exists ( @@ -1896,11 +1907,21 @@ begin where step.execution_id = p_execution_id and step.key = branch->>'stepKey' ) and not exists ( - select 1 from pgconductor._private_executions awaited - where awaited.id = (branch->>'executionId')::uuid - and (branch->>'taskKey' is null - or (awaited.task_key = branch->>'taskKey' and awaited.queue = branch->>'queue')) - ) + select 1 from pgconductor._private_custom_event_subscriptions subscription + where subscription.kind = 'execution_wait' + and subscription.execution_id = p_execution_id + and subscription.step_key = branch->>'stepKey' + ); + + select (branch->>'executionId')::uuid + into v_unknown_id + from jsonb_array_elements(v_new_branches) branch + where not exists ( + select 1 from pgconductor._private_executions awaited + where awaited.id = (branch->>'executionId')::uuid + and (branch->>'taskKey' is null + or (awaited.task_key = branch->>'taskKey' and awaited.queue = branch->>'queue')) + ) limit 1; if found then @@ -1917,15 +1938,11 @@ begin then jsonb_build_object('status', 'cancelled', 'reason', awaited.last_error) else jsonb_build_object('status', 'failed', 'error', awaited.last_error) end as outcome - from jsonb_array_elements(p_branches) branch + from jsonb_array_elements(v_new_branches) branch join pgconductor._private_executions awaited on awaited.id = (branch->>'executionId')::uuid and (branch->>'taskKey' is null or (awaited.task_key = branch->>'taskKey' and awaited.queue = branch->>'queue')) - where not exists ( - select 1 from pgconductor._private_steps step - where step.execution_id = p_execution_id and step.key = branch->>'stepKey' - ) ), settled as ( insert into pgconductor._private_steps (execution_id, queue, key, result) select p_execution_id, p_queue, b.step_key, @@ -2006,14 +2023,16 @@ begin if v_result is null then -- The first call sets the timeout, so later calls find it set. - select max(subscription.expires_at) - into v_expires_at + select max(subscription.expires_at), count(*) + into v_expires_at, v_subscriptions from pgconductor._private_custom_event_subscriptions subscription where subscription.kind = 'execution_wait' and subscription.execution_id = p_execution_id and subscription.step_key = any(v_step_keys); - if v_expires_at is null then + if v_subscriptions = 0 then + raise exception 'waitForAny "%" has no open branch', p_step_key; + elsif v_expires_at is null then v_expires_at := v_now + p_timeout_ms * interval '1 millisecond'; update pgconductor._private_custom_event_subscriptions subscription diff --git a/packages/pgconductor-js/src/query-builder.ts b/packages/pgconductor-js/src/query-builder.ts index be9441d..7e1d9d9 100644 --- a/packages/pgconductor-js/src/query-builder.ts +++ b/packages/pgconductor-js/src/query-builder.ts @@ -719,7 +719,8 @@ export class QueryBuilder { )`); ctes.push(this.sql`parent_steps_all as ( insert into pgconductor._private_steps (execution_id, queue, key, result) - select parent_id, queue, waiting_step_key, result + select parent_id, queue, waiting_step_key, + jsonb_build_object('status', 'completed', 'result', result) from completed_parents where waiting_step_key is not null on conflict (execution_id, key) do nothing @@ -1500,7 +1501,7 @@ export class QueryBuilder { return this.sql>` -- The child is cancelled like any other: cancel_execution locks it before this parent with timed_out as materialized ( - select e.waiting_on_execution_id as child_id + select e.waiting_on_execution_id as child_id, e.waiting_step_key from pgconductor._private_executions e where e.id = ${executionId}::uuid and e.queue = ${queue}::text @@ -1510,6 +1511,12 @@ export class QueryBuilder { ), cancelled_child as materialized ( select pgconductor.cancel_execution(child_id, 'Cancelled: parent timed out') from timed_out + ), stored_timeout as ( + insert into pgconductor._private_steps (execution_id, queue, key, result) + select ${executionId}::uuid, ${queue}::text, waiting_step_key, + jsonb_build_object('status', 'timed_out') + from timed_out + on conflict (execution_id, key) do nothing ) update pgconductor._private_executions e set waiting_on_execution_id = null, waiting_step_key = null diff --git a/packages/pgconductor-js/src/task-context.ts b/packages/pgconductor-js/src/task-context.ts index 12270a7..d1c840e 100644 --- a/packages/pgconductor-js/src/task-context.ts +++ b/packages/pgconductor-js/src/task-context.ts @@ -302,7 +302,8 @@ export class TaskContext< } } - async sleep(id: string, ms: number): Promise { + /** Suspend for `duration`; numbers are milliseconds. Memoized by `id`. */ + async sleep(id: string, duration: DurationInput): Promise { // Check if we already slept const cached = await this.opts.db.loadStep( { @@ -319,7 +320,7 @@ export class TaskContext< return this.abortAndHangup({ reason: "released", - reschedule_in_ms: ms, + reschedule_in_ms: parseDuration(duration), step_key: id, }); } @@ -494,31 +495,46 @@ export class TaskContext< group, metadata, }: { - timeout?: number; + /** Throw and cancel the child when it has not completed after this long; numbers are milliseconds */ + timeout?: DurationInput; group?: string; /** Metadata for this child only. Defaults to the metadata of this execution. */ metadata?: Metadata | ((metadata: Readonly | undefined) => Metadata); } = {}, ): Promise> { - const cached = await this.opts.db.loadStep( + const timeoutMs = timeout === undefined ? undefined : parseDuration(timeout); + const cached = (await this.opts.db.loadStep( { executionId: this.opts.execution.id, queue: this.opts.execution.queue, key, }, { signal: this.signal }, - ); + )) as { status: "completed"; result: InferReturns } | { status: "timed_out" } | undefined; - if (cached !== undefined) { - return cached as InferReturns; + if (cached?.status === "completed") { + return cached.result; } - // Check if we're already waiting (distinguishes first invoke from timeout) - if (this.opts.execution.waiting_on_execution_id !== null) { - // we resumed but no step exists → timeout occurred - // clear waiting state before throwing + // A claim keeps waiting_step_key only while the execution waits on a child + if (cached === undefined && this.opts.execution.waiting_step_key !== key) { + const childMetadata = await this.childMetadata(metadata); + return this.abortAndHangup({ + reason: "child-invocation", + timeout_ms: timeoutMs === undefined ? "infinity" : timeoutMs, + task, + step_key: key, + payload, + group, + trace_context: this.opts.telemetry.traceContext(), + metadata: childMetadata, + }); + } - await this.opts.db.clearWaitingState( + // Resumed at this key without the child's result: the timeout passed. Clearing the wait + // memoizes the timeout, so a replay throws it again. + if (cached === undefined) { + const cleared = await this.opts.db.clearWaitingState( { executionId: this.opts.execution.id, queue: this.opts.execution.queue, @@ -527,23 +543,14 @@ export class TaskContext< }, { signal: this.signal }, ); - - throw new Error( - timeout ? `Child execution timed out after ${timeout}ms` : "Child execution timeout", - ); + if (!cleared) return this.abortAndHangup({ reason: "suspended" }); } - const childMetadata = await this.childMetadata(metadata); - return this.abortAndHangup({ - reason: "child-invocation", - timeout_ms: timeout || "infinity", - task, - step_key: key, - payload, - group, - trace_context: this.opts.telemetry.traceContext(), - metadata: childMetadata, - }); + throw new Error( + timeoutMs === undefined + ? "Child execution timeout" + : `Child execution timed out after ${timeoutMs}ms`, + ); } /** @@ -899,7 +906,7 @@ export class BatchTaskContext { * After sleep, executions may batch with different peers, so it only * returns once every execution in the current batch has slept at `id`. */ - async sleep(id: string, ms: number): Promise { + async sleep(id: string, duration: DurationInput): Promise { const slept = await this.db.countSteps( { executionIds: this.executions.map((exec) => exec.id), @@ -912,7 +919,11 @@ export class BatchTaskContext { return; } - this.abortController.abort({ reason: "released", reschedule_in_ms: ms, step_key: id }); + this.abortController.abort({ + reason: "released", + reschedule_in_ms: parseDuration(duration), + step_key: id, + }); return new Promise(() => {}); } } diff --git a/packages/pgconductor-js/tests/integration/wait-deadlines.test.ts b/packages/pgconductor-js/tests/integration/wait-deadlines.test.ts new file mode 100644 index 0000000..a39d002 --- /dev/null +++ b/packages/pgconductor-js/tests/integration/wait-deadlines.test.ts @@ -0,0 +1,509 @@ +import { setTimeout as sleep } from "node:timers/promises"; +import postgres from "postgres"; +import { z } from "zod"; +import { afterAll, beforeAll, describe, expect, test } from "vitest"; +import { Conductor } from "../../src/conductor"; +import { Orchestrator } from "../../src/orchestrator"; +import { defineEvent } from "../../src/event-definition"; +import { defineTask } from "../../src/task-definition"; +import { TaskSchemas } from "../../src/schemas"; +import { Deferred } from "../../src/lib/deferred"; +import { TestDatabasePool } from "../fixtures/test-database"; +import { waitForCondition } from "../test-utils"; + +let pool: TestDatabasePool; + +beforeAll(async () => { + pool = await TestDatabasePool.create(); +}, 60000); + +afterAll(async () => { + await pool?.destroy(); +}); + +const ping = defineEvent({ name: "ping", payload: z.object({}) }); + +// An outcome that settles after the deadline of a result wait, but before the waiter resumes, +// does not win it +describe.each([ + { + name: "waitForResult with a completion", + cancel: false, + wait: (ctx: any, id: string) => ctx.waitForResult("result", id, { timeout: "1s" }), + timeout: { status: "timeout" }, + }, + { + name: "waitForResult with a cancellation", + cancel: true, + wait: (ctx: any, id: string) => ctx.waitForResult("result", id, { timeout: "1s" }), + timeout: { status: "timeout" }, + }, + { + name: "waitForAny with an event and a completion", + cancel: false, + wait: async (ctx: any, id: string) => { + const sub = await ctx.subscribe("ping", { event: ping }); + return ctx.waitForAny("result", { ping: sub, work: { execution: id } }, { timeout: "1s" }); + }, + timeout: { key: "timeout" }, + }, +])("$name settled after the deadline", ({ cancel, wait, timeout }) => { + test("times out", async () => { + const db = await pool.child(); + await db.clock.set(new Date("2024-01-01T12:00:00Z")); + + const conductor = Conductor.create({ + sql: db.sql, + clock: db.clock, + tasks: TaskSchemas.fromSchema([ + defineTask({ name: "waiter", payload: z.object({}) }), + defineTask({ + name: "work", + queue: "work", + payload: z.object({}), + returns: z.object({ value: z.number() }), + }), + ]), + events: [ping], + context: {}, + }); + await conductor.ensureInstalled(); + + const release = new Deferred(); + const outcomes: unknown[] = []; + const workIds: string[] = []; + const waiter = conductor.createTask({ name: "waiter" }, { invocable: true }, async (_, ctx) => { + const id = await ctx.invoke("start", { name: "work", queue: "work" }, {}); + workIds.push(id); + outcomes.push(await wait(ctx, id)); + }); + const work = conductor.createTask( + { name: "work", queue: "work" }, + { invocable: true }, + async () => { + await release.promise; + return { value: 42 }; + }, + ); + + const config = { pollIntervalMs: 10, flushIntervalMs: 10 }; + const workers = Orchestrator.create({ + conductor, + workers: [conductor.createWorker({ queue: "work", tasks: [work], config })], + }); + const waiters = () => + Orchestrator.create({ conductor, tasks: [waiter], defaultWorker: config }); + + await workers.start(); + const first = waiters(); + await first.start(); + try { + const waiterId = await conductor.invoke({ name: "waiter" }, {}); + + // The waiter suspends with its deadline at 12:00:01, and nothing polls it afterwards + await waitForCondition(async () => { + const [row] = await db.sql` + select 1 from pgconductor._private_executions + where id = ${waiterId} and locked_by is null and waiting_step_key = 'result' + `; + return row !== undefined; + }); + await first.stop(); + + // The deadline passes, then the awaited execution settles + await db.clock.set(new Date("2024-01-01T12:00:02Z")); + const workId = workIds[0] || ""; + if (cancel) await conductor.cancel(workId); + else release.resolve(); + await waitForCondition(async () => { + const [row] = await db.sql` + select 1 from pgconductor._private_executions + where id = ${workId} and (completed_at is not null or failed_at is not null) + `; + return row !== undefined; + }); + + // The waiter resumes at 12:00:03, after its deadline + await db.clock.set(new Date("2024-01-01T12:00:03Z")); + const second = waiters(); + await second.start(); + try { + await conductor.waitForResult(waiterId, { pollIntervalMs: 10 }); + } finally { + await second.stop(); + } + + expect(outcomes).toEqual([timeout]); + } finally { + release.resolve(); + await first.stop(); + await workers.stop(); + } + }); +}); + +// A race without an open branch has nowhere to keep its deadline, so it fails instead of waiting +describe("waitForAny without an open branch", () => { + async function race(handler: (ctx: any) => Promise) { + const db = await pool.child(); + await db.clock.set(new Date("2024-01-01T12:00:00Z")); + const conductor = Conductor.create({ + sql: db.sql, + clock: db.clock, + tasks: TaskSchemas.fromSchema([defineTask({ name: "racer", payload: z.object({}) })]), + events: [ping], + context: {}, + }); + await conductor.ensureInstalled(); + const racer = conductor.createTask( + { name: "racer", maxAttempts: 1 }, + { invocable: true }, + async (_, ctx) => { + await handler(ctx); + }, + ); + const orchestrator = Orchestrator.create({ + conductor, + tasks: [racer], + defaultWorker: { pollIntervalMs: 10, flushIntervalMs: 10 }, + }); + await orchestrator.start(); + const id = await conductor.invoke({ name: "racer" }, {}); + const read = async () => { + const [row] = await db.sql< + { run_at: Date; waiting_step_key: string | null; failed: boolean; last_error: string }[] + >` + select run_at, waiting_step_key, failed_at is not null as failed, last_error + from pgconductor._private_executions where id = ${id} + `; + if (!row) throw new Error("execution not found"); + return row; + }; + return { db, orchestrator, read }; + } + + test("with no branches", async () => { + const { orchestrator, read } = await race((ctx) => + ctx.waitForAny("race", {}, { timeout: "1s" }), + ); + try { + await waitForCondition(async () => (await read()).failed); + expect((await read()).last_error).toBe('waitForAny "race" has no open branch'); + } finally { + await orchestrator.stop(); + } + }); + + test("with only a branch an earlier race closed", async () => { + const { db, orchestrator, read } = await race(async (ctx) => { + const sub = await ctx.subscribe("sub", { event: ping }); + await ctx.waitForAny("first", { a: sub }, { timeout: "1s" }); + await ctx.waitForAny("second", { a: sub }, { timeout: "1s" }); + }); + try { + await waitForCondition(async () => (await read()).waiting_step_key === "first"); + await db.clock.set(new Date((await read()).run_at.getTime() + 1000)); + await waitForCondition(async () => (await read()).failed); + expect((await read()).last_error).toBe('waitForAny "second" has no open branch'); + } finally { + await orchestrator.stop(); + } + }); +}); + +const parentDefinition = defineTask({ + name: "parent", + returns: z.object({ via: z.string() }), +}); +const primaryDefinition = defineTask({ name: "primary" }); +const fallbackDefinition = defineTask({ + name: "fallback", + returns: z.object({ ok: z.boolean() }), +}); +const tasks = TaskSchemas.fromSchema([parentDefinition, primaryDefinition, fallbackDefinition]); + +test("a caught invokeAndWait timeout does not fail the next child invocation", async () => { + const db = await pool.child(); + const start = new Date("2024-01-01T00:00:00Z"); + await db.clock.set(start); + const conductor = Conductor.create({ sql: db.sql, tasks, context: {}, clock: db.clock }); + const errors: string[] = []; + + const primary = conductor.createTask( + { name: "primary" }, + { invocable: true }, + async (_e, ctx) => { + await ctx.sleep("long", 60_000); + }, + ); + const fallback = conductor.createTask({ name: "fallback" }, { invocable: true }, async () => ({ + ok: true, + })); + const parent = conductor.createTask( + { name: "parent", maxAttempts: 1 }, + { invocable: true }, + async (_e, ctx) => { + try { + await ctx.invokeAndWait("primary", { name: "primary" }, {}, { timeout: 1000 }); + return { via: "primary" }; + } catch (err) { + errors.push((err as Error).message); + } + try { + await ctx.invokeAndWait("fallback", { name: "fallback" }, {}); + } catch (err) { + errors.push((err as Error).message); + throw err; + } + return { via: "fallback" }; + }, + ); + + await conductor.ensureInstalled(); + const orchestrator = Orchestrator.create({ + conductor, + tasks: [parent, primary, fallback], + defaultWorker: { pollIntervalMs: 50, flushIntervalMs: 50 }, + }); + try { + await orchestrator.start(); + const parentId = await conductor.invoke({ name: "parent" }, {}); + + // The primary child is asleep and the parent waits on it + await waitForCondition(async () => { + const [r] = await db.sql<{ sleeping: boolean }[]>` + select exists ( + select 1 from pgconductor._private_steps s + join pgconductor._private_executions e on e.id = s.execution_id + where e.task_key = 'primary' and s.key = 'long' + ) and exists ( + select 1 from pgconductor._private_executions + where id = ${parentId}::uuid and waiting_on_execution_id is not null and locked_by is null + ) as sleeping + `; + return r?.sleeping === true; + }); + await db.clock.advance("1500ms"); + + await waitForCondition(async () => { + const [r] = await db.sql<{ settled: boolean }[]>` + select completed_at is not null or failed_at is not null as settled + from pgconductor._private_executions where id = ${parentId}::uuid + `; + return r?.settled === true; + }); + + const [settled] = await db.sql<{ result: unknown; last_error: string | null }[]>` + select result, last_error from pgconductor._private_executions where id = ${parentId}::uuid + `; + const fallbackRuns = await db.sql` + select id from pgconductor._private_executions where task_key = 'fallback' + `; + // The memoized timeout is thrown again when the handler replays after the fallback + expect({ + errors, + fallbackExecutions: fallbackRuns.length, + result: settled?.result, + lastError: settled?.last_error, + }).toEqual({ + errors: ["Child execution timed out after 1000ms", "Child execution timed out after 1000ms"], + fallbackExecutions: 1, + result: { via: "fallback" }, + lastError: null, + }); + } finally { + await orchestrator.stop(); + } +}); + +// Found while reproducing C23: the timeout is not memoized, so a replay after a caught timeout +// (here, resuming from a later sleep) invokes the timed-out child again and waits for it. +test("a replay after a caught invokeAndWait timeout does not invoke the child again", async () => { + const db = await pool.child(); + const start = new Date("2024-01-01T00:00:00Z"); + await db.clock.set(start); + const conductor = Conductor.create({ sql: db.sql, tasks, context: {}, clock: db.clock }); + const errors: string[] = []; + + const primary = conductor.createTask( + { name: "primary" }, + { invocable: true }, + async (_e, ctx) => { + await ctx.sleep("long", 60_000); + }, + ); + const parent = conductor.createTask( + { name: "parent", maxAttempts: 1 }, + { invocable: true }, + async (_e, ctx) => { + try { + await ctx.invokeAndWait("primary", { name: "primary" }, {}, { timeout: 1000 }); + } catch (err) { + errors.push((err as Error).message); + } + await ctx.sleep("after", 1000); + return { via: "sleep" }; + }, + ); + + await conductor.ensureInstalled(); + const orchestrator = Orchestrator.create({ + conductor, + tasks: [parent, primary], + defaultWorker: { pollIntervalMs: 50, flushIntervalMs: 50 }, + }); + const parentState = async (id: string) => { + const [r] = await db.sql<{ completed: boolean; sleeping: boolean; primaries: number }[]>` + select + completed_at is not null as completed, + locked_by is null and exists ( + select 1 from pgconductor._private_steps s where s.execution_id = e.id and s.key = 'after' + ) as sleeping, + (select count(*)::int from pgconductor._private_executions where task_key = 'primary') as primaries + from pgconductor._private_executions e where id = ${id}::uuid + `; + return r; + }; + try { + await orchestrator.start(); + const parentId = await conductor.invoke({ name: "parent" }, {}); + await waitForCondition(async () => { + const [r] = await db.sql<{ waiting: boolean }[]>` + select waiting_on_execution_id is not null and locked_by is null as waiting + from pgconductor._private_executions where id = ${parentId}::uuid + `; + return r?.waiting === true; + }); + await db.clock.advance("1500ms"); + await waitForCondition(async () => (await parentState(parentId))?.sleeping === true); + await db.clock.advance("1500ms"); + // Either the replay completes, or it invoked the primary child again and waits on it + await waitForCondition(async () => { + const state = await parentState(parentId); + return state?.completed === true || (state?.primaries || 0) > 1; + }); + + const { completed, primaries } = (await parentState(parentId)) || {}; + expect({ errors, completed, primaries }).toEqual({ + errors: ["Child execution timed out after 1000ms", "Child execution timed out after 1000ms"], + completed: true, + primaries: 1, + }); + } finally { + await orchestrator.stop(); + } +}); + +test("conductor.waitForResult times out while its read waits on a lock", async () => { + const db = await pool.child(); + const conductor = Conductor.create({ + sql: db.sql, + tasks: TaskSchemas.fromSchema([primaryDefinition]), + context: {}, + }); + await conductor.ensureInstalled(); + const primary = conductor.createTask({ name: "primary" }, { invocable: true }, async () => {}); + const id = await conductor.invoke({ name: "primary" }, {}); + await Orchestrator.create({ conductor, tasks: [primary] }).drain(); + + const locker = postgres(db.url, { max: 1 }); + const locked = new Deferred(); + const unlock = new Deferred(); + const holding = locker.begin(async (tx) => { + await tx`lock table pgconductor._private_executions in access exclusive mode`; + locked.resolve(); + await unlock.promise; + }); + try { + await locked.promise; + const outcome = await Promise.race([ + conductor.waitForResult(id, { timeout: 100 }).then( + () => "resolved", + (error: Error) => error.name, + ), + sleep(5000).then(() => "blocked"), + ]); + expect(outcome).toBe("TimeoutError"); + } finally { + unlock.resolve(); + await holding; + await locker.end(); + } + // The read was cancelled, so the connection is free again + expect(await conductor.waitForResult(id)).toBe(null); +}); + +describe("durations", () => { + test("an invokeAndWait timeout of 0 times out at once", async () => { + const db = await pool.child(); + const conductor = Conductor.create({ sql: db.sql, tasks, context: {} }); + const primary = conductor.createTask( + { name: "primary" }, + { invocable: true }, + async (_e, ctx) => { + await ctx.sleep("long", "1h"); + }, + ); + const parent = conductor.createTask( + { name: "parent", maxAttempts: 1 }, + { invocable: true }, + async (_e, ctx) => { + try { + await ctx.invokeAndWait("primary", { name: "primary" }, {}, { timeout: 0 }); + return { via: "primary" }; + } catch (err) { + return { via: (err as Error).message }; + } + }, + ); + await conductor.ensureInstalled(); + const orchestrator = Orchestrator.create({ + conductor, + tasks: [parent, primary], + defaultWorker: { pollIntervalMs: 50, flushIntervalMs: 50 }, + }); + await orchestrator.start(); + try { + const parentId = await conductor.invoke({ name: "parent" }, {}); + expect(await conductor.waitForResult(parentId, { pollIntervalMs: 50 })).toEqual({ + via: "Child execution timed out after 0ms", + }); + const [child] = await db.sql<{ cancelled: boolean }[]>` + select cancelled from pgconductor._private_executions where task_key = 'primary' + `; + expect(child?.cancelled).toBe(true); + } finally { + await orchestrator.stop(); + } + }); + + test("conductor.waitForResult with a timeout of 0 rejects at once", async () => { + const db = await pool.child(); + const conductor = Conductor.create({ sql: db.sql, tasks, context: {} }); + await conductor.ensureInstalled(); + const id = await conductor.invoke({ name: "primary" }, {}); + await expect(conductor.waitForResult(id, { timeout: 0 })).rejects.toThrow( + expect.objectContaining({ name: "TimeoutError" }), + ); + }); + + test("ctx.sleep takes a duration string", async () => { + const db = await pool.child(); + await db.clock.set(new Date("2024-01-01T00:00:00Z")); + const conductor = Conductor.create({ sql: db.sql, tasks, context: {}, clock: db.clock }); + const primary = conductor.createTask( + { name: "primary" }, + { invocable: true }, + async (_e, ctx) => { + await ctx.sleep("nap", "1m"); + }, + ); + await conductor.ensureInstalled(); + const id = await conductor.invoke({ name: "primary" }, {}); + await Orchestrator.create({ conductor, tasks: [primary] }).drain(); + const [row] = await db.sql<{ run_at: Date }[]>` + select run_at from pgconductor._private_executions where id = ${id}::uuid + `; + expect(row?.run_at).toEqual(new Date("2024-01-01T00:01:00Z")); + }); +}); diff --git a/packages/pgconductor-js/tests/integration/wait-for-result.test.ts b/packages/pgconductor-js/tests/integration/wait-for-result.test.ts index ab2d596..f16d228 100644 --- a/packages/pgconductor-js/tests/integration/wait-for-result.test.ts +++ b/packages/pgconductor-js/tests/integration/wait-for-result.test.ts @@ -749,7 +749,8 @@ describe("ctx.waitForResult and execution branches (Postgres integration)", () = taskKeys: ["work"], }); await database.sql` - update pgconductor._private_executions set waiting_on_execution_id = ${childId}::uuid + update pgconductor._private_executions + set waiting_on_execution_id = ${childId}::uuid, waiting_step_key = 'child' where id = ${parentId}::uuid `; const waiterId = await suspendWaiter(database, conductor, childId); diff --git a/packages/pgconductor-js/tests/mocks/database-client.mock.ts b/packages/pgconductor-js/tests/mocks/database-client.mock.ts index dcd5f39..2aa461a 100644 --- a/packages/pgconductor-js/tests/mocks/database-client.mock.ts +++ b/packages/pgconductor-js/tests/mocks/database-client.mock.ts @@ -33,7 +33,7 @@ export class MockDatabaseClient implements IDatabaseClient { saveStep = vi.fn(async (_args) => true); startExecution = vi.fn(async () => "mock-started-id"); checkMetadata = vi.fn(async () => {}); - clearWaitingState = vi.fn(async () => {}); + clearWaitingState = vi.fn(async () => true); registerEventWait = vi.fn(async () => null); registerEventRace = vi.fn(async () => null); cancelExecution = vi.fn(async () => true);