fix(waits): enforce wait deadlines, fail empty races and memoize child timeouts

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

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.
 

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<void>
 ```
 
@@ -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 "<stepKey>" 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: <error>` 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 `<stepKey>:result`.

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:**
 

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

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,21 +1882,32 @@ 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 (
               select 1 from pgconductor._private_steps step
               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

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`);
 			}