From 2efe73f1db7e71fc82d5c51c1295ba2ecc2c5d7a Mon Sep 17 00:00:00 2001 From: psteinroe Date: Tue, 15 Sep 2026 11:47:40 +0000 Subject: [PATCH 1/3] feat(scheduling): add soft group concurrency --- migrations/0000000001_setup.sql | 164 +++---- packages/pgconductor-js/src/conductor.ts | 1 + .../pgconductor-js/src/database-client.ts | 10 +- packages/pgconductor-js/src/generated/sql.ts | 164 +++---- packages/pgconductor-js/src/lib/assert.ts | 7 + packages/pgconductor-js/src/query-builder.ts | 419 +++++------------- packages/pgconductor-js/src/task-context.ts | 5 + .../pgconductor-js/src/task-definition.ts | 2 +- packages/pgconductor-js/src/task.ts | 6 +- packages/pgconductor-js/src/worker.ts | 23 +- .../tests/integration/concurrency.test.ts | 62 --- .../integration/execution-foundations.test.ts | 62 --- .../integration/group-concurrency.test.ts | 373 ++++++++++++++++ .../tests/mocks/in-memory-database-client.ts | 91 ++-- .../unit/in-memory-database-client.test.ts | 22 - .../tests/unit/throttle-debounce.test.ts | 1 - 16 files changed, 689 insertions(+), 723 deletions(-) create mode 100644 packages/pgconductor-js/tests/integration/group-concurrency.test.ts diff --git a/migrations/0000000001_setup.sql b/migrations/0000000001_setup.sql index 3a0a25b..3284fc6 100644 --- a/migrations/0000000001_setup.sql +++ b/migrations/0000000001_setup.sql @@ -95,7 +95,7 @@ create table pgconductor._private_executions ( run_at timestamptz default pgconductor._private_current_time() not null, locked_at timestamptz, locked_by uuid, - slot_group_number integer, + "group" text, is_available boolean generated always as (locked_at is null and failed_at is null and completed_at is null) stored not null, attempts integer default 0 not null, last_error text, @@ -143,25 +143,18 @@ create table pgconductor._private_tasks ( ) ), - -- concurrency control: maximum number of concurrent executions across all workers + -- concurrency controls are intentionally soft and coordinated at claim time -- NULL means no limit (unlimited concurrency) concurrency_limit integer, + group_concurrency_limit integer, + constraint positive_concurrency_limits check ( + (concurrency_limit is null or concurrency_limit > 0) and + (group_concurrency_limit is null or group_concurrency_limit > 0) + ), primary key (queue, key) ); -create table pgconductor._private_concurrency_slots ( - task_key text not null, - queue text not null, - slot_group_number integer not null, - capacity integer not null, - used integer default 0 not null, - primary key (queue, task_key, slot_group_number) -); - -create index idx_slots_claim - on pgconductor._private_concurrency_slots (queue, task_key, capacity, used); - create table pgconductor._private_steps ( id uuid default pgconductor._private_portable_uuidv7() primary key, key text not null, @@ -254,6 +247,19 @@ begin v_partition_name ); + -- indexes used to count active executions for soft concurrency limits + execute format( + 'create index if not exists %I on pgconductor.%I (task_key) where locked_at is not null and failed_at is null and completed_at is null', + 'idx_' || v_partition_name || '_active_task', + v_partition_name + ); + + execute format( + 'create index if not exists %I on pgconductor.%I (task_key, "group") where "group" is not null and locked_at is not null and failed_at is null and completed_at is null', + 'idx_' || v_partition_name || '_active_task_group', + v_partition_name + ); + RETURN NEW; elsif tg_op = 'UPDATE' then @@ -316,7 +322,8 @@ create type pgconductor.execution_spec as ( dedupe_seconds integer, dedupe_next_slot boolean, cron_expression text, - priority integer + priority integer, + "group" text ); create type pgconductor.task_spec as ( @@ -327,7 +334,8 @@ create type pgconductor.task_spec as ( remove_on_fail_days integer, window_start timetz, window_end timetz, - concurrency_limit integer + concurrency_limit integer, + group_concurrency_limit integer ); create type pgconductor._private_event_operation as enum ( @@ -366,7 +374,7 @@ begin on conflict (name) do nothing; -- step 2: register/update tasks - insert into pgconductor._private_tasks (key, queue, max_attempts, remove_on_complete_days, remove_on_fail_days, window_start, window_end, concurrency_limit) + insert into pgconductor._private_tasks (key, queue, max_attempts, remove_on_complete_days, remove_on_fail_days, window_start, window_end, concurrency_limit, group_concurrency_limit) select spec.key, coalesce(spec.queue, 'default'), @@ -375,7 +383,8 @@ begin spec.remove_on_fail_days, spec.window_start, spec.window_end, - spec.concurrency_limit + spec.concurrency_limit, + spec.group_concurrency_limit from unnest(p_task_specs) as spec on conflict (queue, key) do update set @@ -385,56 +394,26 @@ begin remove_on_fail_days = excluded.remove_on_fail_days, window_start = excluded.window_start, window_end = excluded.window_end, - concurrency_limit = excluded.concurrency_limit; + concurrency_limit = excluded.concurrency_limit, + group_concurrency_limit = excluded.group_concurrency_limit; - -- step 2a: manage concurrency slots - -- create one row per slot (capacity=1 each) - insert into pgconductor._private_concurrency_slots (task_key, queue, slot_group_number, capacity, used) - select - spec.key, - coalesce(spec.queue, p_queue_name), - slot_num, - 1 as capacity, - 0 as used - from unnest(p_task_specs) as spec - cross join lateral generate_series(1, spec.concurrency_limit) as slot_num - where spec.concurrency_limit is not null - on conflict (queue, task_key, slot_group_number) - do update set - capacity = excluded.capacity, - used = least(pgconductor._private_concurrency_slots.used, excluded.capacity); - - -- clean up orphaned slots (tasks removed or concurrency_limit set to null) - delete from pgconductor._private_concurrency_slots - where queue = p_queue_name - and used = 0 - and (task_key, queue) not in ( - select key, queue from pgconductor._private_tasks - where concurrency_limit is not null - ); - - -- clean up excess slots when concurrency decreased - delete from pgconductor._private_concurrency_slots cs - where cs.queue = p_queue_name - and cs.used = 0 - and cs.slot_group_number > ( - select concurrency_limit - from pgconductor._private_tasks t - where t.key = cs.task_key and t.queue = cs.queue - ); - - -- step 3: insert scheduled cron executions (on conflict do nothing) - insert into pgconductor._private_executions (task_key, queue, payload, run_at, dedupe_key, cron_expression) + -- step 3: insert scheduled cron executions + insert into pgconductor._private_executions (task_key, queue, payload, run_at, dedupe_key, cron_expression, "group") select spec.task_key, coalesce(spec.queue, 'default'), coalesce(spec.payload, '{}'::jsonb), coalesce(spec.run_at, pgconductor._private_current_time()), spec.dedupe_key, - spec.cron_expression + spec.cron_expression, + spec."group" from unnest(p_cron_schedules) as spec where spec.dedupe_key is not null - on conflict (task_key, dedupe_key, queue) do nothing; + on conflict (task_key, dedupe_key, queue) do update set + payload = excluded.payload, + run_at = excluded.run_at, + cron_expression = excluded.cron_expression, + "group" = excluded."group"; -- step 4: clean up stale schedules for this queue -- delete future executions for schedules that no longer exist @@ -536,7 +515,7 @@ begin -- clear locked dedupe keys before batch insert with superseded as ( - select e.id, e.queue, e.task_key, e.slot_group_number + select e.id, e.queue, e.task_key from pgconductor._private_executions as e cross join unnest(specs) as spec where e.dedupe_key = spec.dedupe_key @@ -545,15 +524,6 @@ begin and e.locked_at is not null and spec.dedupe_key is not null for update of e - ), - released_slots as ( - update pgconductor._private_concurrency_slots cs - set used = 0 - from superseded s - where cs.queue = s.queue - and cs.task_key = s.task_key - and cs.slot_group_number = s.slot_group_number - and s.slot_group_number is not null ) update pgconductor._private_executions e set @@ -561,8 +531,7 @@ begin locked_by = null, locked_at = null, failed_at = v_now, - last_error = 'superseded by reinvoke', - slot_group_number = null + last_error = 'superseded by reinvoke' from superseded s where e.id = s.id; @@ -579,7 +548,8 @@ begin dedupe_key, singleton_on, cron_expression, - priority + priority, + "group" ) select pgconductor._private_portable_uuidv7(), @@ -598,14 +568,16 @@ begin else null end, spec.cron_expression, - coalesce(spec.priority, 0) + coalesce(spec.priority, 0), + spec."group" from unnest(specs) as spec on conflict (task_key, dedupe_key, queue) do update set payload = excluded.payload, run_at = excluded.run_at, priority = excluded.priority, cron_expression = excluded.cron_expression, - singleton_on = excluded.singleton_on + singleton_on = excluded.singleton_on, + "group" = excluded."group" returning pgconductor._private_executions.id; end; $function$ @@ -620,7 +592,8 @@ create or replace function pgconductor.invoke( p_dedupe_seconds integer default null, p_dedupe_next_slot boolean default false, p_cron_expression text default null, - p_priority integer default null + p_priority integer default null, + p_group text default null ) returns table(id uuid) language plpgsql @@ -640,22 +613,13 @@ begin -- clear locked dedupe key before insert (supersede pattern) if p_dedupe_key is not null then with superseded as ( - select e.id, e.queue, e.task_key, e.slot_group_number + select e.id, e.queue, e.task_key from pgconductor._private_executions e where e.dedupe_key = p_dedupe_key and e.task_key = p_task_key and e.queue = p_queue and e.locked_at is not null for update of e - ), - released_slots as ( - update pgconductor._private_concurrency_slots cs - set used = 0 - from superseded s - where cs.queue = s.queue - and cs.task_key = s.task_key - and cs.slot_group_number = s.slot_group_number - and s.slot_group_number is not null ) update pgconductor._private_executions e set @@ -663,9 +627,8 @@ begin locked_by = null, locked_at = null, failed_at = v_now, - last_error = 'superseded by reinvoke', - slot_group_number = null - from superseded s + last_error = 'superseded by reinvoke' + from superseded s where e.id = s.id; end if; @@ -690,7 +653,8 @@ begin dedupe_key, singleton_on, cron_expression, - priority + priority, + "group" ) values ( pgconductor._private_portable_uuidv7(), p_task_key, @@ -700,7 +664,8 @@ begin p_dedupe_key, v_singleton_on, p_cron_expression, - coalesce(p_priority, 0) + coalesce(p_priority, 0), + p_group ) on conflict (task_key, singleton_on, coalesce(dedupe_key, ''), queue) where singleton_on is not null and completed_at is null and failed_at is null and cancelled = false @@ -721,7 +686,8 @@ begin dedupe_key, singleton_on, cron_expression, - priority + priority, + "group" ) values ( pgconductor._private_portable_uuidv7(), p_task_key, @@ -731,14 +697,17 @@ begin p_dedupe_key, v_next_singleton_on, p_cron_expression, - coalesce(p_priority, 0) + coalesce(p_priority, 0), + p_group ) on conflict (task_key, singleton_on, coalesce(dedupe_key, ''), queue) where singleton_on is not null and completed_at is null and failed_at is null and cancelled = false do update set payload = excluded.payload, run_at = excluded.run_at, - priority = excluded.priority + priority = excluded.priority, + cron_expression = excluded.cron_expression, + "group" = excluded."group" returning _private_executions.id; return; end if; @@ -753,7 +722,8 @@ begin run_at, dedupe_key, cron_expression, - priority + priority, + "group" ) values ( pgconductor._private_portable_uuidv7(), p_task_key, @@ -762,13 +732,15 @@ begin v_run_at, p_dedupe_key, p_cron_expression, - coalesce(p_priority, 0) + coalesce(p_priority, 0), + p_group ) on conflict (task_key, dedupe_key, queue) do update set payload = excluded.payload, run_at = excluded.run_at, priority = excluded.priority, - cron_expression = excluded.cron_expression + cron_expression = excluded.cron_expression, + "group" = excluded."group" returning e.id; end; $function$ diff --git a/packages/pgconductor-js/src/conductor.ts b/packages/pgconductor-js/src/conductor.ts index ce75afc..26c6fb4 100644 --- a/packages/pgconductor-js/src/conductor.ts +++ b/packages/pgconductor-js/src/conductor.ts @@ -270,6 +270,7 @@ export class Conductor< debounce: item.debounce, cron_expression: item.cron_expression, priority: item.priority, + group: item.group, })); return this.db.invokeBatch(specs); } diff --git a/packages/pgconductor-js/src/database-client.ts b/packages/pgconductor-js/src/database-client.ts index ee928ab..c2b523b 100644 --- a/packages/pgconductor-js/src/database-client.ts +++ b/packages/pgconductor-js/src/database-client.ts @@ -33,6 +33,7 @@ export interface ExecutionSpec { debounce?: { seconds: number } | null; cron_expression?: string | null; priority?: number | null; + group?: string | null; parent_execution_id?: string | null; parent_step_key?: string | null; parent_timeout_ms?: number | null; @@ -46,6 +47,7 @@ export interface TaskSpec { removeOnFailDays?: number | null; window?: [string, string] | null; concurrency?: number | null; + groupConcurrency?: number | null; } export interface Execution { @@ -60,7 +62,7 @@ export interface Execution { last_error: string | null; dedupe_key?: string | null; cron_expression?: string | null; - slot_group_number?: number | null; + group?: string | null; } // todo: move all of this to query-builder too or create new types.ts file @@ -89,7 +91,6 @@ export interface ExecutionCompleted { task_key: string; status: "completed"; result?: Payload; - slot_group_number?: number | null; } export interface ExecutionFailed { @@ -99,7 +100,6 @@ export interface ExecutionFailed { task_key: string; status: "failed"; error: string; - slot_group_number?: number | null; } export interface ExecutionReleased { @@ -110,7 +110,6 @@ export interface ExecutionReleased { status: "released"; reschedule_in_ms?: number | "infinity"; step_key?: string; - slot_group_number?: number | null; } export interface ExecutionPermamentlyFailed { @@ -120,10 +119,10 @@ export interface ExecutionPermamentlyFailed { task_key: string; status: "permanently_failed"; error: string; - slot_group_number?: number | null; } export interface ExecutionInvokeChild { + group?: string | null; execution_id: string; queue: string; orchestrator_id: string; @@ -134,7 +133,6 @@ export interface ExecutionInvokeChild { child_task_name: string; child_task_queue: string; child_payload: Payload | null; - slot_group_number?: number | null; } export interface EventSubscriptionSpec { diff --git a/packages/pgconductor-js/src/generated/sql.ts b/packages/pgconductor-js/src/generated/sql.ts index 05bd116..f73bae2 100644 --- a/packages/pgconductor-js/src/generated/sql.ts +++ b/packages/pgconductor-js/src/generated/sql.ts @@ -111,7 +111,7 @@ create table pgconductor._private_executions ( run_at timestamptz default pgconductor._private_current_time() not null, locked_at timestamptz, locked_by uuid, - slot_group_number integer, + "group" text, is_available boolean generated always as (locked_at is null and failed_at is null and completed_at is null) stored not null, attempts integer default 0 not null, last_error text, @@ -159,25 +159,18 @@ create table pgconductor._private_tasks ( ) ), - -- concurrency control: maximum number of concurrent executions across all workers + -- concurrency controls are intentionally soft and coordinated at claim time -- NULL means no limit (unlimited concurrency) concurrency_limit integer, + group_concurrency_limit integer, + constraint positive_concurrency_limits check ( + (concurrency_limit is null or concurrency_limit > 0) and + (group_concurrency_limit is null or group_concurrency_limit > 0) + ), primary key (queue, key) ); -create table pgconductor._private_concurrency_slots ( - task_key text not null, - queue text not null, - slot_group_number integer not null, - capacity integer not null, - used integer default 0 not null, - primary key (queue, task_key, slot_group_number) -); - -create index idx_slots_claim - on pgconductor._private_concurrency_slots (queue, task_key, capacity, used); - create table pgconductor._private_steps ( id uuid default pgconductor._private_portable_uuidv7() primary key, key text not null, @@ -270,6 +263,19 @@ begin v_partition_name ); + -- indexes used to count active executions for soft concurrency limits + execute format( + 'create index if not exists %I on pgconductor.%I (task_key) where locked_at is not null and failed_at is null and completed_at is null', + 'idx_' || v_partition_name || '_active_task', + v_partition_name + ); + + execute format( + 'create index if not exists %I on pgconductor.%I (task_key, "group") where "group" is not null and locked_at is not null and failed_at is null and completed_at is null', + 'idx_' || v_partition_name || '_active_task_group', + v_partition_name + ); + RETURN NEW; elsif tg_op = 'UPDATE' then @@ -332,7 +338,8 @@ create type pgconductor.execution_spec as ( dedupe_seconds integer, dedupe_next_slot boolean, cron_expression text, - priority integer + priority integer, + "group" text ); create type pgconductor.task_spec as ( @@ -343,7 +350,8 @@ create type pgconductor.task_spec as ( remove_on_fail_days integer, window_start timetz, window_end timetz, - concurrency_limit integer + concurrency_limit integer, + group_concurrency_limit integer ); create type pgconductor._private_event_operation as enum ( @@ -382,7 +390,7 @@ begin on conflict (name) do nothing; -- step 2: register/update tasks - insert into pgconductor._private_tasks (key, queue, max_attempts, remove_on_complete_days, remove_on_fail_days, window_start, window_end, concurrency_limit) + insert into pgconductor._private_tasks (key, queue, max_attempts, remove_on_complete_days, remove_on_fail_days, window_start, window_end, concurrency_limit, group_concurrency_limit) select spec.key, coalesce(spec.queue, 'default'), @@ -391,7 +399,8 @@ begin spec.remove_on_fail_days, spec.window_start, spec.window_end, - spec.concurrency_limit + spec.concurrency_limit, + spec.group_concurrency_limit from unnest(p_task_specs) as spec on conflict (queue, key) do update set @@ -401,56 +410,26 @@ begin remove_on_fail_days = excluded.remove_on_fail_days, window_start = excluded.window_start, window_end = excluded.window_end, - concurrency_limit = excluded.concurrency_limit; + concurrency_limit = excluded.concurrency_limit, + group_concurrency_limit = excluded.group_concurrency_limit; - -- step 2a: manage concurrency slots - -- create one row per slot (capacity=1 each) - insert into pgconductor._private_concurrency_slots (task_key, queue, slot_group_number, capacity, used) - select - spec.key, - coalesce(spec.queue, p_queue_name), - slot_num, - 1 as capacity, - 0 as used - from unnest(p_task_specs) as spec - cross join lateral generate_series(1, spec.concurrency_limit) as slot_num - where spec.concurrency_limit is not null - on conflict (queue, task_key, slot_group_number) - do update set - capacity = excluded.capacity, - used = least(pgconductor._private_concurrency_slots.used, excluded.capacity); - - -- clean up orphaned slots (tasks removed or concurrency_limit set to null) - delete from pgconductor._private_concurrency_slots - where queue = p_queue_name - and used = 0 - and (task_key, queue) not in ( - select key, queue from pgconductor._private_tasks - where concurrency_limit is not null - ); - - -- clean up excess slots when concurrency decreased - delete from pgconductor._private_concurrency_slots cs - where cs.queue = p_queue_name - and cs.used = 0 - and cs.slot_group_number > ( - select concurrency_limit - from pgconductor._private_tasks t - where t.key = cs.task_key and t.queue = cs.queue - ); - - -- step 3: insert scheduled cron executions (on conflict do nothing) - insert into pgconductor._private_executions (task_key, queue, payload, run_at, dedupe_key, cron_expression) + -- step 3: insert scheduled cron executions + insert into pgconductor._private_executions (task_key, queue, payload, run_at, dedupe_key, cron_expression, "group") select spec.task_key, coalesce(spec.queue, 'default'), coalesce(spec.payload, '{}'::jsonb), coalesce(spec.run_at, pgconductor._private_current_time()), spec.dedupe_key, - spec.cron_expression + spec.cron_expression, + spec."group" from unnest(p_cron_schedules) as spec where spec.dedupe_key is not null - on conflict (task_key, dedupe_key, queue) do nothing; + on conflict (task_key, dedupe_key, queue) do update set + payload = excluded.payload, + run_at = excluded.run_at, + cron_expression = excluded.cron_expression, + "group" = excluded."group"; -- step 4: clean up stale schedules for this queue -- delete future executions for schedules that no longer exist @@ -552,7 +531,7 @@ begin -- clear locked dedupe keys before batch insert with superseded as ( - select e.id, e.queue, e.task_key, e.slot_group_number + select e.id, e.queue, e.task_key from pgconductor._private_executions as e cross join unnest(specs) as spec where e.dedupe_key = spec.dedupe_key @@ -561,15 +540,6 @@ begin and e.locked_at is not null and spec.dedupe_key is not null for update of e - ), - released_slots as ( - update pgconductor._private_concurrency_slots cs - set used = 0 - from superseded s - where cs.queue = s.queue - and cs.task_key = s.task_key - and cs.slot_group_number = s.slot_group_number - and s.slot_group_number is not null ) update pgconductor._private_executions e set @@ -577,8 +547,7 @@ begin locked_by = null, locked_at = null, failed_at = v_now, - last_error = 'superseded by reinvoke', - slot_group_number = null + last_error = 'superseded by reinvoke' from superseded s where e.id = s.id; @@ -595,7 +564,8 @@ begin dedupe_key, singleton_on, cron_expression, - priority + priority, + "group" ) select pgconductor._private_portable_uuidv7(), @@ -614,14 +584,16 @@ begin else null end, spec.cron_expression, - coalesce(spec.priority, 0) + coalesce(spec.priority, 0), + spec."group" from unnest(specs) as spec on conflict (task_key, dedupe_key, queue) do update set payload = excluded.payload, run_at = excluded.run_at, priority = excluded.priority, cron_expression = excluded.cron_expression, - singleton_on = excluded.singleton_on + singleton_on = excluded.singleton_on, + "group" = excluded."group" returning pgconductor._private_executions.id; end; $function$ @@ -636,7 +608,8 @@ create or replace function pgconductor.invoke( p_dedupe_seconds integer default null, p_dedupe_next_slot boolean default false, p_cron_expression text default null, - p_priority integer default null + p_priority integer default null, + p_group text default null ) returns table(id uuid) language plpgsql @@ -656,22 +629,13 @@ begin -- clear locked dedupe key before insert (supersede pattern) if p_dedupe_key is not null then with superseded as ( - select e.id, e.queue, e.task_key, e.slot_group_number + select e.id, e.queue, e.task_key from pgconductor._private_executions e where e.dedupe_key = p_dedupe_key and e.task_key = p_task_key and e.queue = p_queue and e.locked_at is not null for update of e - ), - released_slots as ( - update pgconductor._private_concurrency_slots cs - set used = 0 - from superseded s - where cs.queue = s.queue - and cs.task_key = s.task_key - and cs.slot_group_number = s.slot_group_number - and s.slot_group_number is not null ) update pgconductor._private_executions e set @@ -679,9 +643,8 @@ begin locked_by = null, locked_at = null, failed_at = v_now, - last_error = 'superseded by reinvoke', - slot_group_number = null - from superseded s + last_error = 'superseded by reinvoke' + from superseded s where e.id = s.id; end if; @@ -706,7 +669,8 @@ begin dedupe_key, singleton_on, cron_expression, - priority + priority, + "group" ) values ( pgconductor._private_portable_uuidv7(), p_task_key, @@ -716,7 +680,8 @@ begin p_dedupe_key, v_singleton_on, p_cron_expression, - coalesce(p_priority, 0) + coalesce(p_priority, 0), + p_group ) on conflict (task_key, singleton_on, coalesce(dedupe_key, ''), queue) where singleton_on is not null and completed_at is null and failed_at is null and cancelled = false @@ -737,7 +702,8 @@ begin dedupe_key, singleton_on, cron_expression, - priority + priority, + "group" ) values ( pgconductor._private_portable_uuidv7(), p_task_key, @@ -747,14 +713,17 @@ begin p_dedupe_key, v_next_singleton_on, p_cron_expression, - coalesce(p_priority, 0) + coalesce(p_priority, 0), + p_group ) on conflict (task_key, singleton_on, coalesce(dedupe_key, ''), queue) where singleton_on is not null and completed_at is null and failed_at is null and cancelled = false do update set payload = excluded.payload, run_at = excluded.run_at, - priority = excluded.priority + priority = excluded.priority, + cron_expression = excluded.cron_expression, + "group" = excluded."group" returning _private_executions.id; return; end if; @@ -769,7 +738,8 @@ begin run_at, dedupe_key, cron_expression, - priority + priority, + "group" ) values ( pgconductor._private_portable_uuidv7(), p_task_key, @@ -778,13 +748,15 @@ begin v_run_at, p_dedupe_key, p_cron_expression, - coalesce(p_priority, 0) + coalesce(p_priority, 0), + p_group ) on conflict (task_key, dedupe_key, queue) do update set payload = excluded.payload, run_at = excluded.run_at, priority = excluded.priority, - cron_expression = excluded.cron_expression + cron_expression = excluded.cron_expression, + "group" = excluded."group" returning e.id; end; $function$ diff --git a/packages/pgconductor-js/src/lib/assert.ts b/packages/pgconductor-js/src/lib/assert.ts index ad0a09d..c2a1bb3 100644 --- a/packages/pgconductor-js/src/lib/assert.ts +++ b/packages/pgconductor-js/src/lib/assert.ts @@ -13,3 +13,10 @@ export function equal(actual: T, expected: T, message?: string): void { export function never(x: never): never { throw new Error(`Unhandled case: ${x}`); } + +export function positiveInteger(value: number | undefined, name: string): number | undefined { + if (value !== undefined) { + ok(Number.isInteger(value) && value > 0, `${name} must be a positive integer`); + } + return value; +} diff --git a/packages/pgconductor-js/src/query-builder.ts b/packages/pgconductor-js/src/query-builder.ts index 587e4f2..6ec295a 100644 --- a/packages/pgconductor-js/src/query-builder.ts +++ b/packages/pgconductor-js/src/query-builder.ts @@ -37,7 +37,6 @@ export type GetExecutionsArgs = { queueName: string; batchSize: number; filterTaskKeys: string[]; - taskKeysWithConcurrency: string[]; }; export type RemoveExecutionsArgs = { @@ -171,36 +170,25 @@ export class QueryBuilder { where o.last_heartbeat_at < pgconductor._private_current_time() - ${maxAge}::interval returning o.id ), - released_slots as ( - update pgconductor._private_concurrency_slots cs - set used = 0 - from pgconductor._private_executions e, expired - where e.locked_by = expired.id - and e.slot_group_number is not null - and cs.queue = e.queue - and cs.task_key = e.task_key - and cs.slot_group_number = e.slot_group_number - ), -- fail cancelled executions from expired orchestrators failed_cancelled as ( update pgconductor._private_executions e set failed_at = pgconductor._private_current_time(), locked_by = null, - locked_at = null, - slot_group_number = null + locked_at = null from expired where e.locked_by = expired.id and e.cancelled = true and e.failed_at is null and e.completed_at is null + returning e.id ) -- unlock remaining (non-cancelled) executions update pgconductor._private_executions e set locked_by = null, - locked_at = null, - slot_group_number = null + locked_at = null from expired where e.locked_by = expired.id and e.cancelled = false @@ -238,36 +226,25 @@ export class QueryBuilder { where id = ${orchestratorId}::uuid returning id ), - released_slots as ( - update pgconductor._private_concurrency_slots cs - set used = 0 - from pgconductor._private_executions e, deleted - where e.locked_by = deleted.id - and e.slot_group_number is not null - and cs.queue = e.queue - and cs.task_key = e.task_key - and cs.slot_group_number = e.slot_group_number - ), -- fail cancelled executions from this orchestrator failed_cancelled as ( update pgconductor._private_executions e set failed_at = pgconductor._private_current_time(), locked_by = null, - locked_at = null, - slot_group_number = null + locked_at = null from deleted where e.locked_by = deleted.id and e.cancelled = true and e.failed_at is null and e.completed_at is null + returning e.id ) -- unlock remaining (non-cancelled) executions update pgconductor._private_executions e set locked_by = null, - locked_at = null, - slot_group_number = null + locked_at = null from deleted where e.locked_by = deleted.id and e.cancelled = false @@ -279,268 +256,97 @@ export class QueryBuilder { queueName, batchSize, filterTaskKeys, - taskKeysWithConcurrency, }: GetExecutionsArgs): PendingQuery { - // fast path: no tasks have concurrency limits - if (!taskKeysWithConcurrency.length) { - return this.sql` - with e as ( - select - e.id, - e.task_key - from pgconductor._private_executions e - where e.queue = ${queueName}::text - ${filterTaskKeys?.length ? this.sql`and not (e.task_key = any(${this.sql.array(filterTaskKeys)}::text[]))` : this.sql``} - and e.run_at <= pgconductor._private_current_time() - and e.is_available = true - order by e.priority asc, e.run_at asc, e.created_at asc, e.id asc - limit ${batchSize}::integer - for update skip locked - ), - - claimed as ( - update pgconductor._private_executions - set - attempts = _private_executions.attempts + 1, - locked_by = ${orchestratorId}::uuid, - locked_at = pgconductor._private_current_time() - from e - where _private_executions.id = e.id - and _private_executions.queue = ${queueName}::text - returning - _private_executions.id, - _private_executions.task_key, - _private_executions.queue, - _private_executions.payload, - _private_executions.waiting_on_execution_id, - _private_executions.waiting_step_key, - _private_executions.cancelled, - _private_executions.last_error, - _private_executions.dedupe_key, - _private_executions.cron_expression, - _private_executions.locked_by, - _private_executions.slot_group_number, - _private_executions.priority, - _private_executions.run_at, - _private_executions.created_at - ) - select - c.id, - c.task_key, - c.queue, - c.payload, - c.waiting_on_execution_id, - c.waiting_step_key, - c.cancelled, - c.last_error, - c.dedupe_key, - c.cron_expression, - c.locked_by, - c.slot_group_number - from claimed c - order by c.priority asc, c.run_at asc, c.created_at asc, c.id asc - `; - } - - // slow path: some tasks have concurrency limits (OPTIMIZED) return this.sql` - with - -- lock up to batchSize slots per concurrency task - locked_slots_raw as ( - select t.task_key, ls.slot_group_number - from unnest(${this.sql.array(taskKeysWithConcurrency)}::text[]) as t(task_key) - cross join lateral ( - select s.slot_group_number - from pgconductor._private_concurrency_slots s - where s.task_key = t.task_key - and s.queue = ${queueName}::text - and s.used = 0 - order by s.slot_group_number - limit ${batchSize}::integer - for update skip locked - ) as ls - ), - - -- count slots per task - slots_per_task as ( - select task_key, count(*) as slot_count - from locked_slots_raw - group by task_key - ), - - -- lock jobs for concurrency tasks (limit by slot count) - concurrency_execs as ( - select t.task_key, le.* - from slots_per_task t - cross join lateral ( - select - e.id, - e.task_key as exec_task_key, - e.queue, - e.payload, - e.waiting_on_execution_id, - e.waiting_step_key, - e.cancelled, - e.last_error, - e.dedupe_key, - e.cron_expression, - e.priority, - e.run_at, - e.created_at - from pgconductor._private_executions e - where e.is_available = true - and e.run_at <= pgconductor._private_current_time() - and e.queue = ${queueName}::text - and e.task_key = t.task_key - ${filterTaskKeys.length ? this.sql`and not (e.task_key = any(${this.sql.array(filterTaskKeys)}::text[]))` : this.sql``} + with active_tasks as ( + select e.task_key, count(*)::integer as active_count + from pgconductor._private_executions e + where e.queue = ${queueName}::text + and e.locked_at is not null + and e.failed_at is null + and e.completed_at is null + group by e.task_key + ), active_groups as ( + select e.task_key, e."group", count(*)::integer as active_count + from pgconductor._private_executions e + where e.queue = ${queueName}::text + and e."group" is not null + and e.locked_at is not null + and e.failed_at is null + and e.completed_at is null + group by e.task_key, e."group" + ), ranked as ( + select + e.id, + e.task_key, + e.queue, + e.priority, + e.run_at, + e.created_at, + e."group", + t.concurrency_limit, + t.group_concurrency_limit, + coalesce(at.active_count, 0) as active_task_count, + coalesce(ag.active_count, 0) as active_group_count, + row_number() over ( + partition by e.task_key order by e.priority asc, e.run_at asc, e.created_at asc, e.id asc - limit t.slot_count - for update skip locked - ) as le - ), - - -- lock jobs from non-concurrency tasks - unlimited_execs as ( - select - e.id, - e.task_key, - e.queue, - e.payload, - e.waiting_on_execution_id, - e.waiting_step_key, - e.cancelled, - e.last_error, - e.dedupe_key, - e.cron_expression, - e.priority, - e.run_at, - e.created_at - from pgconductor._private_executions e - where e.is_available = true - and e.run_at <= pgconductor._private_current_time() - and e.queue = ${queueName}::text - and not (e.task_key = any(${this.sql.array(taskKeysWithConcurrency)}::text[])) - ${filterTaskKeys.length > 0 ? this.sql`and not (e.task_key = any(${this.sql.array(filterTaskKeys)}::text[]))` : this.sql``} - order by e.priority asc, e.run_at asc, e.created_at asc, e.id asc - limit ${batchSize}::integer - for update skip locked - ), - - -- row number concurrency executions by task - concurrency_execs_rn as ( - select - ce.*, - row_number() over (partition by ce.task_key order by ce.priority asc, ce.run_at asc, ce.created_at asc, ce.id asc) as exec_rn - from concurrency_execs ce - ), - - -- row number slots by task - slots_rn as ( - select - ls.*, - row_number() over (partition by ls.task_key order by ls.slot_group_number) as slot_rn - from locked_slots_raw ls - ), - - -- pair concurrency executions with slots - concurrency_paired as ( - select - e.id, - e.exec_task_key as task_key, - e.queue, - e.payload, - e.waiting_on_execution_id, - e.waiting_step_key, - e.cancelled, - e.last_error, - e.dedupe_key, - e.cron_expression, - s.slot_group_number - from concurrency_execs_rn e - join slots_rn s - on e.task_key = s.task_key - and e.exec_rn = s.slot_rn - ), - - -- unlimited executions don't need slots - unlimited_paired as ( - select - id, - task_key, - queue, - payload, - waiting_on_execution_id, - waiting_step_key, - cancelled, - last_error, - dedupe_key, - cron_expression, - null::integer as slot_group_number - from unlimited_execs - ), - - -- merge all paired executions - paired as ( - select * from concurrency_paired - union all - select * from unlimited_paired - ), - - -- mark slots as used - mark_used as ( - update pgconductor._private_concurrency_slots cs - set used = 1 - from paired p - where cs.queue = ${queueName}::text - and cs.task_key = p.task_key - and cs.slot_group_number = p.slot_group_number - and p.slot_group_number is not null - ), - - -- update and return executions. The outer select owns result ordering; - -- update returning order is not defined. - claimed as ( + ) as task_rank, + row_number() over ( + partition by e.task_key, e."group" + order by e.priority asc, e.run_at asc, e.created_at asc, e.id asc + ) as group_rank + from pgconductor._private_executions e + left join pgconductor._private_tasks t + on t.key = e.task_key and t.queue = e.queue + left join active_tasks at on at.task_key = e.task_key + left join active_groups ag on ag.task_key = e.task_key and ag."group" = e."group" + where e.queue = ${queueName}::text + and e.run_at <= pgconductor._private_current_time() + and e.is_available = true + ${filterTaskKeys?.length ? this.sql`and not (e.task_key = any(${this.sql.array(filterTaskKeys)}::text[]))` : this.sql``} + ), group_eligible as ( + select r.*, + row_number() over ( + partition by r.task_key + order by r.priority asc, r.run_at asc, r.created_at asc, r.id asc + ) as available_task_rank + from ranked r + where r.group_concurrency_limit is null + or r."group" is null + or r.active_group_count + r.group_rank <= r.group_concurrency_limit + ), eligible as ( + select r.id, r.priority, r.run_at, r.created_at + from group_eligible r + where r.concurrency_limit is null + or r.active_task_count + r.available_task_rank <= r.concurrency_limit + order by r.priority asc, r.run_at asc, r.created_at asc, r.id asc + -- Keep a bounded candidate pool so SKIP LOCKED can backfill a batch. + limit greatest(${batchSize}::integer * 4, ${batchSize}::integer) + ), locked_candidates as ( + select e.id + from pgconductor._private_executions e + join eligible c on c.id = e.id + where e.queue = ${queueName}::text + order by c.priority asc, c.run_at asc, c.created_at asc, c.id asc + limit ${batchSize}::integer + for update of e skip locked + ), claimed as ( update pgconductor._private_executions e set attempts = e.attempts + 1, locked_by = ${orchestratorId}::uuid, - slot_group_number = p.slot_group_number, locked_at = pgconductor._private_current_time() - from paired p - where e.id = p.id - and e.queue = p.queue - returning - e.id, - e.task_key, - e.queue, - e.payload, - e.waiting_on_execution_id, - e.waiting_step_key, - e.cancelled, - e.last_error, - e.dedupe_key, - e.cron_expression, - e.locked_by, - e.slot_group_number, - e.priority, - e.run_at, - e.created_at + from locked_candidates c + where e.id = c.id and e.queue = ${queueName}::text and e.is_available = true + returning e.id, e.task_key, e.queue, e.payload, e.waiting_on_execution_id, + e.waiting_step_key, e.cancelled, e.last_error, e.dedupe_key, e.cron_expression, + e.locked_by, e."group", e.priority, e.run_at, e.created_at ) - select - c.id, - c.task_key, - c.queue, - c.payload, - c.waiting_on_execution_id, - c.waiting_step_key, - c.cancelled, - c.last_error, - c.dedupe_key, - c.cron_expression, - c.locked_by, - c.slot_group_number - from claimed c - order by c.priority asc, c.run_at asc, c.created_at asc, c.id asc + select id, task_key, queue, payload, waiting_on_execution_id, waiting_step_key, + cancelled, last_error, dedupe_key, cron_expression, locked_by, "group" + from claimed + order by priority asc, run_at asc, created_at asc, id asc `; } @@ -563,13 +369,13 @@ export class QueryBuilder { orchestrator_id uuid, result jsonb, error text, reschedule_in_ms text, step_key text, timeout_ms text, child_task_name text, child_task_queue text, child_payload jsonb, - slot_group_number integer + "group" text ) )`); // Lock the claimed rows for the whole statement. This prevents recovery or a // new claim from racing the side effects below. ctes.push(this.sql`valid_results as materialized ( - select r.*, e.slot_group_number as owned_slot_group_number, + select r.*, e.cancelled as execution_cancelled, e.last_error as execution_last_error from result_data r join pgconductor._private_executions e @@ -599,17 +405,6 @@ export class QueryBuilder { select * from valid_results where status = 'invoke_child' )`); - ctes.push(this.sql`released_slots as ( - update pgconductor._private_concurrency_slots cs - set used = 0 - from valid_results r - where r.owned_slot_group_number is not null - and cs.queue = r.queue - and cs.task_key = r.task_key - and cs.slot_group_number = r.owned_slot_group_number - and cs.used > 0 - )`); - // 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. ctes.push(this.sql`completed_parents as materialized ( @@ -634,10 +429,9 @@ export class QueryBuilder { update pgconductor._private_executions e set failed_at = nt.ts, completed_at = null, last_error = 'Parent timed out before child completed', - locked_by = null, locked_at = null, slot_group_number = null + locked_by = null, locked_at = null from now_ts nt, completed_results r where e.id = r.execution_id and e.queue = r.queue - and e.task_key = r.task_key and e.locked_by = r.orchestrator_id and e.parent_execution_id is not null and not exists ( @@ -649,7 +443,7 @@ export class QueryBuilder { ctes.push(this.sql`updated_parents_all as ( update pgconductor._private_executions e set run_at = nt.ts, waiting_on_execution_id = null, waiting_step_key = null, - locked_by = null, locked_at = null, slot_group_number = null + locked_by = null, locked_at = null from now_ts nt, completed_parents p where e.id = p.parent_id and e.queue = p.queue returning e.id @@ -658,7 +452,6 @@ export class QueryBuilder { delete from pgconductor._private_executions e using completed_results r, task_configs tc where e.id = r.execution_id and e.queue = r.queue - and e.task_key = r.task_key and e.locked_by = r.orchestrator_id and tc.key = r.task_key and tc.queue = r.queue and tc.remove_on_complete_days = 0 and not exists (select 1 from orphaned_children oc where oc.id = e.id) @@ -666,10 +459,9 @@ export class QueryBuilder { )`); ctes.push(this.sql`updated_completed as ( update pgconductor._private_executions e - set completed_at = nt.ts, locked_by = null, locked_at = null, slot_group_number = null + set completed_at = nt.ts, locked_by = null, locked_at = null from now_ts nt, completed_results r, task_configs tc where e.id = r.execution_id and e.queue = r.queue - and e.task_key = r.task_key and e.locked_by = r.orchestrator_id and tc.key = r.task_key and tc.queue = r.queue and (tc.remove_on_complete_days is null or tc.remove_on_complete_days != 0) @@ -730,7 +522,7 @@ export class QueryBuilder { last_error = case when f.is_child then coalesce(f.child_error, 'unknown error') else 'Child execution failed: ' || coalesce(f.child_error, 'unknown error') end, waiting_on_execution_id = null, waiting_step_key = null, - locked_by = null, locked_at = null, slot_group_number = null + locked_by = null, locked_at = null from now_ts nt, failed_updates f where e.id = f.target_id and e.queue = f.queue returning e.id @@ -740,10 +532,9 @@ export class QueryBuilder { set last_error = coalesce(r.error, 'unknown error'), run_at = greatest(nt.ts, coalesce(e.run_at, nt.ts)) + ((array[15, 30, 60, 120, 300, 600, 1200, 2400, 3600, 7200])[least(greatest(e.attempts, 1), 10)] * interval '1 second'), - locked_by = null, locked_at = null, slot_group_number = null + locked_by = null, locked_at = null from now_ts nt, failed_results r, task_configs tc where e.id = r.execution_id and e.queue = r.queue - and e.task_key = r.task_key and e.locked_by = r.orchestrator_id and tc.key = r.task_key and tc.queue = r.queue and r.status <> 'permanently_failed' @@ -766,22 +557,20 @@ export class QueryBuilder { when nullif(trim(r.reschedule_in_ms), '') is not null then nt.ts + (nullif(trim(r.reschedule_in_ms), '')::bigint || ' milliseconds')::interval else nt.ts end, - locked_by = null, locked_at = null, slot_group_number = null + locked_by = null, locked_at = null from now_ts nt, released_results r where e.id = r.execution_id and e.queue = r.queue - and e.task_key = r.task_key and e.locked_by = r.orchestrator_id returning e.id )`); ctes.push(this.sql`inserted_children as ( - insert into pgconductor._private_executions (id, task_key, queue, payload, run_at, parent_execution_id) - select pgconductor._private_portable_uuidv7(), r.child_task_name, r.child_task_queue, r.child_payload, nt.ts, r.execution_id + insert into pgconductor._private_executions (id, task_key, queue, payload, run_at, parent_execution_id, "group") + select pgconductor._private_portable_uuidv7(), r.child_task_name, r.child_task_queue, r.child_payload, nt.ts, r.execution_id, r."group" from invoke_child_data r, now_ts nt where exists ( select 1 from pgconductor._private_executions parent where parent.id = r.execution_id and parent.queue = r.queue - and parent.task_key = r.task_key and parent.locked_by = r.orchestrator_id ) returning id, parent_execution_id @@ -793,11 +582,10 @@ export class QueryBuilder { when nullif(trim(r.timeout_ms), '') is not null then nt.ts + (nullif(trim(r.timeout_ms), '')::bigint || ' milliseconds')::interval else nt.ts end, - locked_by = null, locked_at = null, slot_group_number = null + locked_by = null, locked_at = null from now_ts nt, inserted_children ic join invoke_child_data r on r.execution_id = ic.parent_execution_id where e.id = r.execution_id and e.queue = r.queue - and e.task_key = r.task_key and e.locked_by = r.orchestrator_id returning e.id )`); @@ -847,6 +635,7 @@ export class QueryBuilder { window_start: spec.window?.[0] || null, window_end: spec.window?.[1] || null, concurrency_limit: spec.concurrency || null, + group_concurrency_limit: spec.groupConcurrency || null, })); const cronScheduleRows = cronSchedules.map((spec) => { @@ -860,6 +649,7 @@ export class QueryBuilder { dedupe_key: spec.dedupe_key, cron_expression: spec.cron_expression, priority: spec.priority || null, + group: spec.group || null, }; }); @@ -917,7 +707,8 @@ export class QueryBuilder { p_dedupe_seconds := ${dedupe_seconds}::integer, p_dedupe_next_slot := ${dedupe_next_slot}::boolean, p_cron_expression := ${spec.cron_expression || null}::text, - p_priority := ${spec.priority || null}::integer + p_priority := ${spec.priority || null}::integer, + p_group := ${spec.group || null}::text ) `; } @@ -953,7 +744,8 @@ export class QueryBuilder { p_dedupe_seconds := null::integer, p_dedupe_next_slot := false::boolean, p_cron_expression := ${cronExpression}::text, - p_priority := ${spec.priority || 0}::integer + p_priority := ${spec.priority || 0}::integer, + p_group := ${spec.group || null}::text ) `; } @@ -1023,6 +815,7 @@ export class QueryBuilder { dedupe_next_slot, cron_expression: spec.cron_expression || null, priority: spec.priority, + group: spec.group || null, }; }); diff --git a/packages/pgconductor-js/src/task-context.ts b/packages/pgconductor-js/src/task-context.ts index a74fd15..539410e 100644 --- a/packages/pgconductor-js/src/task-context.ts +++ b/packages/pgconductor-js/src/task-context.ts @@ -38,6 +38,7 @@ export type TaskAbortReasons = step_key: string; task: TaskIdentifier; payload: Payload | null; + group?: string | null; __pgconductorTaskAborted: true; }; @@ -86,6 +87,7 @@ export type TaskContextOptions = { type ScheduleOptions = { cron: string; priority?: number; + group?: string; }; // second argument for tasks @@ -237,6 +239,7 @@ export class TaskContext< task: TaskIdentifier, payload: InferPayload = {} as InferPayload, timeout?: number, + group?: string, ): Promise> { const cached = await this.opts.db.loadStep( { @@ -277,6 +280,7 @@ export class TaskContext< task, step_key: key, payload, + group, }); } @@ -318,6 +322,7 @@ export class TaskContext< run_at: nextTimestamp, cron_expression: options.cron, priority: options.priority || null, + group: options.group || null, }, scheduleName, }, diff --git a/packages/pgconductor-js/src/task-definition.ts b/packages/pgconductor-js/src/task-definition.ts index f847e1b..cf7257b 100644 --- a/packages/pgconductor-js/src/task-definition.ts +++ b/packages/pgconductor-js/src/task-definition.ts @@ -118,7 +118,7 @@ export type FindTaskByIdentifier< // Trigger types export type InvocableTrigger = { invocable: true }; -export type CronTrigger = { cron: string; name: string }; +export type CronTrigger = { cron: string; name: string; group?: string }; // Event trigger - triggers when a custom event is emitted export type CustomEventTrigger< diff --git a/packages/pgconductor-js/src/task.ts b/packages/pgconductor-js/src/task.ts index 22973e5..ae5c479 100644 --- a/packages/pgconductor-js/src/task.ts +++ b/packages/pgconductor-js/src/task.ts @@ -18,6 +18,7 @@ import type { RowType, } from "./event-definition"; import type { SelectedRow } from "./select-columns"; +import * as assert from "./lib/assert"; export type TaskIdentifier = { readonly name: TName; @@ -38,6 +39,7 @@ export type TaskConfiguration< removeOnComplete?: RetentionSettings; removeOnFail?: RetentionSettings; concurrency?: number; + groupConcurrency?: number; batch?: BatchConfig; }; @@ -192,6 +194,7 @@ export class Task< public readonly removeOnComplete: RetentionSettings; public readonly removeOnFail: RetentionSettings; public readonly concurrency?: number; + public readonly groupConcurrency?: number; public readonly batch?: BatchConfig; public readonly triggers: NonEmptyArray; @@ -209,7 +212,8 @@ export class Task< this.window = config.window; this.removeOnComplete = config.removeOnComplete ?? false; this.removeOnFail = config.removeOnFail ?? false; - this.concurrency = config.concurrency; + this.concurrency = assert.positiveInteger(config.concurrency, "concurrency"); + this.groupConcurrency = assert.positiveInteger(config.groupConcurrency, "groupConcurrency"); this.batch = config.batch; this.triggers = Array.isArray(triggers) ? triggers : [triggers]; diff --git a/packages/pgconductor-js/src/worker.ts b/packages/pgconductor-js/src/worker.ts index f5214c8..49a410b 100644 --- a/packages/pgconductor-js/src/worker.ts +++ b/packages/pgconductor-js/src/worker.ts @@ -320,13 +320,14 @@ export class Worker< removeOnFailDays: retentionToDays(task.removeOnFail), window: task.window, concurrency: task.concurrency, + groupConcurrency: task.groupConcurrency, })); const allTasks = Array.from(this.tasks.values()); const cronSchedules: ExecutionSpec[] = allTasks.flatMap((task) => task.triggers - .filter((t): t is { cron: string; name: string } => "cron" in t) + .filter((t): t is { cron: string; name: string; group?: string } => "cron" in t) .map((trigger) => { const nextTimestamp = nextCronOccurrence(trigger.cron, this.clock.now()); const timestampSeconds = Math.floor(nextTimestamp.getTime() / 1000); @@ -336,6 +337,7 @@ export class Worker< run_at: nextTimestamp, dedupe_key: `scheduled::${trigger.name}::${timestampSeconds}`, cron_expression: trigger.cron, + group: trigger.group || null, }; }), ); @@ -402,10 +404,6 @@ export class Worker< for (const task of allTasks) { taskMaxAttempts[task.name] = task.maxAttempts || 3; } - const taskKeysWithConcurrency = allTasks - .filter((task) => task.concurrency != null) - .map((task) => task.name); - // Check if any tasks have windows - only then do we need time-based filtering const tasksWithWindows = allTasks.filter((task) => task.window); @@ -446,7 +444,6 @@ export class Worker< queueName: this.queueName, batchSize: this.fetchBatchSize, filterTaskKeys: disallowedTaskKeys, - taskKeysWithConcurrency, }, { signal: this.signal }, ); @@ -490,7 +487,6 @@ export class Worker< task_key: taskKey, status: "failed", error: `Task not found: ${taskKey}`, - slot_group_number: exec.slot_group_number, })) as ExecutionResult[]; } @@ -505,7 +501,6 @@ export class Worker< task_key: taskKey, status: "permanently_failed", error: exec.last_error || "Execution was cancelled", - slot_group_number: exec.slot_group_number, })) as ExecutionResult[]; } @@ -628,7 +623,7 @@ export class Worker< child_task_name: output.task.name, child_task_queue: output.task.queue || "default", child_payload: output.payload, - slot_group_number: exec.slot_group_number, + group: output.group, } as const; case "cancelled": return { @@ -638,7 +633,6 @@ export class Worker< task_key: exec.task_key, status: "permanently_failed", error: exec.last_error || "Task was cancelled", - slot_group_number: exec.slot_group_number, } as const; case "released": case "parent-aborted": @@ -650,7 +644,6 @@ export class Worker< step_key: output.reason === "released" ? output.step_key : undefined, task_key: exec.task_key, status: "released", - slot_group_number: exec.slot_group_number, } as const; default: assert.never(output); @@ -664,7 +657,6 @@ export class Worker< task_key: exec.task_key, status: "completed", result: output, - slot_group_number: exec.slot_group_number, } as const; } catch (err) { return { @@ -674,7 +666,6 @@ export class Worker< task_key: exec.task_key, status: "failed", error: coerceError(err).message, - slot_group_number: exec.slot_group_number, } as const; } finally { // Clean up running task tracking @@ -750,7 +741,6 @@ export class Worker< status: "released" as const, reschedule_in_ms: result.reschedule_in_ms, step_key: result.step_key, - slot_group_number: exec.slot_group_number, })); } @@ -762,7 +752,6 @@ export class Worker< task_key: taskKey, status: "failed" as const, error: `Task aborted: ${result.reason}`, - slot_group_number: exec.slot_group_number, })); } @@ -775,7 +764,6 @@ export class Worker< task_key: taskKey, status: "completed" as const, result: undefined, - slot_group_number: exec.slot_group_number, })); } @@ -798,7 +786,6 @@ export class Worker< task_key: taskKey, status: "completed" as const, result: result[i], - slot_group_number: exec.slot_group_number, })); } catch (err) { // Handler threw: all fail together @@ -810,7 +797,6 @@ export class Worker< task_key: taskKey, status: "failed" as const, error: errorMsg, - slot_group_number: exec.slot_group_number, })); } } @@ -842,6 +828,7 @@ export class Worker< run_at: nextTimestamp, dedupe_key: nextDedupeKey, cron_expression: execution.cron_expression, + group: execution.group || null, }, { signal: this.signal }, ); diff --git a/packages/pgconductor-js/tests/integration/concurrency.test.ts b/packages/pgconductor-js/tests/integration/concurrency.test.ts index 9446a83..1833b28 100644 --- a/packages/pgconductor-js/tests/integration/concurrency.test.ts +++ b/packages/pgconductor-js/tests/integration/concurrency.test.ts @@ -216,68 +216,6 @@ describe("Task-Level Concurrency", () => { await orchestrator.stopped; }, 30000); - test("high concurrency uses slot groups correctly", async () => { - const db = await pool.child(); - databases.push(db); - - const taskDef = defineTask({ name: "high-concurrency-task" }); - - const conductor = Conductor.create({ - sql: db.sql, - tasks: TaskSchemas.fromSchema([taskDef]), - context: {}, - }); - - let completed = 0; - - const highConcurrencyTask = conductor.createTask( - { name: "high-concurrency-task", concurrency: 500 }, - { invocable: true }, - async () => { - completed++; - }, - ); - - const orchestrator = Orchestrator.create({ - conductor, - tasks: [highConcurrencyTask], - defaultWorker: { pollIntervalMs: 50, flushIntervalMs: 50, concurrency: 50 }, - }); - - await orchestrator.start(); - - // queue 100 tasks - for (let i = 0; i < 100; i++) { - await conductor.invoke({ name: "high-concurrency-task" }, {}); - } - - // wait for completion - await new Promise((r) => setTimeout(r, 5000)); - - await orchestrator.stop(); - await orchestrator.stopped; - - // extra wait to ensure all flushes complete - await new Promise((r) => setTimeout(r, 1000)); - - // all tasks should complete - expect(completed).toBe(100); - - // verify slots were created (one row per slot, concurrency=500 → 500 rows) - const slots = await db.sql` - select * from pgconductor._private_concurrency_slots - where task_key = 'high-concurrency-task' - `; - - expect(slots.length).toBe(500); // one row per slot - expect(slots.every((s) => s.capacity === 1)).toBe(true); // each slot has capacity=1 - - // verify all slots are released (used = 0) - for (const slot of slots) { - expect(slot.used).toBe(0); - } - }, 30000); - test("mixed queue with concurrency and without", async () => { const db = await pool.child(); databases.push(db); diff --git a/packages/pgconductor-js/tests/integration/execution-foundations.test.ts b/packages/pgconductor-js/tests/integration/execution-foundations.test.ts index 1ec8f66..8668e9f 100644 --- a/packages/pgconductor-js/tests/integration/execution-foundations.test.ts +++ b/packages/pgconductor-js/tests/integration/execution-foundations.test.ts @@ -95,7 +95,6 @@ describe("execution foundations", () => { queueName: "queue-a", batchSize: 1, filterTaskKeys: [], - taskKeysWithConcurrency: [], }) )[0]; const second = ( @@ -104,7 +103,6 @@ describe("execution foundations", () => { queueName: "queue-b", batchSize: 1, filterTaskKeys: [], - taskKeysWithConcurrency: [], }) )[0]; expect(first?.queue).toBe("queue-a"); @@ -141,57 +139,6 @@ describe("execution foundations", () => { expect(remaining[0]?.completed_at).not.toBeNull(); }); - test("releases concurrency slots when recovering a stale orchestrator", async () => { - const db = await database(); - await db.client.registerWorker({ - queueName: "recovery-slots", - taskSpecs: [{ key: "slot-task", queue: "recovery-slots", concurrency: 1 }], - cronSchedules: [], - eventSubscriptions: [], - }); - await db.client.invoke({ task_key: "slot-task", queue: "recovery-slots" }); - - const orchestratorId = crypto.randomUUID(); - await db.client.orchestratorHeartbeat({ orchestratorId, version: "test", migrationNumber: 1 }); - const claimed = ( - await db.client.getExecutions({ - orchestratorId, - queueName: "recovery-slots", - batchSize: 1, - filterTaskKeys: [], - taskKeysWithConcurrency: ["slot-task"], - }) - )[0]; - if (!claimed || claimed.slot_group_number == null) { - throw new Error("expected execution to be claimed with a slot"); - } - const slotGroupNumber = claimed.slot_group_number; - - const before = await db.sql<{ used: number }[]>` - select used from pgconductor._private_concurrency_slots - where queue = 'recovery-slots' and task_key = 'slot-task' and slot_group_number = ${slotGroupNumber} - `; - expect(before[0]?.used).toBe(1); - - await db.sql` - update pgconductor._private_orchestrators - set last_heartbeat_at = now() - interval '1 hour' - where id = ${orchestratorId}::uuid - `; - await db.client.recoverStaleOrchestrators({ maxAge: "1 second" }); - - const after = await db.sql<{ used: number }[]>` - select used from pgconductor._private_concurrency_slots - where queue = 'recovery-slots' and task_key = 'slot-task' - `; - const recovered = await db.sql<{ locked_by: string | null }[]>` - select locked_by from pgconductor._private_executions - where queue = 'recovery-slots' and task_key = 'slot-task' - `; - expect(after[0]?.used).toBe(0); - expect(recovered[0]?.locked_by).toBeNull(); - }); - test("retains a parent when its permanently failed child is configured for removal", async () => { const db = await database(); await db.client.registerWorker({ @@ -211,7 +158,6 @@ describe("execution foundations", () => { queueName: "parent-retention", batchSize: 1, filterTaskKeys: [], - taskKeysWithConcurrency: [], }) )[0]; if (!parent) throw new Error("expected parent claim"); @@ -252,7 +198,6 @@ describe("execution foundations", () => { queueName: "parent-retention", batchSize: 1, filterTaskKeys: [], - taskKeysWithConcurrency: [], }) )[0]; if (!child) throw new Error("expected child claim"); @@ -313,7 +258,6 @@ describe("execution foundations", () => { queueName: "parent-queue", batchSize: 1, filterTaskKeys: [], - taskKeysWithConcurrency: [], }) )[0]; if (!parent) throw new Error("expected parent claim"); @@ -347,7 +291,6 @@ describe("execution foundations", () => { queueName: "child-queue", batchSize: 1, filterTaskKeys: [], - taskKeysWithConcurrency: [], }) )[0]; if (!child) throw new Error("expected child claim"); @@ -397,7 +340,6 @@ describe("execution foundations", () => { queueName: "cancel-buffered", batchSize: 1, filterTaskKeys: [], - taskKeysWithConcurrency: [], }) )[0]; if (!claimed) throw new Error("expected claim"); @@ -454,7 +396,6 @@ describe("execution foundations", () => { queueName: "cascade-parent", batchSize: 1, filterTaskKeys: [], - taskKeysWithConcurrency: [], }) )[0]; if (!parent) throw new Error("expected parent claim"); @@ -530,7 +471,6 @@ describe("execution foundations", () => { queueName: "enqueue-order", batchSize: 3, filterTaskKeys: [], - taskKeysWithConcurrency: [], }); expect(claimed.map((execution) => execution.id)).toEqual( expected.map((execution) => execution.id), @@ -560,7 +500,6 @@ describe("execution foundations", () => { queueName: "fenced", batchSize: 1, filterTaskKeys: [], - taskKeysWithConcurrency: [], }) )[0]; if (!oldClaim) throw new Error("expected old claim"); @@ -579,7 +518,6 @@ describe("execution foundations", () => { queueName: "fenced", batchSize: 1, filterTaskKeys: [], - taskKeysWithConcurrency: [], }) )[0]; if (!currentClaim) throw new Error("expected recovered execution to be re-claimed"); diff --git a/packages/pgconductor-js/tests/integration/group-concurrency.test.ts b/packages/pgconductor-js/tests/integration/group-concurrency.test.ts new file mode 100644 index 0000000..2842a5f --- /dev/null +++ b/packages/pgconductor-js/tests/integration/group-concurrency.test.ts @@ -0,0 +1,373 @@ +import { test, expect, describe, beforeAll, afterAll, afterEach } from "bun:test"; +import { z } from "zod"; +import { Conductor } from "../../src/conductor"; +import { Orchestrator } from "../../src/orchestrator"; +import { defineTask } from "../../src/task-definition"; +import { TaskSchemas } from "../../src/schemas"; +import { Deferred } from "../../src/lib/deferred"; +import { TestDatabasePool } from "../fixtures/test-database"; +import type { TestDatabase } from "../fixtures/test-database"; +import { waitForCondition } from "../test-utils"; + +const workerConfig = { + concurrency: 10, + fetchBatchSize: 10, + flushBatchSize: 10, + pollIntervalMs: 10, + flushIntervalMs: 10, +}; + +// Group limits are intentionally soft: concurrent claim transactions may race. +// These tests use blockers and avoid asserting exact global bounds across workers. + +describe("Group concurrency", () => { + let pool: TestDatabasePool; + const databases: TestDatabase[] = []; + + beforeAll(async () => { + pool = await TestDatabasePool.create(); + }, 60000); + + afterEach(async () => { + await Promise.all(databases.map((db) => db.destroy())); + databases.length = 0; + }); + + afterAll(async () => { + await pool?.destroy(); + }); + + test("serializes executions in the same group", async () => { + const db = await pool.child(); + databases.push(db); + const definition = defineTask({ + name: "same-group", + payload: z.object({ id: z.number() }), + }); + const conductor = Conductor.create({ + sql: db.sql, + tasks: TaskSchemas.fromSchema([definition]), + context: {}, + }); + const started: number[] = []; + const blockers = new Map>([ + [1, new Deferred()], + [2, new Deferred()], + ]); + const task = conductor.createTask( + { name: "same-group", groupConcurrency: 1 }, + { invocable: true }, + async (event) => { + started.push(event.payload.id); + const blocker = blockers.get(event.payload.id); + if (blocker) await blocker.promise; + }, + ); + const orchestrator = Orchestrator.create({ + conductor, + tasks: [task], + defaultWorker: workerConfig, + }); + + await orchestrator.start(); + try { + await conductor.invoke({ name: "same-group" }, { id: 1 }, { group: "tenant-a" }); + await conductor.invoke({ name: "same-group" }, { id: 2 }, { group: "tenant-a" }); + await waitForCondition(() => started.length === 1); + await new Promise((resolve) => setTimeout(resolve, 100)); + const first = started[0]; + expect(first === 1 || first === 2).toBe(true); + + blockers.get(first!)?.resolve(); + await waitForCondition(() => started.length === 2); + expect(new Set(started)).toEqual(new Set([1, 2])); + } finally { + blockers.forEach((blocker) => blocker.resolve()); + await orchestrator.stop(); + } + }, 30000); + + test("allows different groups to run in parallel", async () => { + const db = await pool.child(); + databases.push(db); + const definition = defineTask({ + name: "different-groups", + payload: z.object({ id: z.number() }), + }); + const conductor = Conductor.create({ + sql: db.sql, + tasks: TaskSchemas.fromSchema([definition]), + context: {}, + }); + const started: number[] = []; + const blocker = new Deferred(); + const task = conductor.createTask( + { name: "different-groups", groupConcurrency: 1 }, + { invocable: true }, + async (event) => { + started.push(event.payload.id); + await blocker.promise; + }, + ); + const orchestrator = Orchestrator.create({ + conductor, + tasks: [task], + defaultWorker: workerConfig, + }); + + await conductor.ensureInstalled(); + await conductor.invoke({ name: "different-groups" }, { id: 1 }, { group: "tenant-a" }); + await conductor.invoke({ name: "different-groups" }, { id: 2 }, { group: "tenant-b" }); + await orchestrator.start(); + try { + await waitForCondition(() => started.length === 2); + expect(new Set(started)).toEqual(new Set([1, 2])); + } finally { + blocker.resolve(); + await orchestrator.stop(); + } + }, 30000); + + test("does not apply the group limit to ungrouped executions", async () => { + const db = await pool.child(); + databases.push(db); + const definition = defineTask({ + name: "ungrouped", + payload: z.object({ id: z.number() }), + }); + const conductor = Conductor.create({ + sql: db.sql, + tasks: TaskSchemas.fromSchema([definition]), + context: {}, + }); + const started: number[] = []; + const blocker = new Deferred(); + const task = conductor.createTask( + { name: "ungrouped", groupConcurrency: 1 }, + { invocable: true }, + async (event) => { + started.push(event.payload.id); + await blocker.promise; + }, + ); + const orchestrator = Orchestrator.create({ + conductor, + tasks: [task], + defaultWorker: workerConfig, + }); + + await orchestrator.start(); + try { + await conductor.invoke({ name: "ungrouped" }, { id: 1 }, { group: "tenant-a" }); + await conductor.invoke({ name: "ungrouped" }, { id: 2 }); + await waitForCondition(() => started.length === 2); + expect(new Set(started)).toEqual(new Set([1, 2])); + } finally { + blocker.resolve(); + await orchestrator.stop(); + } + }, 30000); + + test("composes task and group concurrency limits", async () => { + const db = await pool.child(); + databases.push(db); + const definition = defineTask({ + name: "composed-limits", + payload: z.object({ id: z.number() }), + }); + const conductor = Conductor.create({ + sql: db.sql, + tasks: TaskSchemas.fromSchema([definition]), + context: {}, + }); + const started: number[] = []; + const blockers = new Map>([ + [1, new Deferred()], + [2, new Deferred()], + [3, new Deferred()], + ]); + const task = conductor.createTask( + { name: "composed-limits", concurrency: 2, groupConcurrency: 1 }, + { invocable: true }, + async (event) => { + started.push(event.payload.id); + const blocker = blockers.get(event.payload.id); + if (blocker) await blocker.promise; + }, + ); + const orchestrator = Orchestrator.create({ + conductor, + tasks: [task], + defaultWorker: workerConfig, + }); + + await orchestrator.start(); + try { + await conductor.invoke({ name: "composed-limits" }, { id: 1 }, { group: "tenant-a" }); + await conductor.invoke({ name: "composed-limits" }, { id: 3 }, { group: "tenant-b" }); + await conductor.invoke({ name: "composed-limits" }, { id: 2 }, { group: "tenant-a" }); + await waitForCondition( + () => started.includes(3) && started.some((id) => id === 1 || id === 2), + ); + await new Promise((resolve) => setTimeout(resolve, 100)); + expect(started).toContain(3); + expect(started.filter((id) => id === 1 || id === 2)).toHaveLength(1); + + blockers.forEach((blocker) => blocker.resolve()); + await waitForCondition(() => started.length === 3); + } finally { + blockers.forEach((blocker) => blocker.resolve()); + await orchestrator.stop(); + } + }, 30000); + + test("does not share a group across tasks or queues", async () => { + const db = await pool.child(); + databases.push(db); + const taskADefinition = defineTask({ name: "scope-a" }); + const taskBDefinition = defineTask({ name: "scope-b" }); + const queueTaskDefinition = defineTask({ name: "scope-queue", queue: "other" }); + const conductor = Conductor.create({ + sql: db.sql, + tasks: TaskSchemas.fromSchema([taskADefinition, taskBDefinition, queueTaskDefinition]), + context: {}, + }); + const started = new Set(); + const blocker = new Deferred(); + const taskA = conductor.createTask( + { name: "scope-a", groupConcurrency: 1 }, + { invocable: true }, + async () => { + started.add("task-a"); + await blocker.promise; + }, + ); + const taskB = conductor.createTask( + { name: "scope-b", groupConcurrency: 1 }, + { invocable: true }, + async () => { + started.add("task-b"); + await blocker.promise; + }, + ); + const queueTask = conductor.createTask( + { name: "scope-queue", queue: "other", groupConcurrency: 1 }, + { invocable: true }, + async () => { + started.add("queue"); + await blocker.promise; + }, + ); + const otherWorker = conductor.createWorker({ + queue: "other", + tasks: [queueTask], + config: workerConfig, + }); + const orchestrator = Orchestrator.create({ + conductor, + tasks: [taskA, taskB], + workers: [otherWorker], + defaultWorker: workerConfig, + }); + + await orchestrator.start(); + try { + await conductor.invoke({ name: "scope-a" }, {}, { group: "shared" }); + await conductor.invoke({ name: "scope-b" }, {}, { group: "shared" }); + await conductor.invoke({ name: "scope-queue", queue: "other" }, {}, { group: "shared" }); + await waitForCondition(() => started.size === 3); + expect(started).toEqual(new Set(["task-a", "task-b", "queue"])); + } finally { + blocker.resolve(); + await orchestrator.stop(); + } + }, 30000); + + test("propagates group metadata through batch invocation", async () => { + const db = await pool.child(); + databases.push(db); + const definition = defineTask({ + name: "batch-group", + payload: z.object({ id: z.number() }), + }); + const conductor = Conductor.create({ + sql: db.sql, + tasks: TaskSchemas.fromSchema([definition]), + context: {}, + }); + const started: number[] = []; + const blockers = new Map>([ + [1, new Deferred()], + [2, new Deferred()], + ]); + const task = conductor.createTask( + { name: "batch-group", groupConcurrency: 1 }, + { invocable: true }, + async (event) => { + started.push(event.payload.id); + const blocker = blockers.get(event.payload.id); + if (blocker) await blocker.promise; + }, + ); + const orchestrator = Orchestrator.create({ + conductor, + tasks: [task], + defaultWorker: workerConfig, + }); + + await orchestrator.start(); + try { + await conductor.invoke({ name: "batch-group" }, [ + { payload: { id: 1 }, group: "batch-tenant" }, + { payload: { id: 2 }, group: "batch-tenant" }, + ]); + const rows = await db.sql<{ group: string | null }[]>` + select "group" + from pgconductor._private_executions + where task_key = 'batch-group' + order by created_at asc, id asc + `; + expect(rows.map((row) => row.group)).toEqual(["batch-tenant", "batch-tenant"]); + + await waitForCondition(() => started.length === 1); + await new Promise((resolve) => setTimeout(resolve, 100)); + const first = started[0]; + expect(first === 1 || first === 2).toBe(true); + blockers.get(first!)?.resolve(); + await waitForCondition(() => started.length === 2); + expect(new Set(started)).toEqual(new Set([1, 2])); + } finally { + blockers.forEach((blocker) => blocker.resolve()); + await orchestrator.stop(); + } + }, 30000); + + test("rejects invalid task and group concurrency values", async () => { + const db = await pool.child(); + databases.push(db); + const definition = defineTask({ name: "validation" }); + const conductor = Conductor.create({ + sql: db.sql, + tasks: TaskSchemas.fromSchema([definition]), + context: {}, + }); + const invalidValues = [0, -1, 1.5]; + + for (const value of invalidValues) { + expect(() => + conductor.createTask( + { name: "validation", concurrency: value }, + { invocable: true }, + async () => {}, + ), + ).toThrow("concurrency must be a positive integer"); + expect(() => + conductor.createTask( + { name: "validation", groupConcurrency: value }, + { invocable: true }, + async () => {}, + ), + ).toThrow("groupConcurrency must be a positive integer"); + } + }, 30000); +}); diff --git a/packages/pgconductor-js/tests/mocks/in-memory-database-client.ts b/packages/pgconductor-js/tests/mocks/in-memory-database-client.ts index d9e8386..487bf85 100644 --- a/packages/pgconductor-js/tests/mocks/in-memory-database-client.ts +++ b/packages/pgconductor-js/tests/mocks/in-memory-database-client.ts @@ -39,6 +39,7 @@ interface StoredExecution { id: string; task_key: string; queue: string; + group: string | null; payload: Payload; state: "pending" | "running" | "completed" | "failed"; run_at: Date; @@ -57,7 +58,6 @@ interface StoredExecution { orchestrator_id: string | null; parent_execution_id: string | null; parent_step_key: string | null; - slot_group_number: number | null; created_at: Date; updated_at: Date; } @@ -78,6 +78,7 @@ interface StoredTask { window_start: string | null; window_end: string | null; concurrency: number | null; + group_concurrency: number | null; } interface StoredCronSchedule { @@ -206,11 +207,14 @@ export class InMemoryDatabaseClient implements IDatabaseClient { // Release executions claimed by stale orchestrator for (const exec of this.executions.values()) { if (exec.orchestrator_id === orchestrator.id && exec.state === "running") { - exec.state = "pending"; + exec.state = exec.cancelled ? "failed" : "pending"; + exec.last_error = exec.cancelled + ? exec.last_error || "Task was cancelled" + : exec.last_error; exec.orchestrator_id = null; - exec.slot_group_number = null; } } + this.orchestrators.delete(orchestrator.id); } } } @@ -246,6 +250,12 @@ export class InMemoryDatabaseClient implements IDatabaseClient { _opts?: { signal?: AbortSignal }, ): Promise { this.orchestrators.delete(args.orchestratorId); + for (const exec of this.executions.values()) { + if (exec.orchestrator_id !== args.orchestratorId || exec.state !== "running") continue; + exec.state = exec.cancelled ? "failed" : "pending"; + exec.last_error = exec.cancelled ? exec.last_error || "Task was cancelled" : exec.last_error; + exec.orchestrator_id = null; + } } // ============================================================================ @@ -283,6 +293,7 @@ export class InMemoryDatabaseClient implements IDatabaseClient { window_start: taskSpec.window?.[0] || null, window_end: taskSpec.window?.[1] || null, concurrency: taskSpec.concurrency || null, + group_concurrency: taskSpec.groupConcurrency || null, }; this.tasks.set(this.taskId(taskSpec.key, task.queue), task); } @@ -312,14 +323,17 @@ export class InMemoryDatabaseClient implements IDatabaseClient { ): Promise { const results: Execution[] = []; const now = this.getInternalTime(); - const taskKeysWithConcurrency = new Set(args.taskKeysWithConcurrency || []); const filterTaskKeys = new Set(args.filterTaskKeys || []); const concurrencyCount = new Map(); - // Count running executions per task for concurrency limits + // Count running executions per task for concurrency limits. for (const exec of this.executions.values()) { - if (exec.state === "running" && taskKeysWithConcurrency.has(exec.task_key)) { - concurrencyCount.set(exec.task_key, (concurrencyCount.get(exec.task_key) || 0) + 1); + const task = this.tasks.get(this.taskId(exec.task_key, exec.queue)); + if (exec.state === "running" && task?.concurrency != null) { + concurrencyCount.set( + this.taskId(exec.task_key, exec.queue), + (concurrencyCount.get(this.taskId(exec.task_key, exec.queue)) || 0) + 1, + ); } } @@ -349,41 +363,34 @@ export class InMemoryDatabaseClient implements IDatabaseClient { if (parent && parent.state !== "completed") continue; } - // Check concurrency limit - if (taskKeysWithConcurrency.has(exec.task_key)) { - const task = this.tasks.get(this.taskId(exec.task_key, exec.queue)); - const limit = task?.concurrency || 1; - const current = concurrencyCount.get(exec.task_key) || 0; - if (current >= limit) continue; + const task = this.tasks.get(this.taskId(exec.task_key, exec.queue)); + // Check concurrency limit. + if (task?.concurrency != null) { + const current = concurrencyCount.get(this.taskId(exec.task_key, exec.queue)) || 0; + if (current >= task.concurrency) continue; + } + if (task?.group_concurrency && exec.group) { + const activeGroup = Array.from(this.executions.values()).filter( + (other) => + other.queue === exec.queue && + other.task_key === exec.task_key && + other.group === exec.group && + other.state === "running", + ).length; + if (activeGroup >= task.group_concurrency) continue; } - // Claim execution with a fresh fencing token and persist slot ownership. + // Claim execution. exec.state = "running"; exec.attempts += 1; exec.orchestrator_id = args.orchestratorId; - if (taskKeysWithConcurrency.has(exec.task_key)) { - const used = new Set( - Array.from(this.executions.values()) - .filter( - (other) => - other.queue === exec.queue && - other.task_key === exec.task_key && - other.state === "running", - ) - .map((other) => other.slot_group_number), - ); - const limit = this.tasks.get(this.taskId(exec.task_key, exec.queue))?.concurrency || 1; - for (let slot = 1; slot <= limit; slot++) { - if (!used.has(slot)) { - exec.slot_group_number = slot; - break; - } - } - } // Update concurrency count - if (taskKeysWithConcurrency.has(exec.task_key)) { - concurrencyCount.set(exec.task_key, (concurrencyCount.get(exec.task_key) || 0) + 1); + if (task?.concurrency != null) { + concurrencyCount.set( + this.taskId(exec.task_key, exec.queue), + (concurrencyCount.get(this.taskId(exec.task_key, exec.queue)) || 0) + 1, + ); } results.push({ @@ -397,8 +404,8 @@ export class InMemoryDatabaseClient implements IDatabaseClient { last_error: exec.last_error, dedupe_key: exec.dedupe_key || undefined, cron_expression: exec.cron_expression || undefined, + group: exec.group, locked_by: exec.orchestrator_id || "", - slot_group_number: exec.slot_group_number || undefined, }); if (results.length >= args.batchSize) break; @@ -443,7 +450,6 @@ export class InMemoryDatabaseClient implements IDatabaseClient { exec.state = "completed"; exec.result = result.result || null; exec.orchestrator_id = null; - exec.slot_group_number = null; // Wake up parent if waiting if (exec.parent_execution_id) { @@ -474,7 +480,6 @@ export class InMemoryDatabaseClient implements IDatabaseClient { case "failed": { exec.last_error = result.error; exec.orchestrator_id = null; - exec.slot_group_number = null; const task = this.tasks.get(this.taskId(exec.task_key, exec.queue)); const maxAttempts = task?.max_attempts || 3; @@ -517,7 +522,6 @@ export class InMemoryDatabaseClient implements IDatabaseClient { case "released": { exec.state = "pending"; exec.orchestrator_id = null; - exec.slot_group_number = null; if (result.reschedule_in_ms === "infinity") { exec.run_at = new Date(8640000000000000); // Max date @@ -533,7 +537,6 @@ export class InMemoryDatabaseClient implements IDatabaseClient { exec.state = "failed"; exec.last_error = result.error; exec.orchestrator_id = null; - exec.slot_group_number = null; // Fail parent if waiting if (exec.parent_execution_id) { @@ -561,6 +564,7 @@ export class InMemoryDatabaseClient implements IDatabaseClient { task_key: result.child_task_name, queue: result.child_task_queue, payload: result.child_payload || {}, + group: result.group, parent_execution_id: exec.id, parent_step_key: result.step_key, }); @@ -570,7 +574,6 @@ export class InMemoryDatabaseClient implements IDatabaseClient { exec.waiting_on_execution_id = childId; exec.waiting_step_key = result.step_key; exec.orchestrator_id = null; - exec.slot_group_number = null; if (result.timeout_ms === "infinity") { exec.waiting_timeout_at = new Date(8640000000000000); @@ -598,7 +601,6 @@ export class InMemoryDatabaseClient implements IDatabaseClient { // exec.state = "pending"; // exec.run_at = new Date(8640000000000000); // Wait indefinitely // exec.orchestrator_id = null; - // exec.slot_group_number = null; // break; // } @@ -623,7 +625,6 @@ export class InMemoryDatabaseClient implements IDatabaseClient { // exec.state = "pending"; // exec.run_at = new Date(8640000000000000); // Wait indefinitely // exec.orchestrator_id = null; - // exec.slot_group_number = null; // break; // } } @@ -678,6 +679,7 @@ export class InMemoryDatabaseClient implements IDatabaseClient { id, task_key: spec.task_key, queue: spec.queue, + group: spec.group || null, payload: spec.payload || {}, state: "pending", run_at: spec.run_at || now, @@ -696,7 +698,6 @@ export class InMemoryDatabaseClient implements IDatabaseClient { orchestrator_id: null, parent_execution_id: spec.parent_execution_id || null, parent_step_key: spec.parent_step_key || null, - slot_group_number: null, created_at: now, updated_at: now, }; @@ -775,7 +776,6 @@ export class InMemoryDatabaseClient implements IDatabaseClient { exec.last_error = "superseded by reinvoke"; exec.dedupe_key = null; exec.orchestrator_id = null; - exec.slot_group_number = null; // Will create new execution below } else { // Unlocked execution - update it with new values (replace behavior) @@ -961,6 +961,7 @@ export class InMemoryDatabaseClient implements IDatabaseClient { run_at: nextRun, dedupe_key: dedupeKey, cron_expression: exec.cron_expression, + group: exec.group, }); } diff --git a/packages/pgconductor-js/tests/unit/in-memory-database-client.test.ts b/packages/pgconductor-js/tests/unit/in-memory-database-client.test.ts index 8c32db3..8b57b09 100644 --- a/packages/pgconductor-js/tests/unit/in-memory-database-client.test.ts +++ b/packages/pgconductor-js/tests/unit/in-memory-database-client.test.ts @@ -51,7 +51,6 @@ describe("InMemoryDatabaseClient", () => { orchestratorId: "test-orch", queueName: "default", batchSize: 10, - taskKeysWithConcurrency: [], filterTaskKeys: [], }); @@ -77,7 +76,6 @@ describe("InMemoryDatabaseClient", () => { orchestratorId: "test-orch", queueName: "default", batchSize: 10, - taskKeysWithConcurrency: [], filterTaskKeys: [], }); @@ -124,7 +122,6 @@ describe("InMemoryDatabaseClient", () => { orchestratorId: "test-orch", queueName: "default", batchSize: 10, - taskKeysWithConcurrency: [], filterTaskKeys: [], }); @@ -177,7 +174,6 @@ describe("InMemoryDatabaseClient", () => { orchestratorId: "test-orch", queueName: "default", batchSize: 10, - taskKeysWithConcurrency: [], filterTaskKeys: [], }); @@ -201,7 +197,6 @@ describe("InMemoryDatabaseClient", () => { orchestratorId: "test-orch", queueName: "default", batchSize: 10, - taskKeysWithConcurrency: [], filterTaskKeys: [], }); @@ -263,7 +258,6 @@ describe("InMemoryDatabaseClient", () => { orchestratorId: "test-orch", queueName: "default", batchSize: 10, - taskKeysWithConcurrency: [], filterTaskKeys: [], }); @@ -349,7 +343,6 @@ describe("InMemoryDatabaseClient", () => { orchestratorId: "test-orch", queueName: "default", batchSize: 10, - taskKeysWithConcurrency: [], filterTaskKeys: [], }); @@ -390,7 +383,6 @@ describe("InMemoryDatabaseClient", () => { orchestratorId: "test-orch", queueName: "default", batchSize: 10, - taskKeysWithConcurrency: [], filterTaskKeys: [], }); @@ -449,7 +441,6 @@ describe("InMemoryDatabaseClient", () => { orchestratorId: "test-orch", queueName: "default", batchSize: 10, - taskKeysWithConcurrency: [], filterTaskKeys: [], }); @@ -494,7 +485,6 @@ describe("InMemoryDatabaseClient", () => { orchestratorId: "test-orch", queueName: "default", batchSize: 10, - taskKeysWithConcurrency: [], filterTaskKeys: [], }); @@ -542,7 +532,6 @@ describe("InMemoryDatabaseClient", () => { orchestratorId: "test-orch", queueName: "default", batchSize: 10, - taskKeysWithConcurrency: [], filterTaskKeys: [], }); @@ -586,7 +575,6 @@ describe("InMemoryDatabaseClient", () => { orchestratorId: "test-orch", queueName: "default", batchSize: 10, - taskKeysWithConcurrency: [], filterTaskKeys: [], }); @@ -612,7 +600,6 @@ describe("InMemoryDatabaseClient", () => { orchestratorId: "test-orch", queueName: "default", batchSize: 10, - taskKeysWithConcurrency: [], filterTaskKeys: [], }); @@ -659,7 +646,6 @@ describe("InMemoryDatabaseClient", () => { orchestratorId: "test-orch", queueName: "default", batchSize: 10, - taskKeysWithConcurrency: [], filterTaskKeys: [], }); @@ -685,7 +671,6 @@ describe("InMemoryDatabaseClient", () => { orchestratorId: "test-orch", queueName: "default", batchSize: 10, - taskKeysWithConcurrency: [], filterTaskKeys: [], }); @@ -721,7 +706,6 @@ describe("InMemoryDatabaseClient", () => { orchestratorId: "test-orch", queueName: "default", batchSize: 1, - taskKeysWithConcurrency: [], filterTaskKeys: [], }) )[0]!; @@ -784,7 +768,6 @@ describe("InMemoryDatabaseClient", () => { orchestratorId: "test-orch", queueName: "default", batchSize: 10, - taskKeysWithConcurrency: [], filterTaskKeys: [], }); @@ -798,7 +781,6 @@ describe("InMemoryDatabaseClient", () => { orchestratorId: "test-orch", queueName: "default", batchSize: 10, - taskKeysWithConcurrency: [], filterTaskKeys: [], }); @@ -836,7 +818,6 @@ describe("InMemoryDatabaseClient", () => { orchestratorId: "test-orch", queueName: "default", batchSize: 10, - taskKeysWithConcurrency: ["limited-task"], filterTaskKeys: [], }); @@ -847,7 +828,6 @@ describe("InMemoryDatabaseClient", () => { orchestratorId: "test-orch", queueName: "default", batchSize: 10, - taskKeysWithConcurrency: ["limited-task"], filterTaskKeys: [], }); @@ -869,7 +849,6 @@ describe("InMemoryDatabaseClient", () => { orchestratorId: "test-orch", queueName: "default", batchSize: 10, - taskKeysWithConcurrency: ["limited-task"], filterTaskKeys: [], }); @@ -916,7 +895,6 @@ describe("InMemoryDatabaseClient", () => { orchestratorId: "test-orch", queueName: "default", batchSize: 10, - taskKeysWithConcurrency: [], filterTaskKeys: [], }); diff --git a/packages/pgconductor-js/tests/unit/throttle-debounce.test.ts b/packages/pgconductor-js/tests/unit/throttle-debounce.test.ts index 23816d5..6ae108d 100644 --- a/packages/pgconductor-js/tests/unit/throttle-debounce.test.ts +++ b/packages/pgconductor-js/tests/unit/throttle-debounce.test.ts @@ -134,7 +134,6 @@ describe("Throttle and Debounce", () => { queueName: "default", batchSize: 100, orchestratorId: "test-orch", - taskKeysWithConcurrency: [], filterTaskKeys: [], }); From 14f2b70d22e4f4a08ffacfc1ca5e637ffec9dbff Mon Sep 17 00:00:00 2001 From: psteinroe Date: Tue, 15 Sep 2026 11:47:40 +0000 Subject: [PATCH 2/3] docs(scheduling): explain soft group limits --- designs/GROUP_CONCURRENCY.md | 341 --------------------- docs/content/task-execution/concurrency.md | 23 +- 2 files changed, 11 insertions(+), 353 deletions(-) delete mode 100644 designs/GROUP_CONCURRENCY.md diff --git a/designs/GROUP_CONCURRENCY.md b/designs/GROUP_CONCURRENCY.md deleted file mode 100644 index 67ed662..0000000 --- a/designs/GROUP_CONCURRENCY.md +++ /dev/null @@ -1,341 +0,0 @@ -# Group-Based Concurrency Design - -## Current System - -### Schema -``` -_private_tasks: - - concurrency_limit (integer, nullable) - -_private_concurrency_slots: - - task_key + slot_group_number (PK) - - capacity (always 1) - - used (0 or 1) -``` - -### Characteristics -- **Single dimension**: One global concurrency limit per task (e.g., `concurrency: 5`) -- **Pre-allocated slots**: Slots created at task registration (5 rows for limit=5) -- **Batch processing**: Query pairs up to 100 executions with available slots using ROW_NUMBER -- **Fast path**: When no tasks have concurrency, uses simpler query (no slot overhead) - -### Limitations -Cannot express multi-dimensional rate limits like: -- "Max 20 executions per tenant" -- "Max 1 execution per message" -- Both constraints applied simultaneously to same task - -## Problem Statement - -### Use Case -``` -Task: "process-message" -Requirements: - - Max 20 concurrent executions per tenant (tenant-level rate limit) - - Max 1 concurrent execution per message (message-level deduplication) - - Constraints apply independently and simultaneously - -Examples: - { tenant: "acme", message: "msg-1" } → needs slots from tenant=acme AND message=msg-1 - { tenant: "acme", message: "msg-2" } → needs slots from tenant=acme AND message=msg-2 - { tenant: "beta", message: "msg-1" } → needs slots from tenant=beta AND message=msg-1 -``` - -### Core Challenges -1. **Dynamic slot space**: Cannot pre-create slots for unknown tenant/message IDs -2. **Multi-dimensional claiming**: Each execution needs slots from multiple independent groups -3. **All-or-nothing atomicity**: Must claim ALL required group slots or none (avoid deadlocks) -4. **Optional groups**: Some executions may skip certain groups (no key provided) -5. **Query complexity**: Current batch approach breaks down - each execution has unique group keys - -## Design Options - -### Option 1: Fully Independent Groups (Recommended) - -#### Schema Changes -``` -_private_task_concurrency_groups: - - task_key + group_name (PK) - - concurrency_limit (integer) - -_private_concurrency_slots: - - task_key + group_name + group_key + slot_number (PK) - - used_by_execution_id (nullable UUID) - -_private_execution_group_keys: - - execution_id + group_name (PK) - - group_key (text) -``` - -#### API Surface -```typescript -// Task definition -defineTask({ - name: "process-message", - concurrency: { - tenant: 20, // max 20 per tenant - message: 1, // max 1 per message - // groups are optional per execution - } -}) - -// Invocation -invoke({ name: "process-message" }, { - payload: { ... }, - concurrencyGroups: { - tenant: "acme", - message: "msg-123" - } -}) -``` - -#### Semantics -- **Independent constraints**: Each group is an independent concurrency limit -- **All must pass**: Execution must satisfy ALL provided group constraints -- **Optional participation**: If execution doesn't provide key for a group, that group's constraint is skipped -- **Example**: `{ tenant: 20, message: 1 }` means "max 20 per tenant" AND "max 1 per message" - -#### Slot Lifecycle -1. **Creation**: On-demand when first execution with that (task, group, key) arrives -2. **Claiming**: Execution atomically claims one slot from EACH provided group -3. **Release**: All slots released when execution completes/fails/released -4. **Garbage collection**: Delete unused slots after TTL (e.g., 24 hours of inactivity) - -#### Query Strategy (Conceptual) -``` -for each execution candidate: - 1. extract concurrencyGroups from payload/metadata - 2. for each group defined on task: - - if execution provides key for group: - → check if slot available for (task, group, key) - → if no slot available → skip this execution - - if execution doesn't provide key: - → skip this group check (no constraint) - 3. if ALL checks pass: - → claim all required slots atomically (for update) - → return execution for processing -``` - -#### Pros -- ✅ Flexible: supports any number of groups -- ✅ Optional: groups can be skipped per execution -- ✅ No deadlocks: all-or-nothing claiming -- ✅ Intuitive semantics -- ✅ Matches stated use case exactly - -#### Cons -- ❌ Query complexity: O(executions × groups) checks needed -- ❌ Hard to batch: each execution has unique group keys -- ❌ Slot table growth: one row per unique (task, group, key, slot) combination -- ❌ Performance unknown: needs prototyping - -### Option 2: Single Primary Group + Global Limit - -#### Simplified Model -```typescript -defineTask({ - name: "process-message", - concurrency: 100, // global task limit (existing system) - concurrencyGroup: { // ONE group dimension only - name: "tenant", - limit: 20 - } -}) -``` - -#### Semantics -- Global limit (100) applied first using existing system -- Group limit (20 per tenant) applied as secondary filter -- Only one group dimension supported - -#### Pros -- ✅ Simpler to implement -- ✅ Covers common case (rate limiting per tenant) -- ✅ Can still use batch processing somewhat -- ✅ Lower query complexity - -#### Cons -- ❌ Only one group (not extensible to multi-group case) -- ❌ Doesn't solve stated use case (tenant + message simultaneously) -- ❌ Coupling between global and group limits may be confusing - -### Option 3: Composite Keys (Pre-computed) - -#### Idea -User declares all possible group key combinations upfront - -```typescript -defineTask({ - name: "process-message", - concurrency: { - tenant: ["acme", "beta", "gamma"], // fixed list of keys - message: 1 // per-key limit - } -}) -``` - -#### Pros -- ✅ Can pre-create slots (known space) -- ✅ Query similar to current system - -#### Cons -- ❌ Not dynamic (can't handle new tenants at runtime) -- ❌ Combinatorial explosion with multiple groups (tenant × message slots) -- ❌ Doesn't match stated use case (unknown keys upfront) - -## Open Design Questions - -### 1. Query Performance Strategy - -**Problem:** Current query efficiently batches 100 executions. With groups, need per-execution evaluation. - -**Options:** -- **A) PostgreSQL function with loop**: Procedural approach iterating through candidates -- **B) CTE-based with LATERAL joins**: Declarative query checking all groups per execution -- **C) Two-phase approach**: Batch-lock likely slots, then check per execution - -**Decision needed:** Which approach balances performance vs complexity? - -### 2. Slot Creation & Lifecycle - -**When to create slots?** -- Lazy: on first execution with that group key -- Eager: during registration if keys are known upfront -- Explicit: via separate admin API - -**Garbage collection:** -- Delete slots unused for X hours/days? -- Keep forever (infinite growth)? -- Manual cleanup API? - -**Decision needed:** What's the lifecycle management strategy? - -### 3. Atomicity & Locking Model - -**Scenario:** Execution needs [tenant=acme slot, message=msg-1 slot] - -**Option A - Optimistic:** -- Check both available -- Attempt to claim both -- If second claim fails → rollback first - -**Option B - Pessimistic:** -- Lock all required slots with for update -- Then assign to execution - -**Option C - Hierarchical:** -- Order groups (alphabetically?) to prevent deadlocks -- Always lock in same order - -**Decision needed:** Which locking strategy avoids deadlocks and contention best? - -### 4. Backwards Compatibility - -**Current API:** -```typescript -concurrency: 5 -``` - -**Proposed API:** -```typescript -concurrency: { tenant: 20 } -``` - -**Migration Options:** -- **A) Support both**: `number | Record` (number maps to global limit) -- **B) Explicit migration**: `concurrency: 5` becomes `concurrency: { default: 5 }` -- **C) Deprecate number**: Breaking change, require object format - -**Decision needed:** Migration path for existing tasks? - -### 5. Fast Path Optimization - -**Current:** When no tasks have concurrency, uses optimized query (no slot logic) - -**With groups:** Need to: -1. Check if task defines groups -2. Check if execution provides group keys -3. Mix grouped and ungrouped tasks in same queue - -**Options:** -- **A) Multiple fast paths**: ungrouped, single-group, multi-group -- **B) Unified query**: Always handle groups (slower but simpler) -- **C) Separate queues**: Group-enabled tasks use different queue - -**Decision needed:** Can we preserve fast path optimization? - -### 6. Payload vs Metadata - -**Where do group keys live?** - -**Option A - In payload:** -```typescript -invoke({ name: "task" }, { - tenant: "acme", // part of payload - message: "msg-1" -}) -``` -- ✅ Simple -- ❌ Mixes business data with infrastructure concerns - -**Option B - Separate metadata:** -```typescript -invoke({ name: "task" }, { - payload: { /* business data */ }, - concurrencyGroups: { - tenant: "acme", - message: "msg-1" - } -}) -``` -- ✅ Clean separation -- ❌ More verbose -- ❌ Need to persist metadata separately - -**Decision needed:** API ergonomics and data modeling? - -## Recommendation - -### Start with Option 1 (Fully Independent Groups) - -**Rationale:** -1. Matches stated use case exactly (tenant + message simultaneously) -2. Extensible to N groups without redesign -3. Optional groups handle partial constraints elegantly -4. Most flexible for future requirements - -### But Prototype Query Performance First - -Before full implementation, validate performance with prototype: - -**Test scenario:** -- 1000 pending executions -- 10 concurrency groups defined -- 100 unique keys per group -- Measure query time for group-based slot matching -- Target: maintain ~300 tasks/sec throughput (current optimized baseline) - -**Critical query sections to prototype:** -1. Extract concurrencyGroups from execution metadata (JSONB operations) -2. Check slot availability across N groups (LATERAL joins? function?) -3. Atomic all-or-nothing claiming (locking strategy) -4. Mixing grouped and ungrouped tasks (fast path preservation) - -**Success criteria:** -- Throughput degradation < 50% for grouped tasks -- Ungrouped tasks maintain fast path performance -- No deadlocks under concurrent worker load - -## Next Steps - -1. **Design review**: Validate Option 1 approach and semantics -2. **Decide on open questions**: Particularly #1 (query strategy) and #3 (locking) -3. **Query prototype**: Implement conceptual query logic in SQL -4. **Performance benchmark**: Compare against current system (diagnose-variance.ts) -5. **API finalization**: Lock down TypeScript interfaces -6. **Implementation**: Schema migration → query → tests → docs - -## Related Documents - -- `CONCURRENCY.md` - Current single-dimension concurrency implementation -- `perf/diagnose-variance.ts` - Benchmarking methodology for throughput testing diff --git a/docs/content/task-execution/concurrency.md b/docs/content/task-execution/concurrency.md index eb93dd5..81be609 100644 --- a/docs/content/task-execution/concurrency.md +++ b/docs/content/task-execution/concurrency.md @@ -10,7 +10,8 @@ Control the maximum number of concurrent executions for a specific task: const processVideo = conductor.createTask( { name: "process-video", - concurrency: 3, // Max 3 videos processing at once + concurrency: 3, // Soft max of 3 videos at once + groupConcurrency: 1, // Soft max of 1 per invocation group }, { invocable: true }, async (event, ctx) => { @@ -20,22 +21,14 @@ const processVideo = conductor.createTask( ); ``` -When the limit is reached, additional executions wait in the queue until a slot becomes available. +When the limit is reached, additional executions remain queued until capacity becomes available. ## How It Works -Postgres Conductor uses a slot-based system to enforce concurrency limits: - -1. **Slot allocation**: When a task has `concurrency: N`, Postgres creates N slots in the `_private_concurrency_slots` table -2. **Claiming slots**: Workers claim available slots using `FOR UPDATE SKIP LOCKED` -3. **Execution**: Task runs while holding the slot -4. **Release**: Slot is released when execution completes or fails +Postgres Conductor evaluates active executions when claiming work. Task and group limits are coordinated with `FOR UPDATE SKIP LOCKED`; limits are intentionally soft across concurrent workers. Grouped candidates whose group is full do not consume task-level capacity, so another available group can be claimed in the same batch. This happens entirely in Postgres - no external coordination needed. -> [!WARNING] -> Setting concurrency on any task in a queue reduces throughput by up to 50% for the entire queue, regardless of how many tasks have concurrency limits. - ## Concurrency vs Worker Concurrency **Task-level concurrency** (this page): @@ -50,8 +43,14 @@ This happens entirely in Postgres - no external coordination needed. - Set on worker/queue with `config: { concurrency }` - Independent per worker instance +Child invocations use the group supplied to `ctx.invoke`. Dynamic cron schedules accept `group` alongside `cron`, and each subsequent cron execution preserves it. + ## What's Next? - [Worker Configuration](../api/worker-config.md) - Configure worker-level concurrency -- [Priority](priority.md) - Control execution order when waiting for slots +- [Priority](priority.md) - Control execution order while work is queued - [Batching](batching.md) - Process multiple executions together + +## Group concurrency + +`group` may be supplied when invoking a task. `groupConcurrency` limits active executions within each `(queue, task, group)` scope; ungrouped invocations bypass that limit. Task and group limits compose and are intentionally soft across concurrent workers. From a127ed351e4374a703d170809f58b5177e024cb3 Mon Sep 17 00:00:00 2001 From: psteinroe Date: Fri, 18 Sep 2026 12:02:46 +0000 Subject: [PATCH 3/3] perf(scheduling): optimize execution claims --- migrations/0000000001_setup.sql | 12 +- packages/pgconductor-js/src/generated/sql.ts | 12 +- packages/pgconductor-js/src/query-builder.ts | 112 ++++++++++++------- 3 files changed, 78 insertions(+), 58 deletions(-) diff --git a/migrations/0000000001_setup.sql b/migrations/0000000001_setup.sql index 3284fc6..dd65de7 100644 --- a/migrations/0000000001_setup.sql +++ b/migrations/0000000001_setup.sql @@ -247,16 +247,10 @@ begin v_partition_name ); - -- indexes used to count active executions for soft concurrency limits + -- covering index used to count active executions for soft task and group limits execute format( - 'create index if not exists %I on pgconductor.%I (task_key) where locked_at is not null and failed_at is null and completed_at is null', - 'idx_' || v_partition_name || '_active_task', - v_partition_name - ); - - execute format( - 'create index if not exists %I on pgconductor.%I (task_key, "group") where "group" is not null and locked_at is not null and failed_at is null and completed_at is null', - 'idx_' || v_partition_name || '_active_task_group', + 'create index if not exists %I on pgconductor.%I (task_key, "group") where locked_at is not null and failed_at is null and completed_at is null', + 'idx_' || v_partition_name || '_active_concurrency', v_partition_name ); diff --git a/packages/pgconductor-js/src/generated/sql.ts b/packages/pgconductor-js/src/generated/sql.ts index f73bae2..97abfcd 100644 --- a/packages/pgconductor-js/src/generated/sql.ts +++ b/packages/pgconductor-js/src/generated/sql.ts @@ -263,16 +263,10 @@ begin v_partition_name ); - -- indexes used to count active executions for soft concurrency limits + -- covering index used to count active executions for soft task and group limits execute format( - 'create index if not exists %I on pgconductor.%I (task_key) where locked_at is not null and failed_at is null and completed_at is null', - 'idx_' || v_partition_name || '_active_task', - v_partition_name - ); - - execute format( - 'create index if not exists %I on pgconductor.%I (task_key, "group") where "group" is not null and locked_at is not null and failed_at is null and completed_at is null', - 'idx_' || v_partition_name || '_active_task_group', + 'create index if not exists %I on pgconductor.%I (task_key, "group") where locked_at is not null and failed_at is null and completed_at is null', + 'idx_' || v_partition_name || '_active_concurrency', v_partition_name ); diff --git a/packages/pgconductor-js/src/query-builder.ts b/packages/pgconductor-js/src/query-builder.ts index 6ec295a..664872c 100644 --- a/packages/pgconductor-js/src/query-builder.ts +++ b/packages/pgconductor-js/src/query-builder.ts @@ -258,24 +258,57 @@ export class QueryBuilder { filterTaskKeys, }: GetExecutionsArgs): PendingQuery { return this.sql` - with active_tasks as ( - select e.task_key, count(*)::integer as active_count + -- Read limit mode from the database so rolling workers cannot bypass new limits. + with task_limits as materialized ( + select exists ( + select 1 + from pgconductor._private_tasks t + where t.queue = ${queueName}::text + and (t.concurrency_limit is not null or t.group_concurrency_limit is not null) + ) as enabled + ), unconstrained_candidates as ( + select e.id + from pgconductor._private_executions e + where not (select enabled from task_limits) + and e.queue = ${queueName}::text + and e.run_at <= pgconductor._private_current_time() + and e.is_available = true + ${filterTaskKeys?.length ? this.sql`and not (e.task_key = any(${this.sql.array(filterTaskKeys)}::text[]))` : this.sql``} + order by e.priority asc, e.run_at asc, e.created_at asc, e.id asc + limit ${batchSize}::integer + for update of e skip locked + ), active_executions as materialized ( + select e.task_key, e."group" from pgconductor._private_executions e - where e.queue = ${queueName}::text + where (select enabled from task_limits) + and e.queue = ${queueName}::text and e.locked_at is not null and e.failed_at is null and e.completed_at is null + ), active_tasks as ( + select e.task_key, count(*)::integer as active_count + from active_executions e group by e.task_key ), active_groups as ( select e.task_key, e."group", count(*)::integer as active_count - from pgconductor._private_executions e - where e.queue = ${queueName}::text - and e."group" is not null - and e.locked_at is not null - and e.failed_at is null - and e.completed_at is null + from active_executions e + where e."group" is not null group by e.task_key, e."group" - ), ranked as ( + ), + -- A full task makes the candidate branch a no-op instead of scanning its backlog. + available_tasks as materialized ( + select + t.key, + t.queue, + t.concurrency_limit, + t.group_concurrency_limit, + coalesce(at.active_count, 0) as active_task_count + from pgconductor._private_tasks t + left join active_tasks at on at.task_key = t.key + where t.queue = ${queueName}::text + and (t.concurrency_limit is null + or coalesce(at.active_count, 0) < t.concurrency_limit) + ), candidates as ( select e.id, e.task_key, @@ -286,58 +319,57 @@ export class QueryBuilder { e."group", t.concurrency_limit, t.group_concurrency_limit, - coalesce(at.active_count, 0) as active_task_count, - coalesce(ag.active_count, 0) as active_group_count, - row_number() over ( - partition by e.task_key - order by e.priority asc, e.run_at asc, e.created_at asc, e.id asc - ) as task_rank, - row_number() over ( - partition by e.task_key, e."group" - order by e.priority asc, e.run_at asc, e.created_at asc, e.id asc - ) as group_rank + t.active_task_count, + coalesce(ag.active_count, 0) as active_group_count from pgconductor._private_executions e - left join pgconductor._private_tasks t - on t.key = e.task_key and t.queue = e.queue - left join active_tasks at on at.task_key = e.task_key + join available_tasks t on t.key = e.task_key and t.queue = e.queue left join active_groups ag on ag.task_key = e.task_key and ag."group" = e."group" - where e.queue = ${queueName}::text + where (select enabled from task_limits) + and (select exists (select 1 from available_tasks)) + and e.queue = ${queueName}::text and e.run_at <= pgconductor._private_current_time() and e.is_available = true + and (t.group_concurrency_limit is null + or e."group" is null + or coalesce(ag.active_count, 0) < t.group_concurrency_limit) ${filterTaskKeys?.length ? this.sql`and not (e.task_key = any(${this.sql.array(filterTaskKeys)}::text[]))` : this.sql``} + -- Bound both window functions to one locked candidate batch. + order by e.priority asc, e.run_at asc, e.created_at asc, e.id asc + limit ${batchSize}::integer + for update of e skip locked + ), group_ranked as ( + select c.*, + row_number() over ( + partition by c.task_key, c."group" + order by c.priority asc, c.run_at asc, c.created_at asc, c.id asc + ) as group_rank + from candidates c ), group_eligible as ( select r.*, row_number() over ( partition by r.task_key order by r.priority asc, r.run_at asc, r.created_at asc, r.id asc - ) as available_task_rank - from ranked r + ) as task_rank + from group_ranked r where r.group_concurrency_limit is null or r."group" is null or r.active_group_count + r.group_rank <= r.group_concurrency_limit ), eligible as ( - select r.id, r.priority, r.run_at, r.created_at + select r.id from group_eligible r where r.concurrency_limit is null - or r.active_task_count + r.available_task_rank <= r.concurrency_limit - order by r.priority asc, r.run_at asc, r.created_at asc, r.id asc - -- Keep a bounded candidate pool so SKIP LOCKED can backfill a batch. - limit greatest(${batchSize}::integer * 4, ${batchSize}::integer) - ), locked_candidates as ( - select e.id - from pgconductor._private_executions e - join eligible c on c.id = e.id - where e.queue = ${queueName}::text - order by c.priority asc, c.run_at asc, c.created_at asc, c.id asc - limit ${batchSize}::integer - for update of e skip locked + or r.active_task_count + r.task_rank <= r.concurrency_limit + ), claimable as ( + select id from unconstrained_candidates + union all + select id from eligible ), claimed as ( update pgconductor._private_executions e set attempts = e.attempts + 1, locked_by = ${orchestratorId}::uuid, locked_at = pgconductor._private_current_time() - from locked_candidates c + from claimable c where e.id = c.id and e.queue = ${queueName}::text and e.is_available = true returning e.id, e.task_key, e.queue, e.payload, e.waiting_on_execution_id, e.waiting_step_key, e.cancelled, e.last_error, e.dedupe_key, e.cron_expression,