From 1233911636acfb96618ba93fa048870279e47069 Mon Sep 17 00:00:00 2001 From: Agent Date: Thu, 22 Oct 2026 19:24:26 +0200 Subject: [PATCH 1/3] fix(worker): preserve outcomes across result reporting failures Telemetry or completion RPC errors previously entered the handler-failure path. Separate execution from one-shot settlement, surface logging and acknowledgement errors through the worker boundary, and classify interruption from the shutdown signal instead of the exception name. Use nullish output handling and SQL coalesce because postgres.js binds sql.json(null) as SQL NULL. Preserve valid falsy values and normalize nullish results to JSON null, with actual-storage integration coverage. Affected unit, type, lint, export, integration, Edge/Node/Bun runtime, client E2E, and CLI installer checks pass. Both Sol review axes report no findings. The full core pgTAP gate remains open: six existing input-assembly timing assertions fail in the full run; the single focused reproduction passes all eleven assertions. No performance thresholds or schema code changed. Refs #701 --- pkgs/core/src/PgflowSqlClient.ts | 3 +- pkgs/edge-worker/src/flow/StepTaskExecutor.ts | 87 +++-- .../integration/pgflowSqlClientOutput.test.ts | 48 +++ .../tests/unit/StepTaskExecutor.test.ts | 321 ++++++++++++++++++ 4 files changed, 424 insertions(+), 35 deletions(-) create mode 100644 pkgs/edge-worker/tests/integration/pgflowSqlClientOutput.test.ts create mode 100644 pkgs/edge-worker/tests/unit/StepTaskExecutor.test.ts diff --git a/pkgs/core/src/PgflowSqlClient.ts b/pkgs/core/src/PgflowSqlClient.ts index 5587c9e5b..7fac2ba05 100644 --- a/pkgs/core/src/PgflowSqlClient.ts +++ b/pkgs/core/src/PgflowSqlClient.ts @@ -56,12 +56,13 @@ export class PgflowSqlClient } async completeTask(stepTask: StepTaskKey, output?: Json): Promise { + // postgres.js binds sql.json(null) as SQL NULL; coalesce to JSON null. await this.sql` SELECT pgflow.complete_task( run_id => ${stepTask.run_id}::uuid, step_slug => ${stepTask.step_slug}::text, task_index => ${stepTask.task_index}::int, - output => ${this.sql.json(output || null)}::jsonb + output => coalesce(${this.sql.json(output ?? null)}::jsonb, 'null'::jsonb) ); `; } diff --git a/pkgs/edge-worker/src/flow/StepTaskExecutor.ts b/pkgs/edge-worker/src/flow/StepTaskExecutor.ts index 18e113f66..64be16a63 100644 --- a/pkgs/edge-worker/src/flow/StepTaskExecutor.ts +++ b/pkgs/edge-worker/src/flow/StepTaskExecutor.ts @@ -1,17 +1,10 @@ import type { AnyFlow } from '@pgflow/dsl'; -import type { StepTaskRecord } from '@pgflow/core'; +import type { Json, StepTaskRecord } from '@pgflow/core'; import type { IPgflowClient } from '@pgflow/core'; import type { IExecutor } from '../core/types.js'; import type { Logger, TaskLogContext } from '../platform/types.js'; import type { StepTaskHandlerContext } from '../core/context.js'; -class AbortError extends Error { - constructor() { - super('Operation aborted'); - this.name = 'AbortError'; - } -} - /** * Configuration for worker identity in logging */ @@ -81,21 +74,23 @@ export class StepTaskExecutor { + if (this.signal.aborted) { + this.logger.debug(`Aborted execution for step task ${this.msgId}`); + return; + } + const logContext = this.buildLogContext(); + // Telemetry before execution is not a handler failure. + this.logger.taskStarted(logContext); + const startTime = Date.now(); + let result: Json; try { - if (this.signal.aborted) { - throw new AbortError(); - } - - // Check if already aborted before starting + // Logging may have triggered shutdown before the handler starts. this.signal.throwIfAborted(); const stepSlug = this.stepTask.step_slug; - // Log task started at debug level - this.logger.taskStarted(logContext); - // Get the step handler from the flow with proper typing const stepDef = this.flow.getStepDefinition(stepSlug); @@ -103,9 +98,6 @@ export class StepTaskExecutor this.logger.taskCompleted(logContext, durationMs), + () => this.adapter.completeTask(this.stepTask, result) + ); + } - // Log task completed at verbose level - this.logger.taskCompleted(logContext, durationMs); - await this.adapter.completeTask(this.stepTask, result); + /** Report once even if telemetry throws, then surface both errors if needed. */ + private async reportResult(log: () => void, report: () => Promise) { + const errors: unknown[] = []; + try { + log(); } catch (error) { - await this.handleExecutionError(error, logContext); + errors.push(error); + } + try { + await report(); + } catch (error) { + errors.push(error); + } + if (errors.length === 1) { + throw errors[0]; + } + if (errors.length > 1) { + throw new AggregateError(errors, 'Task result logging and reporting failed'); } } /** * Handles the error that occurred during execution. * - * If the error is an AbortError, it means that the worker was aborted and stopping, - * the task will be picked up by another worker later. - * - * Otherwise, it marks the task as failed. + * Only the worker shutdown signal classifies an execution as interrupted. + * An unrelated AbortError from a handler is still a handler failure. */ private async handleExecutionError(error: unknown, logContext: TaskLogContext) { - if (error instanceof Error && error.name === 'AbortError') { + if (this.signal.aborted) { this.logger.debug(`Aborted execution for step task ${this.msgId}`); // Do not mark as failed - the worker was aborted and stopping, // the task will be picked up by another worker later } else { - // Log task failed at verbose level using structured method - const errorObj = error instanceof Error ? error : new Error(String(error)); - this.logger.taskFailed(logContext, errorObj); - await this.adapter.failTask(this.stepTask, error); + await this.reportResult( + () => { + const errorObj = error instanceof Error ? error : new Error(String(error)); + this.logger.taskFailed(logContext, errorObj); + }, + () => this.adapter.failTask(this.stepTask, error) + ); } } } diff --git a/pkgs/edge-worker/tests/integration/pgflowSqlClientOutput.test.ts b/pkgs/edge-worker/tests/integration/pgflowSqlClientOutput.test.ts new file mode 100644 index 000000000..3da2b15cb --- /dev/null +++ b/pkgs/edge-worker/tests/integration/pgflowSqlClientOutput.test.ts @@ -0,0 +1,48 @@ +import { assertEquals } from '@std/assert'; +import { PgflowSqlClient } from '@pgflow/core'; +import type { Json } from '@pgflow/core'; +import { withTransaction } from '../db.ts'; + +const cases: Array<{ name: string; output: Json | undefined; json: string }> = [ + { name: 'zero', output: 0, json: '0' }, + { name: 'false', output: false, json: 'false' }, + { name: 'empty string', output: '', json: '""' }, + { name: 'JSON null', output: null, json: 'null' }, + { name: 'undefined as JSON null', output: undefined, json: 'null' }, +]; + +for (const { name, output, json } of cases) { + Deno.test( + `PgflowSqlClient.completeTask stores ${name} in PostgreSQL`, + withTransaction(async (sql) => { + const flowSlug = 'sql_client_output'; + await sql`select pgflow.create_flow(${flowSlug})`; + await sql`select pgflow.add_step(${flowSlug}, 'work')`; + const [worker] = await sql<{ worker_id: string }[]>` + insert into pgflow.workers (worker_id, queue_name, function_name) + values (gen_random_uuid(), ${flowSlug}, 'output_test') + returning worker_id + `; + const client = new PgflowSqlClient(sql); + const run = await client.startFlow(flowSlug, {}); + const messages = await client.readMessages(flowSlug, 10, 1, 1, 10); + assertEquals(messages.length, 1); + const tasks = await client.startTasks( + flowSlug, + messages.map((message) => message.msg_id), + worker.worker_id, + flowSlug + ); + assertEquals(tasks.length, 1); + + await client.completeTask(tasks[0], output); + + const [stored] = await sql` + select status, output::text as json, output is null as sql_null + from pgflow.step_tasks + where run_id = ${run.run_id}::uuid and step_slug = 'work' and task_index = 0 + `; + assertEquals(stored, { status: 'completed', json, sql_null: false }); + }) + ); +} diff --git a/pkgs/edge-worker/tests/unit/StepTaskExecutor.test.ts b/pkgs/edge-worker/tests/unit/StepTaskExecutor.test.ts new file mode 100644 index 000000000..f857ce376 --- /dev/null +++ b/pkgs/edge-worker/tests/unit/StepTaskExecutor.test.ts @@ -0,0 +1,321 @@ +import { assertEquals, assertRejects, assertStrictEquals } from '@std/assert'; +import { Flow } from '@pgflow/dsl'; +import type { IPgflowClient, Json, StepTaskKey, StepTaskRecord } from '@pgflow/core'; +import { StepTaskExecutor } from '../../src/flow/StepTaskExecutor.ts'; +import { ExecutionController } from '../../src/core/ExecutionController.ts'; +import type { Logger } from '../../src/platform/types.ts'; +import { fakeLogger } from '../fakes.ts'; + +function setup( + handler: () => Json | Promise, + options: { + controller?: AbortController; + logger?: Partial; + completeTask?: IPgflowClient['completeTask']; + failTask?: IPgflowClient['failTask']; + } = {} +) { + const controller = options.controller ?? new AbortController(); + let handlerCalls = 0; + const flow = new Flow({ slug: 'result_boundary' }).step({ slug: 'work' }, () => { + handlerCalls++; + return handler(); + }); + const task: StepTaskRecord = { + flow_slug: flow.slug, + step_slug: 'work', + run_id: 'test-run', + task_index: 0, + msg_id: '1', + input: {}, + flow_input: {}, + }; + const completions: Array<{ task: StepTaskKey; output: Json | undefined }> = []; + const failures: Array<{ task: StepTaskKey; error: unknown }> = []; + const adapter: IPgflowClient = { + startFlow: () => { throw new Error('Unexpected startFlow'); }, + readMessages: () => { throw new Error('Unexpected readMessages'); }, + startTasks: () => { throw new Error('Unexpected startTasks'); }, + completeTask: async (task, output) => { + completions.push({ task, output }); + await options.completeTask?.(task, output); + }, + failTask: async (task, error) => { + failures.push({ task, error }); + await options.failTask?.(task, error); + }, + }; + const executor = new StepTaskExecutor( + flow, + adapter, + controller.signal, + { ...fakeLogger, ...options.logger }, + { + env: {}, + shutdownSignal: controller.signal, + stepTask: task, + rawMessage: { + msg_id: '1', + read_ct: 1, + enqueued_at: '2026-01-01T00:00:00Z', + vt: '2026-01-01T00:01:00Z', + message: {}, + }, + workerConfig: { + maxConcurrent: 1, + maxPollSeconds: 1, + pollIntervalMs: 100, + batchSize: 1, + visibilityTimeout: 2, + }, + flowInput: Promise.resolve({}), + }, + { workerId: 'worker', workerName: 'test', queueName: flow.slug } + ); + return { executor, task, completions, failures, get handlerCalls() { return handlerCalls; } }; +} + +Deno.test('StepTaskExecutor - completion rejection surfaces without reporting handler failure or replay', async () => { + const error = new Error('Completion acknowledgement unknown'); + const test = setup(() => ({ value: 42 }), { + completeTask: () => Promise.reject(error), + }); + + const caught = await assertRejects(() => test.executor.execute()); + + assertStrictEquals(caught, error); + assertEquals(test.completions, [{ task: test.task, output: { value: 42 } }]); + assertEquals(test.failures, []); + assertEquals(test.handlerCalls, 1); +}); + +Deno.test('StepTaskExecutor - completion logging failure surfaces after the original result is reported', async () => { + const error = new Error('Completion logger unavailable'); + const output = { value: 42 }; + const test = setup(() => output, { + logger: { taskCompleted: () => { throw error; } }, + }); + + const caught = await assertRejects(() => test.executor.execute()); + + assertStrictEquals(caught, error); + assertEquals(test.completions, [{ task: test.task, output }]); + assertStrictEquals(test.completions[0].output, output); + assertEquals(test.failures, []); + assertEquals(test.handlerCalls, 1); +}); + +Deno.test('StepTaskExecutor - failure logging error surfaces after the original handler error is reported', async () => { + const handlerError = new Error('Handler failed'); + const logError = new Error('Failure logger unavailable'); + const test = setup(() => { throw handlerError; }, { + logger: { taskFailed: () => { throw logError; } }, + }); + + const caught = await assertRejects(() => test.executor.execute()); + + assertStrictEquals(caught, logError); + assertEquals(test.failures, [{ task: test.task, error: handlerError }]); + assertStrictEquals(test.failures[0].error, handlerError); + assertEquals(test.completions, []); + assertEquals(test.handlerCalls, 1); +}); + +Deno.test('StepTaskExecutor - pre-handler logging failure does not run or fail the handler', async () => { + const error = new Error('Start logger unavailable'); + const test = setup(() => 42, { + logger: { taskStarted: () => { throw error; } }, + }); + + const caught = await assertRejects(() => test.executor.execute()); + + assertStrictEquals(caught, error); + assertEquals(test.handlerCalls, 0); + assertEquals(test.completions, []); + assertEquals(test.failures, []); +}); + +Deno.test('StepTaskExecutor - unrelated AbortError with a live shutdown signal is a handler failure', async () => { + const error = new DOMException('Request timed out', 'AbortError'); + const test = setup(() => { throw error; }); + + await test.executor.execute(); + + assertEquals(test.failures, [{ task: test.task, error }]); + assertStrictEquals(test.failures[0].error, error); + assertEquals(test.completions, []); + assertEquals(test.handlerCalls, 1); +}); + +Deno.test('StepTaskExecutor - already-aborted execution does not start or report a handler failure', async () => { + const controller = new AbortController(); + controller.abort('Worker stopped'); + let started = false; + const test = setup(() => 42, { + controller, + logger: { taskStarted: () => { started = true; } }, + }); + + await test.executor.execute(); + + assertEquals(started, false); + assertEquals(test.handlerCalls, 0); + assertEquals(test.completions, []); + assertEquals(test.failures, []); +}); + +Deno.test('StepTaskExecutor - failure acknowledgement rejection does not report twice or replay the handler', async () => { + const handlerError = new Error('Handler failed'); + const reportError = new Error('Failure acknowledgement unknown'); + const test = setup(() => { throw handlerError; }, { + failTask: () => Promise.reject(reportError), + }); + + const caught = await assertRejects(() => test.executor.execute()); + + assertStrictEquals(caught, reportError); + assertEquals(test.failures, [{ task: test.task, error: handlerError }]); + assertStrictEquals(test.failures[0].error, handlerError); + assertEquals(test.completions, []); + assertEquals(test.handlerCalls, 1); +}); + +Deno.test('StepTaskExecutor - actual shutdown with arbitrary thrown reasons does not report business failure', async () => { + for (const reason of [new Error('Connection closed'), 'Worker stopped', undefined]) { + const controller = new AbortController(); + const test = setup(() => { + controller.abort('Shutdown requested'); + throw reason; + }, { controller }); + + await test.executor.execute(); + + assertEquals(test.handlerCalls, 1); + assertEquals(test.completions, []); + assertEquals(test.failures, []); + } +}); + +Deno.test('StepTaskExecutor - shutdown during start logging does not start or fail a handler', async () => { + const controller = new AbortController(); + const test = setup(() => 42, { + controller, + logger: { taskStarted: () => controller.abort('Shutdown requested') }, + }); + + await test.executor.execute(); + + assertEquals(test.handlerCalls, 0); + assertEquals(test.completions, []); + assertEquals(test.failures, []); +}); + +Deno.test('StepTaskExecutor - success still reports once when shutdown starts after the handler returns', async () => { + const controller = new AbortController(); + const output = { value: 42 }; + const test = setup(() => output, { + controller, + logger: { taskCompleted: () => controller.abort('Shutdown requested') }, + }); + + await test.executor.execute(); + + assertEquals(controller.signal.aborted, true); + assertEquals(test.completions, [{ task: test.task, output }]); + assertEquals(test.failures, []); + assertEquals(test.handlerCalls, 1); +}); + +Deno.test('StepTaskExecutor - acknowledgement rejection during shutdown retains handler success', async () => { + const controller = new AbortController(); + const error = new Error('Connection closed during shutdown'); + const test = setup(() => { + controller.abort('Shutdown requested'); + return false; + }, { + controller, + completeTask: () => Promise.reject(error), + }); + + const caught = await assertRejects(() => test.executor.execute()); + + assertStrictEquals(caught, error); + assertEquals(test.completions, [{ task: test.task, output: false }]); + assertEquals(test.failures, []); + assertEquals(test.handlerCalls, 1); +}); + +Deno.test('StepTaskExecutor - serialization failure after success never becomes handler failure', async () => { + const output: { self?: Json } = {}; + output.self = output; + const test = setup(() => output, { + completeTask: (_task, result) => { + JSON.stringify(result); + return Promise.resolve(); + }, + }); + + await assertRejects(() => test.executor.execute(), TypeError); + + assertEquals(test.completions.length, 1); + assertStrictEquals(test.completions[0].output, output); + assertEquals(test.failures, []); + assertEquals(test.handlerCalls, 1); +}); + +Deno.test('StepTaskExecutor - telemetry and reporting errors both reach the existing worker error boundary', async () => { + for (const outcome of ['success', 'failure'] as const) { + const controller = new AbortController(); + const handlerError = new Error('Handler failed'); + const logError = new Error('Telemetry unavailable'); + const reportError = new Error('Acknowledgement unknown'); + const test = setup(() => { + if (outcome === 'failure') throw handlerError; + return 42; + }, { + controller, + logger: { + taskCompleted: () => { throw logError; }, + taskFailed: () => { throw logError; }, + }, + completeTask: () => Promise.reject(reportError), + failTask: () => Promise.reject(reportError), + }); + const observed: unknown[] = []; + const executions = new ExecutionController( + () => test.executor, + controller.signal, + { maxConcurrent: 1 }, + { ...fakeLogger, error: (_message, error) => { observed.push(error); } } + ); + + const caught = await assertRejects( + () => executions.start({ msg_id: '1' }), + AggregateError + ); + await executions.awaitCompletion(); + + assertEquals(caught.errors.length, 2); + assertStrictEquals(caught.errors[0], logError); + assertStrictEquals(caught.errors[1], reportError); + assertStrictEquals(observed[0], caught); + assertEquals(observed.length, 1); + assertEquals(test.handlerCalls, 1); + assertEquals(test.completions.length, outcome === 'success' ? 1 : 0); + assertEquals(test.failures.length, outcome === 'failure' ? 1 : 0); + if (outcome === 'failure') assertStrictEquals(test.failures[0].error, handlerError); + } +}); + +Deno.test('StepTaskExecutor - live handler failures report the original thrown value once', async () => { + for (const error of [new Error('Handler failed'), 'Bad input', undefined]) { + const test = setup(() => { throw error; }); + + await test.executor.execute(); + + assertEquals(test.failures, [{ task: test.task, error }]); + assertStrictEquals(test.failures[0].error, error); + assertEquals(test.completions, []); + assertEquals(test.handlerCalls, 1); + } +}); From ad67fd95618a0ae385e061e2e634043441f1e77d Mon Sep 17 00:00:00 2001 From: Agent Date: Fri, 23 Oct 2026 13:49:54 +0200 Subject: [PATCH 2/3] chore: add worker result boundary patch changeset Record patch releases for the core SQL client and shared worker fixes in #701. Keep the fixed package group in sync and describe the falsy-output, shutdown, and one-shot settlement behavior in release notes. --- .changeset/worker-result-boundaries.md | 8 ++++++++ 1 file changed, 8 insertions(+) create mode 100644 .changeset/worker-result-boundaries.md diff --git a/.changeset/worker-result-boundaries.md b/.changeset/worker-result-boundaries.md new file mode 100644 index 000000000..a7c486404 --- /dev/null +++ b/.changeset/worker-result-boundaries.md @@ -0,0 +1,8 @@ +--- +"@pgflow/core": patch +"@pgflow/edge-worker": patch +--- + +Preserve handler outcomes when worker logging or task settlement fails, and classify interruption from the worker shutdown signal instead of the exception name. Successful handlers get one completion attempt even during shutdown; settlement errors reach the worker error boundary without replay or conversion into handler failures. + +Preserve falsy task outputs (`0`, `false`, and `""`) and store null or undefined outputs as JSON null instead of SQL NULL. From a7ec425b8ff2c9b568638b05d405898ecb880b90 Mon Sep 17 00:00:00 2001 From: Agent Date: Fri, 23 Oct 2026 14:27:57 +0200 Subject: [PATCH 3/3] style(core): format SQL arrays for Sqruff 0.43 CI installs Sqruff 0.43.0, which rejects whitespace before array brackets in four existing schema files. Run pnpm nx fix-sql core with the CI version to clear the five LT01 violations without changing SQL behavior. Checked pnpm nx lint:sqruff core with Sqruff 0.43.0 and git diff --ignore-all-space --exit-code; both pass. --- pkgs/core/schemas/0100_function_add_step.sql | 2 +- pkgs/core/schemas/0100_function_compare_flow_shapes.sql | 2 +- pkgs/core/schemas/0110_function_set_vt_batch.sql | 4 ++-- pkgs/core/schemas/0120_function_start_tasks.sql | 2 +- 4 files changed, 5 insertions(+), 5 deletions(-) diff --git a/pkgs/core/schemas/0100_function_add_step.sql b/pkgs/core/schemas/0100_function_add_step.sql index cb493611e..38311a957 100644 --- a/pkgs/core/schemas/0100_function_add_step.sql +++ b/pkgs/core/schemas/0100_function_add_step.sql @@ -1,7 +1,7 @@ create or replace function pgflow.add_step( flow_slug text, step_slug text, - deps_slugs text [] default '{}', + deps_slugs text[] default '{}', max_attempts int default null, base_delay int default null, timeout int default null, diff --git a/pkgs/core/schemas/0100_function_compare_flow_shapes.sql b/pkgs/core/schemas/0100_function_compare_flow_shapes.sql index 2f97f16d5..657221959 100644 --- a/pkgs/core/schemas/0100_function_compare_flow_shapes.sql +++ b/pkgs/core/schemas/0100_function_compare_flow_shapes.sql @@ -4,7 +4,7 @@ create or replace function pgflow._compare_flow_shapes( p_local jsonb, p_db jsonb ) -returns text [] +returns text[] language plpgsql stable set search_path to '' diff --git a/pkgs/core/schemas/0110_function_set_vt_batch.sql b/pkgs/core/schemas/0110_function_set_vt_batch.sql index 1e11c08ed..a356d8d1c 100644 --- a/pkgs/core/schemas/0110_function_set_vt_batch.sql +++ b/pkgs/core/schemas/0110_function_set_vt_batch.sql @@ -12,8 +12,8 @@ create or replace function pgflow.set_vt_batch( queue_name TEXT, - msg_ids BIGINT [], - vt_offsets INTEGER [] + msg_ids BIGINT[], + vt_offsets INTEGER[] ) returns table ( msg_id BIGINT, diff --git a/pkgs/core/schemas/0120_function_start_tasks.sql b/pkgs/core/schemas/0120_function_start_tasks.sql index 0fdd40691..315f04fa8 100644 --- a/pkgs/core/schemas/0120_function_start_tasks.sql +++ b/pkgs/core/schemas/0120_function_start_tasks.sql @@ -25,7 +25,7 @@ -- message is touched. create or replace function pgflow.start_tasks( flow_slug text, - msg_ids bigint [], + msg_ids bigint[], worker_id uuid, queue_name text, step_slug text default null