fix(cancellation): cascade from a fresh graph and take cancellation locks at once

AGENTS.md

@@ -214,8 +214,15 @@ Tables live in the `pgconductor` schema with a `_private_` prefix (`_private_exe
 
 **Cancel With Parent**
 - `cancel_with_parent` on a child row links it to `parent_execution_id` without a waiting edge
-- `cancel_execution()` on a pending execution cancels the linked children of every row it failed by calling itself, with that row's reason. A running execution only gets the flag; its settlement cascades
-- Settlement cascades from `terminal_failures` (permanent failures, cancellations and failed waiting ancestors) in the same statement: `linked_descendants` follows linked children and, below the first level, the children they wait on. Running ones get the cancelled flag; pending ones fail as cancelled. A row in the same batch whose outcome would keep it alive (retry, sleep, child, signal re-queue) is cancelled instead, so those CTEs exclude `cancelled_descendants`
+- `_private_cancel_linked_children()` calls `cancel_execution()` for the linked children of ended executions, each with the reason its parent ended with. Every path that ends an execution calls it after its own writes: `cancel_execution()` for every row it failed, settlement for `terminal_failures` (permanent failures, cancellations and failed waiting ancestors) and `invoke()` for the execution it supersedes. A timed-out `invokeAndWait` (`clearWaitingState`) cancels the awaited child with `cancel_execution()`
+- `cancel_execution()` on a pending execution fails it with the executions it waits on and cancels their linked children; a running execution only gets the flag, and its settlement cascades
+- The cascade reads the graph with fresh snapshots, so it finds children started or suspended while the caller waited for a lock. Settlement calls it in the main query after the `writes` CTE, which runs every write of the statement to completion, so it sees a linked child settled in the same batch in its new state (a retried, sleeping or suspended child is cancelled; a completed one stays completed) and writes no row the statement writes later. A new write CTE in settlement must be added to `writes`
+
+**Lock Order**
+- An awaited execution is locked before the executions waiting on it (`waiting_on_execution_id` and `result_waiters`): settlement locks its claimed rows, then waiting parents, failed ancestors (child first) and result waiters; `clearWaitingState` cancels the child before it clears the parent
+- A parent is locked before its linked children: the cascade runs after the writes of the path that ended the parent
+- `cancel_execution()` and `_private_register_event_race()` lock their whole set at once with `_private_lock_executions()`: `for update nowait`, and when a row is busy, it waits for each row alone in a subtransaction it rolls back, then tries again. They never wait while holding part of the set, so its order does not matter. `cancel_execution()` includes the linked subtree of the rows it fails, so its calls for linked children find their locks held
+- Queue configuration: registration and `drop_queue()` lock queue rows in name order before tasks and subscriptions
 
 **Cron Schedules**
 - A claimed occurrence inserts its successor before the handler runs, through `_private_schedule_next_occurrence()`: it takes a share lock on the queue row, then inserts only if the occurrence is still claimed and not cancelled
@@ -252,9 +259,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 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`
 - 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 (settlement excludes `cancelled_descendants`), or the later update of that row fails or is lost
+- `_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
 - The delivery is a plpgsql function called only when a settled execution has waiters: settlement is planned on every call, and each reference to the partitioned executions table adds planning cost per queue
 - Waiting creates no parent link: a failed or cancelled execution never fails its waiter, and timeouts or cancelling the waiter leave it running. An unknown execution raises at registration

docs/content/crafting-tasks/child-invocation.md

@@ -61,7 +61,7 @@ const result = await ctx.invokeAndWait(
 );
 ```
 
-If the child doesn't complete within the timeout, `ctx.invokeAndWait()` throws a timeout error and the child is cancelled.
+If the child doesn't complete within the timeout, `ctx.invokeAndWait()` throws a timeout error and the child is cancelled with `Cancelled: parent timed out`, together with the executions it started with `cancelWithParent` and the children it waits on.
 
 ## Error Handling
 
@@ -136,7 +136,7 @@ Retrying the invoking attempt never starts a second execution, even if it crashe
 
 ### Cancelling with the Parent
 
-Pass `cancelWithParent: true` to cancel the execution together with the caller: when the caller is cancelled or fails permanently, the execution is cancelled with the same reason, whatever state it is in (pending, running or suspended in a sleep, wait or child). This applies transitively to executions it started with `cancelWithParent` and to the children it waits on:
+Pass `cancelWithParent: true` to cancel the execution together with the caller: when the caller is cancelled, fails permanently or is [superseded](../task-execution/deduplication.md#duplicate-keys), the execution is cancelled with the same reason, whatever state it is in (pending, running or suspended in a sleep, wait or child). This applies transitively to executions it started with `cancelWithParent` and to the children it waits on:
 
 ```typescript
 const session = conductor.createTask(

docs/content/task-execution/deduplication.md

@@ -29,7 +29,7 @@ What a repeated invoke does depends on the execution that holds the key:
 | Existing execution | Result |
 |--------------------|--------|
 | Waiting to run | Its payload, `run_at`, `priority` and `group` are replaced. The same ID is returned. A sleeping execution resumes right away with the new payload. |
-| Running | It is superseded: marked failed with `superseded by reinvoke`, and a new execution is created with the new payload. The running handler is not stopped, but its result is discarded. |
+| Running | It is superseded: marked failed with `superseded by reinvoke`, and a new execution is created with the new payload. The running handler is not stopped, but its result is discarded. Executions it started with `cancelWithParent` are cancelled. |
 | Completed or failed | Nothing runs. Its stored payload is replaced and the same ID is returned. |
 
 A key stays taken as long as its execution is stored. Once [retention](../crafting-tasks/retention.md) removes a completed or failed execution, the next invoke with that key creates a new execution.

migrations/0000000001_setup.sql

@@ -749,7 +749,7 @@ declare
     v_singleton_on timestamptz;
     v_next_singleton_on timestamptz;
     v_run_at timestamptz;
-    v_resolved integer;
+    v_superseded_id uuid;
 begin
   perform pgconductor._private_check_metadata(p_metadata);
   v_now := pgconductor._private_current_time();
@@ -784,14 +784,25 @@ begin
               last_error = 'superseded by reinvoke'
             from superseded s
           where e.id = s.id and e.queue = s.queue and s.should_remove is not true
+      ),
+      resolved as (
+          select pgconductor._private_resolve_result_waits(jsonb_agg(w.value || jsonb_build_object(
+              'outcome', jsonb_build_object('status', 'failed', 'error', 'superseded by reinvoke')
+          )))
+          from superseded s
+          cross join lateral jsonb_array_elements(s.result_waiters) w
+          having count(*) > 0
       )
-      select pgconductor._private_resolve_result_waits(jsonb_agg(w.value || jsonb_build_object(
-          'outcome', jsonb_build_object('status', 'failed', 'error', 'superseded by reinvoke')
-      )))
-      into v_resolved
+      select s.id
+      into v_superseded_id
       from superseded s
-      cross join lateral jsonb_array_elements(s.result_waiters) w
-      having count(*) > 0;
+      left join resolved on true;
+
+      if v_superseded_id is not null then
+          perform pgconductor._private_cancel_linked_children(jsonb_build_array(jsonb_build_object(
+              'execution_id', v_superseded_id, 'reason', 'superseded by reinvoke'
+          )));
+      end if;
   end if;
 
   -- singleton throttle/debounce logic
@@ -1070,6 +1081,58 @@ begin
 end;
 $function$;
 
+-- locks the executions in p_ids all at once, or waits without holding any of them: waiting for
+-- one while holding another can deadlock with a statement that locks them the other way round
+create or replace function pgconductor._private_lock_executions(p_ids uuid[])
+returns void
+language plpgsql
+volatile
+set search_path to ''
+as $function$
+declare
+  v_id uuid;
+begin
+  loop
+    begin
+      perform 1
+      from pgconductor._private_executions e
+      where e.id = any(p_ids)
+      for update of e nowait;
+      return;
+    exception when lock_not_available then
+      -- wait for each one alone, rolling back to release it, then try them all again
+      foreach v_id in array p_ids loop
+        begin
+          perform 1 from pgconductor._private_executions e where e.id = v_id for update of e;
+          raise exception using errcode = 'PGC01';
+        exception when sqlstate 'PGC01' then
+          null;
+        end;
+      end loop;
+    end;
+  end loop;
+end;
+$function$;
+
+-- cancels the linked children of ended executions ([{execution_id, reason}]) with the reason
+-- their parent ended with. callers run it after their own writes, and it reads the children
+-- with a fresh snapshot, so it also finds the ones started while the caller waited for a lock.
+create or replace function pgconductor._private_cancel_linked_children(p_parents jsonb)
+returns void
+language plpgsql
+volatile
+set search_path to ''
+as $function$
+begin
+  perform pgconductor.cancel_execution(child.id, parent.reason)
+  from jsonb_to_recordset(p_parents) as parent(execution_id uuid, reason text)
+  join pgconductor._private_executions child on child.parent_execution_id = parent.execution_id
+  where child.cancel_with_parent
+    and child.completed_at is null
+    and child.failed_at is null;
+end;
+$function$;
+
 -- cancel an execution
 create or replace function pgconductor.cancel_execution(
   p_execution_id uuid,
@@ -1085,21 +1148,21 @@ declare
   v_queue text;
   v_descendant_ids uuid[];
   v_ancestor_ids uuid[];
+  v_linked_ids uuid[];
   v_locked_ids uuid[];
   v_rows_affected integer;
   v_resolved integer;
 begin
-  -- a waiting execution is cancelled with the executions it waits on, and
-  -- fails the executions waiting on it. the graph may change while we wait for
-  -- locks, so it is read again after locking. a changed graph releases the
-  -- stale locks by rolling back, then locks the new graph children first.
+  -- a waiting execution is cancelled with the executions it waits on and its
+  -- linked children, and fails the executions waiting on it. the graph may
+  -- change while we wait for locks, so it is read again after locking. a
+  -- changed graph releases the stale locks by rolling back, then locks the new
+  -- graph. the whole graph is locked at once, so a settlement that locks a
+  -- child before the parent waiting on it cannot deadlock with this, and the
+  -- calls for linked children find their locks held.
   loop
     begin
-      perform 1
-      from unnest(v_locked_ids) with ordinality as chain(id, position)
-      join pgconductor._private_executions e on e.id = chain.id
-      order by chain.position
-      for update of e;
+      perform pgconductor._private_lock_executions(v_locked_ids);
 
       with recursive descendants as (
         select e.waiting_on_execution_id as id, 1 as depth
@@ -1125,8 +1188,28 @@ begin
       )
       select array_agg(id order by depth) into v_ancestor_ids from ancestors;
 
-      exit when v_locked_ids = v_descendant_ids || p_execution_id || v_ancestor_ids;
-      v_locked_ids := v_descendant_ids || p_execution_id || v_ancestor_ids;
+      -- a running execution is only flagged; its settlement cancels its linked children
+      with recursive linked as (
+        select child.id, child.waiting_on_execution_id
+        from pgconductor._private_executions e
+        join pgconductor._private_executions child
+          on child.parent_execution_id = e.id and child.cancel_with_parent
+        where e.id = any(v_descendant_ids || p_execution_id || v_ancestor_ids)
+          and (e.id <> p_execution_id or e.locked_by is null)
+          and child.completed_at is null
+          and child.failed_at is null
+        union all
+        select child.id, child.waiting_on_execution_id
+        from linked l
+        join pgconductor._private_executions child
+          on child.parent_execution_id = l.id
+          and (child.cancel_with_parent or child.id = l.waiting_on_execution_id)
+        where child.completed_at is null and child.failed_at is null
+      )
+      select array_agg(id order by id) into v_linked_ids from linked;
+
+      exit when v_locked_ids = v_linked_ids || v_descendant_ids || p_execution_id || v_ancestor_ids;
+      v_locked_ids := v_linked_ids || v_descendant_ids || p_execution_id || v_ancestor_ids;
       raise exception using errcode = 'PGC01';
     exception when sqlstate 'PGC01' then
       null;
@@ -1221,17 +1304,12 @@ begin
   cross join lateral jsonb_array_elements(p.result_waiters) w
   having count(*) > 0;
 
-  -- linked children are cancelled with the reason their parent ended with
-  perform pgconductor.cancel_execution(
-    e.id,
-    case when e.parent_execution_id = any(v_ancestor_ids) then 'Child execution failed: ' || p_reason
+  perform pgconductor._private_cancel_linked_children(jsonb_agg(jsonb_build_object(
+    'execution_id', id,
+    'reason', case when id = any(v_ancestor_ids) then 'Child execution failed: ' || p_reason
       else p_reason end
-  )
-  from pgconductor._private_executions e
-  where e.parent_execution_id = any(v_descendant_ids || p_execution_id || v_ancestor_ids)
-    and e.cancel_with_parent
-    and e.completed_at is null
-    and e.failed_at is null;
+  )))
+  from unnest(v_descendant_ids || p_execution_id || v_ancestor_ids) id;
 
   return true;
 end;
@@ -1302,6 +1380,18 @@ create or replace function pgconductor.drop_queue(queue_name text)
  volatile
  set search_path to ''
 as $function$
+  -- registration locks a queue before its tasks, so this locks the queue and the queues whose
+  -- tasks dead-letter into it first, in the same order
+  select 1
+  from pgconductor._private_queues queue
+  where queue.name = drop_queue.queue_name
+    or queue.name in (
+      select task.queue from pgconductor._private_tasks task
+      where task.dead_letter_queue = drop_queue.queue_name
+    )
+  order by queue.name
+  for update;
+
   update pgconductor._private_tasks
   set dead_letter_queue = null, dead_letter_task_key = null
   where dead_letter_queue = drop_queue.queue_name;
@@ -1709,13 +1799,10 @@ declare
     v_expires_at timestamptz;
     v_now timestamptz := pgconductor._private_current_time();
 begin
-    -- Settlement locks an execution before its waiters, so awaited executions are locked first
+    -- Settlement locks an execution before its waiters, so awaited executions are locked first,
+    -- all at once: one of them may be waiting on another
     if cardinality(v_awaited_ids) > 0 then
-        perform 1
-        from pgconductor._private_executions awaited
-        where awaited.id = any(v_awaited_ids)
-        order by awaited.id
-        for update;
+        perform pgconductor._private_lock_executions(v_awaited_ids);
     end if;
 
     perform 1