Skip to content

perf(settlement): plan settlement for one partition #103

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
psteinroe merged 1 commit into from
Oct 26, 2026
Merged

perf(settlement): plan settlement for one partition #103

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 4 additions & 3 deletions AGENTS.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -179,6 +179,7 @@ Tables live in the `pgconductor` schema with a `_private_` prefix (`_private_exe
**Important**: All core logic lives in Postgres. The TypeScript layer is intentionally thin - it validates input and orchestrates calls via `DatabaseClient`.

- Claiming (`buildGetExecutions`) and settlement (`buildReturnExecutions`) are single statements in `query-builder.ts`
- Settlement settles executions of one queue (`returnExecutions` takes the worker's `queueName`) and narrows every reference to the executions table to its partition. The queue is inlined as a literal, because a cached generic plan prunes partitions only by constants: each queue gets its own prepared statement, planned once per connection, whatever the number of queues. Reads of executions in other queues (waiting parents) run in functions called only when a settled execution has a parent: `_private_complete_children()` wakes the parents of completed children and returns the children no parent waits on, `_private_fail_waiting_parents()` fails waiting ancestors. Their statements reach every partition, so they force generic plans, made once per connection, and look rows up by arrays of ids to stay on indexes. They read with fresh snapshots, so they come first in `writes`. A new reference to executions in settlement must be narrowed to the queue or go into such a function
- SQL functions in the migration: `invoke()`, `invoke_batch()`, `cancel_execution()`, `signal_execution()`, `signal_with_start()`, `emit_event()`, `drop_queue()`, `_private_register_worker()`, `_private_current_time()`, `_private_portable_uuidv7()`, ...

### Key Design Patterns
Expand All @@ -198,7 +199,7 @@ Tables live in the `pgconductor` schema with a `_private_` prefix (`_private_exe

**Cascade Failures**
- When a child fails permanently (attempts >= max_attempts), its parent fails too
- Implemented in `buildReturnExecutions()` via the `permanently_failed_children` CTE
- Implemented by `_private_fail_waiting_parents()`, which settlement calls for the `permanently_failed_children` that have a parent
- Parent receives error like "Child execution failed: <child_error>"

**Claims and Retries**
Expand Down Expand Up @@ -273,7 +274,7 @@ Tables live in the `pgconductor` schema with a `_private_` prefix (`_private_exe
- 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
- 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
- The delivery is a plpgsql function called only when a settled execution has waiters, so settlement does not plan its references to executions of other queues
- 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

**Signals**
Expand Down Expand Up @@ -344,7 +345,7 @@ Guidelines for SQL:

Common issues:
- **Timing issues**: Use fake time instead of waiting (backoff schedule is 15s, 30s, 60s...)
- **Cascade failures**: Check `permanently_failed_children` CTE logic
- **Cascade failures**: Check the `permanently_failed_children` CTE and `_private_fail_waiting_parents()`
- **Infinity serialization**: postgres.js serializes infinity as null in JSON

Inspect database state in tests:
Expand Down
145 changes: 145 additions & 0 deletions migrations/0000000001_setup.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1248,6 +1248,151 @@ begin
end;
$function$;

-- settlement of completed children ([{execution_id, result}]) that have a parent: stores the result
-- for the parents waiting on them and wakes those parents. returns the children no execution waits
-- on any more. parents may be in any queue, so settlement calls this only when needed instead of
-- planning these lookups on every call. they cannot be narrowed to a partition, so their plans are
-- generic and made once per connection. the input is estimated at 100 rows, so lookups use arrays
-- of ids to stay on indexes.
create or replace function pgconductor._private_complete_children(p_children jsonb)
returns setof uuid
language plpgsql
volatile
set search_path to ''
set plan_cache_mode to force_generic_plan
as $function$
declare
v_ids uuid[] := array(
select c.execution_id from jsonb_to_recordset(p_children) as c(execution_id uuid)
);
v_orphans uuid[];
begin
v_orphans := array(
select unnest(v_ids)
except
select parent.waiting_on_execution_id
from pgconductor._private_executions parent
where parent.waiting_on_execution_id = any(v_ids)
);

-- a completed child may wake only a parent which is still waiting and is not
-- currently claimed. the row lock makes this check race-safe.
with parents as materialized (
select parent.id, parent.queue, parent.waiting_step_key, c.result
from pgconductor._private_executions parent
join jsonb_to_recordset(p_children) as c(execution_id uuid, result jsonb)
on c.execution_id = parent.waiting_on_execution_id
where parent.waiting_on_execution_id = any(v_ids)
and parent.completed_at is null
and parent.failed_at is null
and parent.locked_by is null
for update of parent
),
parent_steps as (
insert into pgconductor._private_steps (execution_id, queue, key, result)
select p.id, p.queue, p.waiting_step_key,
jsonb_build_object('status', 'completed', 'result', p.result)
from parents p
where p.waiting_step_key is not null
on conflict (execution_id, key) do nothing
)
update pgconductor._private_executions e
set run_at = pgconductor._private_current_time(), waiting_on_execution_id = null,
waiting_step_key = null, locked_by = null, locked_at = null
where e.id = any(array(select p.id from parents p));

return query select unnest(v_orphans);
end;
$function$;

-- settlement of terminally failed children ([{execution_id, error, cancelled}]) that have a parent:
-- fails every execution still waiting on them, and the ones waiting on those, and returns them for
-- dead-lettering, result waits and the cancel cascade. like _private_complete_children, settlement
-- calls this only when needed.
create or replace function pgconductor._private_fail_waiting_parents(p_children jsonb)
returns table (
execution_id uuid, queue text, task_key text, execution_group text, payload jsonb,
failure_error text, failure_attempts integer, execution_cancelled boolean,
dead_letter_queue text, dead_letter_task_key text, trace_context jsonb, metadata jsonb,
result_waiters jsonb
)
language plpgsql
volatile
set search_path to ''
set plan_cache_mode to force_generic_plan
as $function$
declare
v_ids uuid[] := array(
select c.execution_id from jsonb_to_recordset(p_children) as c(execution_id uuid)
);
v_ancestors uuid[] := '{}';
begin
loop
v_ids := array(
select parent.id
from pgconductor._private_executions parent
where parent.waiting_on_execution_id = any(v_ids)
and parent.completed_at is null and parent.failed_at is null and parent.locked_by is null
);
exit when cardinality(v_ids) = 0;
v_ancestors := v_ancestors || v_ids;
end loop;

-- ancestors are locked child first, then the chain is followed again on the locked rows
return query
with recursive locked_ancestors as materialized (
select parent.id, parent.queue, parent.waiting_on_execution_id, parent.task_key,
parent."group", parent.payload, parent.attempts, parent.trace_context, parent.metadata,
parent.result_waiters,
t.remove_on_fail_days = 0 as should_remove,
t.dead_letter_queue, t.dead_letter_task_key
from pgconductor._private_executions parent
join pgconductor._private_tasks t on t.key = parent.task_key and t.queue = parent.queue
where parent.id = any(v_ancestors)
and parent.completed_at is null and parent.failed_at is null and parent.locked_by is null
order by array_position(v_ancestors, parent.id)
for update of parent
),
targets as (
select c.error, c.cancelled, a.*
from jsonb_to_recordset(p_children) as c(execution_id uuid, error text, cancelled boolean)
join locked_ancestors a on a.waiting_on_execution_id = c.execution_id
union all
select t.error, t.cancelled, a.*
from targets t
join locked_ancestors a on a.waiting_on_execution_id = t.id
),
removed as (
delete from pgconductor._private_executions e
using targets t
where e.id = any(v_ancestors) and e.id = t.id and e.queue = t.queue
and t.should_remove and e.locked_by is null
),
failed as (
update pgconductor._private_executions e
set failed_at = pgconductor._private_current_time(),
last_error = 'Child execution failed: ' || t.error,
waiting_on_execution_id = null, waiting_step_key = null,
locked_by = null, locked_at = null
from targets t
where e.id = any(v_ancestors) and e.id = t.id and e.queue = t.queue
and t.should_remove is not true
returning e.id, e.queue
),
deleted_waits as (
delete from pgconductor._private_custom_event_subscriptions subscription
using failed f
where subscription.kind = 'execution_wait'
and subscription.execution_id = f.id
and subscription.queue = f.queue
)
select t.id, t.queue, t.task_key, t."group", t.payload, 'Child execution failed: ' || t.error,
t.attempts, t.cancelled, t.dead_letter_queue, t.dead_letter_task_key, t.trace_context,
t.metadata, t.result_waiters
from targets t;
end;
$function$;

-- cancel an execution
create or replace function pgconductor.cancel_execution(
p_execution_id uuid,
Expand Down
7 changes: 2 additions & 5 deletions packages/pgconductor-js/src/database-client.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -133,12 +133,13 @@ export type ExecutionResult =

export type ReturnExecutionsArgs = {
orchestratorId: string;
/** The queue of every settled execution */
queueName: string;
results: ExecutionResult[];
};

export type ExecutionCompleted = {
execution_id: string;
queue: string;
task_key: string;
claim_token: string;
status: "completed";
Expand All @@ -149,7 +150,6 @@ export type ExecutionCompleted = {

export type ExecutionFailed = {
execution_id: string;
queue: string;
task_key: string;
claim_token: string;
status: "failed";
Expand All @@ -158,7 +158,6 @@ export type ExecutionFailed = {

export type ExecutionReleased = {
execution_id: string;
queue: string;
task_key: string;
claim_token: string;
status: "released";
Expand All @@ -168,7 +167,6 @@ export type ExecutionReleased = {

export type ExecutionPermamentlyFailed = {
execution_id: string;
queue: string;
task_key: string;
claim_token: string;
status: "permanently_failed";
Expand All @@ -178,7 +176,6 @@ export type ExecutionPermamentlyFailed = {
export type ExecutionInvokeChild = {
group?: string | null;
execution_id: string;
queue: string;
task_key: string;
claim_token: string;
status: "invoke_child";
Expand Down
145 changes: 145 additions & 0 deletions packages/pgconductor-js/src/generated/sql.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1262,6 +1262,151 @@ begin
end;
$function$;

-- settlement of completed children ([{execution_id, result}]) that have a parent: stores the result
-- for the parents waiting on them and wakes those parents. returns the children no execution waits
-- on any more. parents may be in any queue, so settlement calls this only when needed instead of
-- planning these lookups on every call. they cannot be narrowed to a partition, so their plans are
-- generic and made once per connection. the input is estimated at 100 rows, so lookups use arrays
-- of ids to stay on indexes.
create or replace function pgconductor._private_complete_children(p_children jsonb)
returns setof uuid
language plpgsql
volatile
set search_path to ''
set plan_cache_mode to force_generic_plan
as $function$
declare
v_ids uuid[] := array(
select c.execution_id from jsonb_to_recordset(p_children) as c(execution_id uuid)
);
v_orphans uuid[];
begin
v_orphans := array(
select unnest(v_ids)
except
select parent.waiting_on_execution_id
from pgconductor._private_executions parent
where parent.waiting_on_execution_id = any(v_ids)
);

-- a completed child may wake only a parent which is still waiting and is not
-- currently claimed. the row lock makes this check race-safe.
with parents as materialized (
select parent.id, parent.queue, parent.waiting_step_key, c.result
from pgconductor._private_executions parent
join jsonb_to_recordset(p_children) as c(execution_id uuid, result jsonb)
on c.execution_id = parent.waiting_on_execution_id
where parent.waiting_on_execution_id = any(v_ids)
and parent.completed_at is null
and parent.failed_at is null
and parent.locked_by is null
for update of parent
),
parent_steps as (
insert into pgconductor._private_steps (execution_id, queue, key, result)
select p.id, p.queue, p.waiting_step_key,
jsonb_build_object('status', 'completed', 'result', p.result)
from parents p
where p.waiting_step_key is not null
on conflict (execution_id, key) do nothing
)
update pgconductor._private_executions e
set run_at = pgconductor._private_current_time(), waiting_on_execution_id = null,
waiting_step_key = null, locked_by = null, locked_at = null
where e.id = any(array(select p.id from parents p));

return query select unnest(v_orphans);
end;
$function$;

-- settlement of terminally failed children ([{execution_id, error, cancelled}]) that have a parent:
-- fails every execution still waiting on them, and the ones waiting on those, and returns them for
-- dead-lettering, result waits and the cancel cascade. like _private_complete_children, settlement
-- calls this only when needed.
create or replace function pgconductor._private_fail_waiting_parents(p_children jsonb)
returns table (
execution_id uuid, queue text, task_key text, execution_group text, payload jsonb,
failure_error text, failure_attempts integer, execution_cancelled boolean,
dead_letter_queue text, dead_letter_task_key text, trace_context jsonb, metadata jsonb,
result_waiters jsonb
)
language plpgsql
volatile
set search_path to ''
set plan_cache_mode to force_generic_plan
as $function$
declare
v_ids uuid[] := array(
select c.execution_id from jsonb_to_recordset(p_children) as c(execution_id uuid)
);
v_ancestors uuid[] := '{}';
begin
loop
v_ids := array(
select parent.id
from pgconductor._private_executions parent
where parent.waiting_on_execution_id = any(v_ids)
and parent.completed_at is null and parent.failed_at is null and parent.locked_by is null
);
exit when cardinality(v_ids) = 0;
v_ancestors := v_ancestors || v_ids;
end loop;

-- ancestors are locked child first, then the chain is followed again on the locked rows
return query
with recursive locked_ancestors as materialized (
select parent.id, parent.queue, parent.waiting_on_execution_id, parent.task_key,
parent."group", parent.payload, parent.attempts, parent.trace_context, parent.metadata,
parent.result_waiters,
t.remove_on_fail_days = 0 as should_remove,
t.dead_letter_queue, t.dead_letter_task_key
from pgconductor._private_executions parent
join pgconductor._private_tasks t on t.key = parent.task_key and t.queue = parent.queue
where parent.id = any(v_ancestors)
and parent.completed_at is null and parent.failed_at is null and parent.locked_by is null
order by array_position(v_ancestors, parent.id)
for update of parent
),
targets as (
select c.error, c.cancelled, a.*
from jsonb_to_recordset(p_children) as c(execution_id uuid, error text, cancelled boolean)
join locked_ancestors a on a.waiting_on_execution_id = c.execution_id
union all
select t.error, t.cancelled, a.*
from targets t
join locked_ancestors a on a.waiting_on_execution_id = t.id
),
removed as (
delete from pgconductor._private_executions e
using targets t
where e.id = any(v_ancestors) and e.id = t.id and e.queue = t.queue
and t.should_remove and e.locked_by is null
),
failed as (
update pgconductor._private_executions e
set failed_at = pgconductor._private_current_time(),
last_error = 'Child execution failed: ' || t.error,
waiting_on_execution_id = null, waiting_step_key = null,
locked_by = null, locked_at = null
from targets t
where e.id = any(v_ancestors) and e.id = t.id and e.queue = t.queue
and t.should_remove is not true
returning e.id, e.queue
),
deleted_waits as (
delete from pgconductor._private_custom_event_subscriptions subscription
using failed f
where subscription.kind = 'execution_wait'
and subscription.execution_id = f.id
and subscription.queue = f.queue
)
select t.id, t.queue, t.task_key, t."group", t.payload, 'Child execution failed: ' || t.error,
t.attempts, t.cancelled, t.dead_letter_queue, t.dead_letter_task_key, t.trace_context,
t.metadata, t.result_waiters
from targets t;
end;
$function$;

-- cancel an execution
create or replace function pgconductor.cancel_execution(
p_execution_id uuid,
Expand Down
Loading
Loading