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. 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 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); + } +});