From ffdafa4710abb1bd5d70371510200fda7e5fd02d Mon Sep 17 00:00:00 2001 From: psteinroe Date: Tue, 20 Oct 2026 16:53:47 +0000 Subject: [PATCH 1/2] fix(worker): release an attempt whose signal delivery fails during stop() --- AGENTS.md | 1 + docs/content/task-execution/signals.md | 2 +- packages/pgconductor-js/src/worker.ts | 4 + .../integration/worker-lifecycle.test.ts | 154 ++++++++++++++++++ 4 files changed, 160 insertions(+), 1 deletion(-) diff --git a/AGENTS.md b/AGENTS.md index ba5b49a..32285f2 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -271,6 +271,7 @@ Tables live in the `pgconductor` schema with a `_private_` prefix (`_private_exe - `_private_execution_signals` stores one row per `(execution_id, id)`; `seq` orders delivery and `delivered_at` is set when a `ctx.onSignal` handler returned (at least once per id) - Senders lock the execution row and increment `_private_executions.signal_count`. Settlement reads `signal_count` from its locked row, which is newer than its snapshot: a completion whose count exceeds the signals it can see, or that has undelivered signals for names the attempt listened for (`signal_names` in the result), is re-queued instead of completed - Claiming an execution reads its undelivered signals under the row lock senders take, so signals stored later have a higher `seq` and later claims read them with `seq > after_seq`. Signal reads and acknowledgements are fenced by the claim token. `SignalInbox` delivers in `seq` order and skips a signal whose name has no handler yet; it stays undelivered, so settlement re-queues if a handler for it registers later. On completion the worker drains the inbox (`close({ drain: true })`): signals already read are delivered before settling, and only ones stored after the last read re-queue +- A delivery failure (a throwing `ctx.onSignal` handler or a failed acknowledgement) aborts the attempt as `signal-failed`, except after the worker signal aborted: stopping ends acknowledgement retries, so the handler keeps its slot and the attempt is released once it returns - Re-queues are unbounded by design: each needs a signal stored after the previous attempt started, which the next attempt receives at claim time, so an execution re-runs only while signals keep arriving - `key` on `_private_executions` with a partial unique index makes at most one non-terminal, non-cancelled execution per `(task_key, key, queue)`; `signal_with_start()` takes a transaction advisory lock on the key, then signals the existing execution or inserts one diff --git a/docs/content/task-execution/signals.md b/docs/content/task-execution/signals.md index 033823c..43e777c 100644 --- a/docs/content/task-execution/signals.md +++ b/docs/content/task-execution/signals.md @@ -95,4 +95,4 @@ A handler does not have to wait for its signals: one that registers listeners an ## Errors -A signal handler that throws fails the attempt like an uncaught error in the handler, with the error `Signal handler for "" failed: `. The signal is not marked delivered, so the retry receives it again. +A signal handler that throws fails the attempt like an uncaught error in the handler, with the error `Signal handler for "" failed: `. The signal is not marked delivered, so the retry receives it again. During `orchestrator.stop()`, a signal handler that throws, or a signal that cannot be marked delivered, releases the attempt once the handler returns, without using up a retry. diff --git a/packages/pgconductor-js/src/worker.ts b/packages/pgconductor-js/src/worker.ts index dde09cd..6693fc8 100644 --- a/packages/pgconductor-js/src/worker.ts +++ b/packages/pgconductor-js/src/worker.ts @@ -573,6 +573,9 @@ export class Worker< { signal: this.signal }, ), onError: (error) => { + // Stopping ends the retries of an acknowledgement. The handler keeps its slot, and the + // attempt is released with the signal undelivered. + if (this.signal.aborted) return; taskAbortController.abort({ reason: "signal-failed", error: error.message }); }, }); @@ -705,6 +708,7 @@ export class Worker< } if (outcome.signalFailure) { + if (this.signal.aborted) return { ...identity, status: "released" } as const; return { ...identity, status: "failed", diff --git a/packages/pgconductor-js/tests/integration/worker-lifecycle.test.ts b/packages/pgconductor-js/tests/integration/worker-lifecycle.test.ts index 0f119a6..c4e3a7d 100644 --- a/packages/pgconductor-js/tests/integration/worker-lifecycle.test.ts +++ b/packages/pgconductor-js/tests/integration/worker-lifecycle.test.ts @@ -387,6 +387,160 @@ describe("drain()", () => { }); }); +describe("stop() while a handler runs a step", () => { + const ackStatement = "set delivered_at = pgconductor._private_current_time()"; + + // `ack` replaces the response of every signal acknowledgement + async function setup({ + signal, + ack = (query) => query, + onInput, + }: { + signal: boolean; + ack?: (query: Promise) => Promise; + onInput?: (signal: AbortSignal) => Promise; + }) { + const db = await child(); + const conductor = Conductor.create({ + sql: new Proxy(db.sql, { + apply(target, thisArg, args) { + const query = Reflect.apply(target, thisArg, args); + return String(args[0]).includes(ackStatement) ? ack(query) : query; + }, + }), + tasks: TaskSchemas.fromSchema([sessionDefinition]), + context: {}, + }); + const received = new Deferred(); + const stepRunning = new Deferred(); + const gate = new Deferred(); + let finished = false; + const task = conductor.createTask(sessionDefinition, { invocable: true }, async (_e, ctx) => { + ctx.onSignal("input", async () => { + received.resolve(); + await onInput?.(ctx.signal); + }); + if (signal) await received.promise; + await ctx.step("work", async () => { + stepRunning.resolve(); + await gate.promise; + }); + finished = true; + }); + await conductor.ensureInstalled(); + const orchestrator = Orchestrator.create({ + conductor, + tasks: [task], + defaultWorker: { pollIntervalMs: 50, flushIntervalMs: 50 }, + }); + const id = await conductor.invoke({ name: "session" }, {}); + if (signal) await conductor.signal(id, { name: "input", id: "s1" }); + await orchestrator.start(); + await stepRunning.promise; + const delivered = async () => { + const [s] = await db.sql<{ delivered: boolean }[]>` + select delivered_at is not null as delivered from pgconductor._private_execution_signals + `; + return s?.delivered; + }; + return { db, id, orchestrator, gate, delivered, finished: () => finished }; + } + + test("waits for the handler when no signal is involved", async () => { + const { db, id, orchestrator, gate, finished } = await setup({ signal: false }); + try { + const stopped = orchestrator.stop(); + expect(await within(stopped, 300)).toBe("timeout"); + gate.resolve(); + await stopped; + expect(finished()).toBe(true); + expect(await row(db, id)).toMatchObject({ completed: true, failed: false }); + } finally { + gate.resolve(); + await orchestrator.stop(); + } + }); + + test("completes once a slow acknowledgement returns", async () => { + const release = new Deferred(); + const { db, id, orchestrator, gate, delivered } = await setup({ + signal: true, + ack: (query) => query.then(async (rows) => (await release.promise, rows)), + }); + try { + gate.resolve(); + await sleep(200); + expect(await row(db, id)).toMatchObject({ completed: false, locked: true }); + release.resolve(); + await waitForCondition(async () => (await row(db, id))?.completed === true); + expect(await delivered()).toBe(true); + } finally { + release.resolve(); + gate.resolve(); + await orchestrator.stop(); + } + }); + + test("waits for the handler while an acknowledgement is in flight", async () => { + const release = new Deferred(); + const { db, id, orchestrator, gate, delivered, finished } = await setup({ + signal: true, + ack: (query) => query.then(async (rows) => (await release.promise, rows)), + }); + try { + const stopped = orchestrator.stop(); + expect(await within(stopped, 300)).toBe("timeout"); + release.resolve(); + expect(await within(stopped, 300)).toBe("timeout"); + gate.resolve(); + await stopped; + expect(finished()).toBe(true); + expect(await row(db, id)).toMatchObject({ completed: true, failed: false }); + expect(await delivered()).toBe(true); + } finally { + release.resolve(); + gate.resolve(); + await orchestrator.stop(); + } + }); + + test.each([ + ["stop() ends the acknowledgement's retries", { ack: () => Promise.reject(connectionError()) }], + [ + "the signal handler stops on ctx.signal", + { + onInput: (signal: AbortSignal) => + new Promise((_, reject) => + signal.addEventListener("abort", () => reject(new Error("stopped"))), + ), + }, + ], + ])("waits for the handler and releases it when %s", async (_name, options) => { + const { db, id, orchestrator, gate, delivered, finished } = await setup({ + signal: true, + ...options, + }); + try { + const stopped = orchestrator.stop(); + expect(await within(stopped, 300)).toBe("timeout"); + gate.resolve(); + await stopped; + expect(finished()).toBe(true); + expect(await row(db, id)).toMatchObject({ + completed: false, + failed: false, + locked: false, + attempts: 0, + last_error: null, + }); + expect(await delivered()).toBe(false); + } finally { + gate.resolve(); + await orchestrator.stop(); + } + }); +}); + test("stop() interrupts a worker registration that keeps retrying", async () => { const db = await child(); let attempts = 0; From a16e211f393c8a017a4aef9797b9e3ba305697f2 Mon Sep 17 00:00:00 2001 From: psteinroe Date: Tue, 20 Oct 2026 18:51:59 +0000 Subject: [PATCH 2/2] fix(worker): end an attempt only after its concurrent steps settle --- AGENTS.md | 1 + .../crafting-tasks/retries-and-steps.md | 2 + .../pgconductor-js/src/lib/map-concurrent.ts | 2 + packages/pgconductor-js/src/task-context.ts | 31 ++-- packages/pgconductor-js/src/worker.ts | 14 +- .../integration/worker-lifecycle.test.ts | 144 ++++++++++++++++++ .../tests/unit/lib/map-concurrent.test.ts | 47 ++++++ 7 files changed, 228 insertions(+), 13 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 32285f2..7d154f1 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -147,6 +147,7 @@ Tables live in the `pgconductor` schema with a `_private_` prefix (`_private_exe - `invoke()` inserts an independent execution and memoizes its id as a step in one statement. Only with `cancelWithParent` does it set `parent_execution_id`, together with `cancel_with_parent`; settlement fails a completed child that has a parent but neither a waiter nor that flag - Memoized by step key, so a replay returns the stored result: `step()`, `sleep()`, `invoke()`, `invokeAndWait()`, `waitForResult()`, `waitForEvent()`, `subscribe()`, `waitForAny()` and `signalWithStart()`. `emit()` (deduplicated only with a `dedupe_key`), `sendSignal()` (deduplicated by signal id), `schedule()` (replaces the pending occurrence), `unschedule()` and `cancel()` are not memoized and run again on every replay - `checkpoint()` hangs up when the abort signal fired or the time window closed; it persists nothing +- Each attempt tracks its step callbacks and their saves (`TaskContextOptions.steps`). Before it reports a result, the worker waits for those still running, bounded by `timeoutMs`, so a sibling's hangup or error (or `stop()`) does not end an attempt while a concurrent step runs. A timeout, cancellation or signal failure ends it at once - `step()` checkpoints after reading the step, so `fn` never starts once the attempt aborted, and hangs up as suspended when the claim fence rejects its save - Hangup pattern: abort execution and return never-resolving promise - Resume happens automatically when database wakes execution diff --git a/docs/content/crafting-tasks/retries-and-steps.md b/docs/content/crafting-tasks/retries-and-steps.md index 31c44e4..cf36889 100644 --- a/docs/content/crafting-tasks/retries-and-steps.md +++ b/docs/content/crafting-tasks/retries-and-steps.md @@ -85,6 +85,8 @@ Steps are therefore at-least-once around their callback. For side effects outsid The handler always runs from the start: on every retry and every resume after `ctx.sleep()` or a wait, code outside steps runs again. +Steps can run concurrently, for example with `Promise.all()`. When one of them throws or the handler hangs up in another (`ctx.sleep()`, a wait, or `orchestrator.stop()`), the attempt ends only after the callbacks still running return and save their results, or once `timeoutMs` elapses. A timeout or cancellation ends the attempt at once. + ## Working with Loops Loops without steps re-process all items on retry: diff --git a/packages/pgconductor-js/src/lib/map-concurrent.ts b/packages/pgconductor-js/src/lib/map-concurrent.ts index bebeb2d..662e9a7 100644 --- a/packages/pgconductor-js/src/lib/map-concurrent.ts +++ b/packages/pgconductor-js/src/lib/map-concurrent.ts @@ -89,5 +89,7 @@ export async function* mapConcurrent( } } finally { itemWait?.cancel(); + // Also on an error or an early exit, the consumer continues once every mapper settled + await Promise.allSettled(Array.from(active.values(), (task) => task.promise)); } } diff --git a/packages/pgconductor-js/src/task-context.ts b/packages/pgconductor-js/src/task-context.ts index 68a9ccb..1fd57ad 100644 --- a/packages/pgconductor-js/src/task-context.ts +++ b/packages/pgconductor-js/src/task-context.ts @@ -133,6 +133,8 @@ export type TaskContextOptions = { window?: [string, string]; telemetry: Telemetry; signals: SignalInbox; + /** Step callbacks and their saves in flight */ + steps: Set>; }; /** A signal sent to an execution with `conductor.signal()` or `signalWithStart()`. */ @@ -243,18 +245,23 @@ export class TaskContext< "pgconductor.queue": this.opts.execution.queue, }, run: async () => { - const result = await fn(); - const saved = await this.opts.db.saveStep( - { - executionId: this.opts.execution.id, - queue: this.opts.execution.queue, - orchestratorId: this.opts.execution.locked_by, - claimToken: this.opts.execution.claim_token, - key: name, - result: { result: result as JsonValue }, - }, - { signal: this.signal }, - ); + const running = (async () => { + const result = await fn(); + const saved = await this.opts.db.saveStep( + { + executionId: this.opts.execution.id, + queue: this.opts.execution.queue, + orchestratorId: this.opts.execution.locked_by, + claimToken: this.opts.execution.claim_token, + key: name, + result: { result: result as JsonValue }, + }, + { signal: this.signal }, + ); + return { result, saved }; + })(); + this.opts.steps.add(running); + const { result, saved } = await running.finally(() => this.opts.steps.delete(running)); return saved ? result : this.abortAndHangup({ reason: "suspended" }); }, }); diff --git a/packages/pgconductor-js/src/worker.ts b/packages/pgconductor-js/src/worker.ts index 6693fc8..042730e 100644 --- a/packages/pgconductor-js/src/worker.ts +++ b/packages/pgconductor-js/src/worker.ts @@ -543,9 +543,14 @@ export class Worker< ): Promise { const taskAbortController = new TypedAbortController(); const signal = AbortSignal.any([taskAbortController.signal, this.signal]); + const timedOut = new Deferred(); const timeout = task.timeoutMs - ? setTimeout(() => taskAbortController.abort({ reason: "timed-out" }), task.timeoutMs) + ? setTimeout(() => { + timedOut.resolve(); + taskAbortController.abort({ reason: "timed-out" }); + }, task.timeoutMs) : undefined; + const steps = new Set>(); // Wrapped, so no handler output can be mistaken for an abort const aborted = new Promise<{ abort: TaskAbortReasons }>((resolve) => { @@ -610,6 +615,7 @@ export class Worker< window: task.window, telemetry: this.telemetry, signals, + steps, }, extraContext, ); @@ -739,6 +745,12 @@ export class Worker< error: coerceError(err).message, } as const; } finally { + // Steps still running when a sibling hung up or threw end with the attempt, unless the + // worker aborted it. They save their results before the attempt settles. + const reason = taskAbortController.signal.reason?.reason; + if (reason !== "timed-out" && reason !== "cancelled" && reason !== "signal-failed") { + await Promise.race([Promise.allSettled(steps), timedOut.promise]); + } clearTimeout(timeout); this._running.delete(exec.id); // A hung-up or failed attempt leaves undelivered signals for the next one diff --git a/packages/pgconductor-js/tests/integration/worker-lifecycle.test.ts b/packages/pgconductor-js/tests/integration/worker-lifecycle.test.ts index c4e3a7d..54097da 100644 --- a/packages/pgconductor-js/tests/integration/worker-lifecycle.test.ts +++ b/packages/pgconductor-js/tests/integration/worker-lifecycle.test.ts @@ -6,6 +6,7 @@ import { Conductor } from "../../src/conductor"; import type { JsonValue } from "../../src/database-client"; import { Orchestrator } from "../../src/orchestrator"; import { defineTask } from "../../src/task-definition"; +import type { TaskContext } from "../../src/task-context"; import { TaskSchemas } from "../../src/schemas"; import { Deferred } from "../../src/lib/deferred"; import { TestDatabasePool, type TestDatabase } from "../fixtures/test-database"; @@ -387,6 +388,149 @@ describe("drain()", () => { }); }); +describe("a step callback still running when a sibling ends the attempt", () => { + async function setup( + handler: (ctx: TaskContext, blocked: () => Promise) => Promise, + { + sql = (s: Sql) => s, + timeoutMs, + running = new Deferred(), + }: { sql?: (s: Sql) => Sql; timeoutMs?: number; running?: Deferred } = {}, + ) { + const db = await child(); + const conductor = Conductor.create({ + sql: sql(db.sql), + tasks: TaskSchemas.fromSchema([sessionDefinition]), + context: {}, + }); + const aborted = new Deferred(); + const gate = new Deferred(); + const task = conductor.createTask( + { name: "session", maxAttempts: 1, timeoutMs }, + { invocable: true }, + async (_e, ctx) => { + ctx.signal.addEventListener("abort", () => aborted.resolve()); + await handler(ctx, () => + ctx.step("blocked", async () => { + running.resolve(); + await gate.promise; + return "done"; + }), + ); + }, + ); + await conductor.ensureInstalled(); + const orchestrator = Orchestrator.create({ + conductor, + tasks: [task], + defaultWorker: { pollIntervalMs: 50, flushIntervalMs: 50 }, + }); + const id = await conductor.invoke({ name: "session" }, {}); + await orchestrator.start(); + await running.promise; + const steps = async () => + ( + await db.sql<{ key: string }[]>`select key from pgconductor._private_steps order by key` + ).map((r) => r.key); + return { db, id, orchestrator, gate, aborted, steps }; + } + + test("stop() waits for it when the sibling's checkpoint releases the attempt", async () => { + const release = new Deferred(); + const delivered = new Deferred(); + const { db, id, orchestrator, gate, aborted, steps } = await setup( + (ctx, blocked) => Promise.allSettled([ctx.step("sibling", () => "done"), blocked()]), + { sql: (sql) => delayLoadStep(sql, "sibling", release.promise, delivered) }, + ); + try { + const stopped = orchestrator.stop(); + await aborted.promise; + release.resolve(); + await delivered.promise; + expect(await within(stopped, 300)).toBe("timeout"); + gate.resolve(); + await stopped; + expect(await row(db, id)).toMatchObject({ + completed: false, + failed: false, + locked: false, + attempts: 0, + }); + expect(await steps()).toEqual(["blocked"]); + } finally { + release.resolve(); + gate.resolve(); + await orchestrator.stop(); + } + }); + + test("a sibling sleep settles the attempt once it returns", async () => { + const running = new Deferred(); + const delivered = new Deferred(); + const { db, id, orchestrator, gate, steps } = await setup( + (ctx, blocked) => Promise.allSettled([ctx.sleep("nap", 60_000), blocked()]), + { running, sql: (sql) => delayLoadStep(sql, "nap", running.promise, delivered) }, + ); + try { + await delivered.promise; + await sleep(300); + expect(await row(db, id)).toMatchObject({ locked: true }); + gate.resolve(); + await waitForCondition(async () => (await row(db, id))?.locked === false); + expect(await row(db, id)).toMatchObject({ completed: false, failed: false }); + expect(await steps()).toEqual(["blocked", "nap"]); + } finally { + gate.resolve(); + await orchestrator.stop(); + } + }); + + test("a sibling that throws fails the attempt once it returns", async () => { + const running = new Deferred(); + const { db, id, orchestrator, gate, steps } = await setup( + (ctx, blocked) => + Promise.all([ + blocked(), + ctx.step("sibling", async () => { + await running.promise; + throw new Error("sibling failed"); + }), + ]), + { running }, + ); + try { + await sleep(300); + expect(await row(db, id)).toMatchObject({ locked: true }); + gate.resolve(); + await waitForCondition(async () => (await row(db, id))?.locked === false); + expect(await row(db, id)).toMatchObject({ failed: true, last_error: "sibling failed" }); + expect(await steps()).toEqual(["blocked"]); + } finally { + gate.resolve(); + await orchestrator.stop(); + } + }); + + test("timeoutMs bounds the wait", async () => { + const running = new Deferred(); + const { db, id, orchestrator, gate } = await setup( + (ctx, blocked) => Promise.allSettled([ctx.sleep("nap", 60_000), blocked()]), + { + timeoutMs: 500, + running, + sql: (sql) => delayLoadStep(sql, "nap", running.promise, new Deferred()), + }, + ); + try { + await waitForCondition(async () => (await row(db, id))?.locked === false, 5000); + expect(await row(db, id)).toMatchObject({ completed: false, failed: false }); + } finally { + gate.resolve(); + await orchestrator.stop(); + } + }); +}); + describe("stop() while a handler runs a step", () => { const ackStatement = "set delivered_at = pgconductor._private_current_time()"; diff --git a/packages/pgconductor-js/tests/unit/lib/map-concurrent.test.ts b/packages/pgconductor-js/tests/unit/lib/map-concurrent.test.ts index 8608ed9..f8f0e96 100644 --- a/packages/pgconductor-js/tests/unit/lib/map-concurrent.test.ts +++ b/packages/pgconductor-js/tests/unit/lib/map-concurrent.test.ts @@ -4,6 +4,7 @@ import { mapConcurrent } from "../../../src/lib/map-concurrent"; import { AsyncQueue, type PollableAsyncIterable } from "../../../src/lib/async-queue"; import { BatchingAsyncQueue } from "../../../src/lib/batching-async-queue"; import { Deferred } from "../../../src/lib/deferred"; +import { within } from "../../test-utils"; class PollableGenerator implements PollableAsyncIterable { private buffer: T[] = []; @@ -201,6 +202,52 @@ describe("mapConcurrent", () => { } }); + test("waits for running mappers before rethrowing a mapper's error", async () => { + const source = new AsyncQueue(2); + await source.push(0); + await source.push(1); + source.close(); + const gate = new Deferred(); + let finished = false; + const consuming = (async () => { + for await (const _ of mapConcurrent(source, 2, async (n) => { + if (n === 0) throw new Error("test error"); + await gate.promise; + finished = true; + })) { + // Should throw before yielding + } + })(); + + expect(await within(consuming, 20)).toBe("timeout"); + gate.resolve(); + await expect(consuming).rejects.toThrow("test error"); + expect(finished).toBe(true); + }); + + test("waits for running mappers when the consumer exits early", async () => { + const source = new AsyncQueue(2); + await source.push(0); + await source.push(1); + source.close(); + const gate = new Deferred(); + let finished = false; + const consuming = (async () => { + for await (const _ of mapConcurrent(source, 2, async (n) => { + if (n === 0) return; + await gate.promise; + finished = true; + })) { + break; + } + })(); + + expect(await within(consuming, 20)).toBe("timeout"); + gate.resolve(); + await consuming; + expect(finished).toBe(true); + }); + test("handles single concurrency", async () => { const order: number[] = []; const source = pollable(generateNumbers(5));