From 26c18d55b93968a48bccdc1cda61e9ce5a6cd4f5 Mon Sep 17 00:00:00 2001 From: psteinroe Date: Tue, 20 Oct 2026 12:57:12 +0000 Subject: [PATCH 1/2] feat(tasks): validate task payloads and return values at runtime --- AGENTS.md | 1 + docs/content/crafting-tasks/defining-tasks.md | 18 +- docs/content/task-execution/batching.md | 2 +- .../task-execution/dead-letter-queue.md | 2 +- packages/pgconductor-js/src/conductor.ts | 6 + .../pgconductor-js/src/lib/validate-schema.ts | 32 +- packages/pgconductor-js/src/task.ts | 9 + packages/pgconductor-js/src/worker.ts | 105 +++++- .../tests/integration/task-validation.test.ts | 314 ++++++++++++++++++ 9 files changed, 458 insertions(+), 31 deletions(-) create mode 100644 packages/pgconductor-js/tests/integration/task-validation.test.ts diff --git a/AGENTS.md b/AGENTS.md index 2797838a..a16f527e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -140,6 +140,7 @@ Tables live in the `pgconductor` schema with a `_private_` prefix (`_private_exe - Claiming returns each execution's undelivered signals, which prime its `SignalInbox` before the handler runs - The claim statement also reads the cancellation flag and newer undelivered signals of the worker's running executions (`running` in `getExecutions`), aborts cancelled ones and hands signals to their `SignalInbox`. While all slots are busy the worker still runs it every `pollIntervalMs`, with a batch size of 0, as long as an execution runs. Batch members are polled too and share the batch's controller; cancelling one releases the others without spending their attempt. After a drain's last empty claim the worker keeps these zero-sized polls until its pipeline ends, then joins the fetch loop before resetting - Handles graceful shutdown via AbortController; a stop during registration retries ends them and settles `stopped` +- Validates the payload of every invocation (any origin, including SQL and dead letters) and the return value against the task's Standard Schemas (`Task.schemas`, resolved from `TaskSchemas` in `createTask`) and passes their output on. Issues fail the execution as `permanently_failed`; a throwing validator fails the attempt. Batches validate per member **TaskContext** (`task-context.ts`) - Provides API to task functions: `step()`, `sleep()`, `invoke()`, `invokeAndWait()`, `waitForResult()`, `waitForEvent()`, `subscribe()`, `waitForAny()`, `onSignal()`, `sendSignal()`, `signalWithStart()`, plus `executionId` and `queue` diff --git a/docs/content/crafting-tasks/defining-tasks.md b/docs/content/crafting-tasks/defining-tasks.md index 4c4db0f9..bf913664 100644 --- a/docs/content/crafting-tasks/defining-tasks.md +++ b/docs/content/crafting-tasks/defining-tasks.md @@ -8,7 +8,7 @@ Schema definitions give you: - **Type safety** - TypeScript infers payload and return types automatically - **Shared schemas** - One source of truth for worker and client -- **Runtime validation** - Optional validation with StandardSchema (Zod, Valibot, etc.) +- **Runtime validation** - Workers validate payloads and return values with StandardSchema (Zod, Valibot, etc.) - **Better DX** - Autocomplete for payloads and return values ## Task Definitions @@ -78,6 +78,20 @@ const conductor = Conductor.create({ }); ``` +## Runtime Validation + +Workers validate every execution of a task defined with `defineTask()` against its schemas, whichever way it was invoked: `conductor.invoke()`, `ctx.invoke()`, `signalWithStart()`, `pgconductor.invoke()` from SQL, or a dead-letter delivery. + +- The payload is validated before middleware and the handler run. The handler receives the schema's output, so transforms, defaults and stripped keys apply. +- The return value is validated before the execution completes. The schema's output is stored as the result, serialized as JSON. +- A payload or return value that does not match fails the execution without retries, with the issues as its error, for example `Invalid payload for task "send-email": to: Invalid email address`. Like any permanent failure it fails waiting parents and is delivered to a dead-letter queue. +- A validator that throws instead of returning issues fails the attempt, which is retried. +- Async validators are awaited. +- Batch tasks validate each execution on its own: members with an invalid payload fail without reaching the handler, and the others run as the batch. Return values are validated per item. +- Invoking does not validate, so the stored payload is the one you passed. Cron and custom event triggers carry no task payload. + +Type-only definitions (`DefineTask` with `TaskSchemas.fromUnion()`) have no runtime schema and are never validated. + ## Event Definitions Events enable reactive, event-driven patterns. @@ -140,7 +154,7 @@ const task = conductor.createTask( { name: "process-data" }, { invocable: true }, async (event, ctx) => { - // Return value must match schema + // The return value is validated against the schema return { processed: event.payload.items.length, errors: [], diff --git a/docs/content/task-execution/batching.md b/docs/content/task-execution/batching.md index 7f2f51ca..cc4a8f12 100644 --- a/docs/content/task-execution/batching.md +++ b/docs/content/task-execution/batching.md @@ -73,7 +73,7 @@ const processBatch = conductor.createTask( - `size` - Maximum number of events to batch together - `timeoutMs` - Maximum time to wait for a full batch before processing what's available -Each batch event has `{ name, payload?, execution }`. The read-only `execution` object contains `id`, `queue`, `task_key`, and `locked_by`; the event payload remains unchanged. Batch handlers can use `event.execution.id` for their own idempotency keys. +Each batch event has `{ name, payload?, execution }`. The read-only `execution` object contains `id`, `queue`, `task_key`, and `locked_by`. With a task schema, payloads and return values are validated per execution: one with an invalid payload fails on its own and the batch runs without it (see [Runtime Validation](../crafting-tasks/defining-tasks.md#runtime-validation)). Batch handlers can use `event.execution.id` for their own idempotency keys. ### Batch Tasks with Returns diff --git a/docs/content/task-execution/dead-letter-queue.md b/docs/content/task-execution/dead-letter-queue.md index 24151e16..25695dd2 100644 --- a/docs/content/task-execution/dead-letter-queue.md +++ b/docs/content/task-execution/dead-letter-queue.md @@ -21,4 +21,4 @@ const chargeCard = conductor.createTask( The destination is a new execution with the original payload. Its execution row contains machine-readable source execution ID, source queue and task, final error, attempt count, and failure timestamp. The source remains a normal failed execution unless its retention policy removes it. -Retries and cancellation do not deliver to a dead-letter queue. Delivery is transactional, claim-fenced, and idempotent. A destination may have its own retry, retention, and concurrency settings. Chains are supported, but a task cannot target itself directly. The destination task must accept the source payload. +Retries and cancellation do not deliver to a dead-letter queue. Delivery is transactional, claim-fenced, and idempotent. A destination may have its own retry, retention, and concurrency settings. Chains are supported, but a task cannot target itself directly. The destination task must accept the source payload. A destination defined with a payload schema validates it like any execution, so a source that failed on an invalid payload also fails there; give such a destination a schema that accepts it, or a type-only definition. diff --git a/packages/pgconductor-js/src/conductor.ts b/packages/pgconductor-js/src/conductor.ts index 33db23c2..f505461e 100644 --- a/packages/pgconductor-js/src/conductor.ts +++ b/packages/pgconductor-js/src/conductor.ts @@ -277,6 +277,11 @@ export class Conductor< TaskContext & ExtraContext & MiddlewareContext, TaskEventFromTriggers, Events> > { + const queue = definition.queue || "default"; + const schemas = this.options.tasks?.definitions.find( + (task: TaskDefinition) => + task.name === definition.name && task.queue === queue, + ); return Task.create< TDef["name"], ResolvedQueue, @@ -298,6 +303,7 @@ export class Conductor< >, this.events, this.options.metadata, + schemas, ); } diff --git a/packages/pgconductor-js/src/lib/validate-schema.ts b/packages/pgconductor-js/src/lib/validate-schema.ts index a246a663..a0539bbb 100644 --- a/packages/pgconductor-js/src/lib/validate-schema.ts +++ b/packages/pgconductor-js/src/lib/validate-schema.ts @@ -1,20 +1,28 @@ import type { StandardSchemaV1 } from "@standard-schema/spec"; +export async function parseSchema( + schema: StandardSchemaV1 | undefined, + value: T, + label: string, +): Promise<{ value: T } | { error: string }> { + if (!schema) return { value }; + const result = await schema["~standard"].validate(value); + if (!result.issues) return { value: result.value }; + const issues = result.issues.map((issue) => { + const path = issue.path + ?.map((segment) => String(typeof segment === "object" ? segment.key : segment)) + .join("."); + return path ? `${path}: ${issue.message}` : issue.message; + }); + return { error: `Invalid ${label}: ${issues.join("; ")}` }; +} + export async function validateSchema( schema: StandardSchemaV1 | undefined, value: T, label: string, ): Promise { - if (!schema) return value; - const result = await schema["~standard"].validate(value); - if (result.issues) { - const issues = result.issues.map((issue) => { - const path = issue.path - ?.map((segment) => String(typeof segment === "object" ? segment.key : segment)) - .join("."); - return path ? `${path}: ${issue.message}` : issue.message; - }); - throw new Error(`Invalid ${label}: ${issues.join("; ")}`); - } - return result.value; + const parsed = await parseSchema(schema, value, label); + if ("error" in parsed) throw new Error(parsed.error); + return parsed.value; } diff --git a/packages/pgconductor-js/src/task.ts b/packages/pgconductor-js/src/task.ts index 8b0dd0e4..cf23c355 100644 --- a/packages/pgconductor-js/src/task.ts +++ b/packages/pgconductor-js/src/task.ts @@ -77,6 +77,12 @@ export type ValidateDeadLetterConfiguration = T exte export type RetentionSettings = boolean | { days: number }; +/** Standard Schemas the worker validates the payload and return value of every execution with. */ +export type TaskRuntimeSchemas = { + readonly payload?: StandardSchemaV1; + readonly returns?: StandardSchemaV1; +}; + export type TaskEvent

= | { name: "pgconductor.cron" } | { name: "pgconductor.invoke"; payload: P }; @@ -174,6 +180,7 @@ export class Task< public readonly execute: ExecuteFunction, public readonly eventDefinitions: readonly EventDefinition[] = [], public readonly metadataSchema?: StandardSchemaV1, + public readonly schemas: TaskRuntimeSchemas = {}, ) { const { name, queue, ...config } = definition; this.name = name; @@ -212,6 +219,7 @@ export class Task< execute: ExecuteFunction, eventDefinitions: readonly EventDefinition[] = [], metadataSchema?: StandardSchemaV1, + schemas: TaskRuntimeSchemas = {}, ): Task { return new Task( definition, @@ -219,6 +227,7 @@ export class Task< execute, eventDefinitions, metadataSchema, + schemas, ); } } diff --git a/packages/pgconductor-js/src/worker.ts b/packages/pgconductor-js/src/worker.ts index 3fe2a163..dde09cd3 100644 --- a/packages/pgconductor-js/src/worker.ts +++ b/packages/pgconductor-js/src/worker.ts @@ -4,7 +4,9 @@ import type { Execution, ExecutionResult, ExecutionSpec, + Payload, } from "./database-client.js"; +import type { StandardSchemaV1 } from "@standard-schema/spec"; import type { AnyTask, BatchConfig } from "./task.js"; import type { TaskDefinition } from "./task-definition.js"; import { waitFor } from "./lib/wait-for.js"; @@ -25,6 +27,7 @@ import { Telemetry } from "./telemetry.js"; import { noop } from "./lib/noop.js"; import type { Middleware, MiddlewareResult } from "./middleware.js"; import { SignalInbox } from "./signal-inbox.js"; +import { parseSchema } from "./lib/validate-schema.js"; /** * The configuration options for the Worker. @@ -69,6 +72,30 @@ function taskEventFor(execution: Execution): { name: string; payload?: unknown } return { name: "pgconductor.invoke", payload: execution.payload }; } +// A value the schema rejects fails the execution without retries; a validator that throws fails the attempt +async function parseOrFail( + schema: StandardSchemaV1 | undefined, + value: unknown, + label: string, +): Promise<{ value: unknown } | { status: "failed" | "permanently_failed"; error: string }> { + try { + const parsed = await parseSchema(schema, value, label); + return "error" in parsed ? { status: "permanently_failed", error: parsed.error } : parsed; + } catch (err) { + return { status: "failed", error: coerceError(err).message }; + } +} + +// Invocations carry the payload of the task's schema, events and cron occurrences do not +async function parsePayload(task: AnyTask, event: { name: string; payload?: unknown }) { + if (event.name !== "pgconductor.invoke") return { value: event.payload }; + return parseOrFail(task.schemas.payload, event.payload, `payload for task "${task.name}"`); +} + +function parseResult(task: AnyTask, result: unknown) { + return parseOrFail(task.schemas.returns, result, `return value of task "${task.name}"`); +} + /** * Worker implemented as async pipeline: fetch → execute → flush. * Uses async iterators for clean composition and natural backpressure. @@ -603,6 +630,10 @@ export class Worker< ); try { + const payload = await parsePayload(task, taskEvent); + if ("error" in payload) return { ...identity, ...payload }; + taskEvent.payload = payload.value; + await this.scheduleNextExecution(exec); const outcome = await this.telemetry.process({ @@ -681,10 +712,13 @@ export class Worker< } as const; } + const result = await parseResult(task, outcome.output); + if ("error" in result) return { ...identity, ...result }; + return { ...identity, status: "completed", - result: outcome.output, + result: result.value as Payload | undefined, signal_names: signals.names, } as const; } catch (err) { @@ -718,20 +752,52 @@ export class Worker< private async executeBatchTask( task: AnyTask, taskKey: string, - executions: Execution[], + claimed: Execution[], ): Promise { - const events = executions.map((exec) => { - return { - ...taskEventFor(exec), + const parsed = await Promise.all( + claimed.map(async (exec) => { + const event = taskEventFor(exec); + return { exec, event, payload: await parsePayload(task, event) }; + }), + ); + + // Members with an invalid payload settle on their own, the others run as the batch + const rejected: ExecutionResult[] = []; + const executions: Execution[] = []; + const events: object[] = []; + for (const { exec, event, payload } of parsed) { + if ("error" in payload) { + rejected.push({ + execution_id: exec.id, + claim_token: exec.claim_token, + queue: exec.queue, + task_key: taskKey, + ...payload, + }); + continue; + } + executions.push(exec); + events.push({ + ...event, + payload: payload.value, execution: { id: exec.id, queue: exec.queue, task_key: exec.task_key, locked_by: exec.locked_by, }, - }; - }); + }); + } + if (executions.length === 0) return rejected; + return [...rejected, ...(await this.executeBatch(task, taskKey, executions, events))]; + } + private async executeBatch( + task: AnyTask, + taskKey: string, + executions: Execution[], + events: object[], + ): Promise { const taskAbortController = new TypedAbortController(); const timeout = task.timeoutMs ? setTimeout(() => taskAbortController.abort({ reason: "timed-out" }), task.timeoutMs) @@ -828,14 +894,23 @@ export class Worker< })); } - return executions.map((exec, i) => ({ - execution_id: exec.id, - claim_token: exec.claim_token, - queue: exec.queue, - task_key: taskKey, - status: "completed" as const, - result: outcome.output?.[i], - })); + return Promise.all( + executions.map(async (exec, i) => { + const identity = { + execution_id: exec.id, + claim_token: exec.claim_token, + queue: exec.queue, + task_key: taskKey, + }; + const result = await parseResult(task, outcome.output?.[i]); + if ("error" in result) return { ...identity, ...result }; + return { + ...identity, + status: "completed" as const, + result: result.value as Payload | undefined, + }; + }), + ); } catch (err) { if (this.signal.aborted) { return executions.map((exec) => ({ diff --git a/packages/pgconductor-js/tests/integration/task-validation.test.ts b/packages/pgconductor-js/tests/integration/task-validation.test.ts new file mode 100644 index 00000000..efde10ae --- /dev/null +++ b/packages/pgconductor-js/tests/integration/task-validation.test.ts @@ -0,0 +1,314 @@ +import { z } from "zod"; +import type { StandardSchemaV1 } from "@standard-schema/spec"; +import { test, expect, describe, beforeAll, afterAll, afterEach } from "vitest"; +import { Conductor } from "../../src/conductor"; +import { Orchestrator } from "../../src/orchestrator"; +import { defineTask, type DefineTask } from "../../src/task-definition"; +import { TestDatabasePool, type TestDatabase } from "../fixtures/test-database"; +import { TaskSchemas } from "../../src/schemas"; + +// Resolves asynchronously, like a validator that checks a value against a service +const asyncCode: StandardSchemaV1<{ code: string }, { code: string }> = { + "~standard": { + version: 1, + vendor: "test", + validate: async (value) => { + await new Promise((resolve) => setTimeout(resolve, 10)); + const code = typeof value === "object" && value && "code" in value ? value.code : null; + return code === "valid" + ? { value: { code } } + : { issues: [{ message: "unknown code", path: ["code"] }] }; + }, + }, +}; + +const throwing: StandardSchemaV1<{}, {}> = { + "~standard": { + version: 1, + vendor: "test", + validate: () => { + throw new Error("validator unavailable"); + }, + }, +}; + +const charge = defineTask({ + name: "charge", + payload: z.object({ amount: z.number().positive() }), + returns: z.object({ charged: z.number() }), +}); + +const normalize = defineTask({ + name: "normalize", + payload: z.object({ + email: z.string().trim().toLowerCase(), + retries: z.number().default(3), + }), + returns: z.object({ email: z.string(), at: z.string().transform((at) => at.slice(0, 4)) }), +}); + +const redeem = defineTask({ name: "redeem", payload: asyncCode }); + +const sum = defineTask({ + name: "sum", + payload: z.object({ a: z.number() }), + returns: z.object({ total: z.number().int() }), +}); + +const unavailable = defineTask({ name: "unavailable", payload: throwing }); + +const batchSquare = defineTask({ + name: "batch-square", + payload: z.object({ value: z.number().int() }), + returns: z.object({ square: z.number().max(50) }), +}); + +type Untyped = DefineTask<{ name: "untyped"; payload: { amount: number } }>; + +describe("Task schema validation", () => { + 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(); + }); + + async function setup() { + const db = await pool.child(); + databases.push(db); + + const conductor = Conductor.create({ + sql: db.sql, + tasks: TaskSchemas.fromSchema([ + charge, + normalize, + redeem, + sum, + unavailable, + batchSquare, + ]).fromUnion(), + context: {}, + }); + + const received: { task: string; payload: unknown }[] = []; + const record = (task: string, event: { name: string; payload?: unknown }) => { + received.push({ task, payload: event.payload }); + }; + + const tasks = [ + conductor.createTask({ name: "charge" }, { invocable: true }, async (event) => { + record("charge", event); + return { charged: event.payload.amount }; + }), + conductor.createTask({ name: "normalize" }, { invocable: true }, async (event) => { + record("normalize", event); + return { email: event.payload.email, at: "2024-01-01" }; + }), + conductor.createTask({ name: "redeem" }, { invocable: true }, async (event) => { + record("redeem", event); + }), + conductor.createTask({ name: "sum" }, { invocable: true }, async (event) => { + record("sum", event); + return { total: event.payload.a }; + }), + conductor.createTask({ name: "unavailable" }, { invocable: true }, async (event) => { + record("unavailable", event); + }), + conductor.createTask( + { name: "batch-square", batch: { size: 4, timeoutMs: 50 } }, + { invocable: true }, + async (events) => + events.map((event) => { + record("batch-square", event); + return { square: event.payload.value ** 2 }; + }), + ), + conductor.createTask({ name: "untyped" }, { invocable: true }, async (event) => { + record("untyped", event); + }), + ]; + + const orchestrator = Orchestrator.create({ + conductor, + tasks, + defaultWorker: { pollIntervalMs: 50, flushIntervalMs: 50, concurrency: 2 }, + }); + + await conductor.ensureInstalled(); + return { db, conductor, orchestrator, received }; + } + + test("an invalid payload fails without retries and never reaches the handler", async () => { + const { conductor, orchestrator, received } = await setup(); + + const id = await conductor.invoke({ name: "charge" }, { amount: -1 }); + await orchestrator.drain(); + + expect(await conductor.getExecution(id)).toMatchObject({ + status: "failed", + attempts: 1, + error: 'Invalid payload for task "charge": amount: Too small: expected number to be >0', + }); + expect(received).toEqual([]); + }); + + test("payloads of executions invoked from SQL are validated", async () => { + const { db, conductor, orchestrator, received } = await setup(); + + const [invalid] = await db.sql<{ id: string }[]>` + select id from pgconductor.invoke('charge', 'default', '{"amount": "10"}'::jsonb) + `; + const [missing] = await db.sql<{ id: string }[]>` + select id from pgconductor.invoke('charge') + `; + const [valid] = await db.sql<{ id: string }[]>` + select id from pgconductor.invoke('charge', 'default', '{"amount": 10}'::jsonb) + `; + await orchestrator.drain(); + + expect(await conductor.getExecution(invalid?.id || "")).toMatchObject({ + status: "failed", + error: + 'Invalid payload for task "charge": amount: Invalid input: expected number, received string', + }); + expect(await conductor.getExecution(missing?.id || "")).toMatchObject({ status: "failed" }); + expect(await conductor.getExecution(valid?.id || "")).toMatchObject({ + status: "completed", + result: { charged: 10 }, + }); + expect(received).toEqual([{ task: "charge", payload: { amount: 10 } }]); + }); + + test("the handler receives the parsed payload and the parsed return value is stored", async () => { + const { conductor, orchestrator, received } = await setup(); + + const id = await conductor.invoke({ name: "normalize" }, { + email: " Alice@Example.COM ", + retries: 1, + ignored: true, + } as { + email: string; + retries: number; + }); + await orchestrator.drain(); + + expect(received).toEqual([ + { task: "normalize", payload: { email: "alice@example.com", retries: 1 } }, + ]); + expect(await conductor.getExecution(id)).toMatchObject({ + status: "completed", + result: { email: "alice@example.com", at: "2024" }, + }); + }); + + test("schema defaults apply to payloads", async () => { + const { db, orchestrator, received } = await setup(); + + await db.sql`select pgconductor.invoke('normalize', 'default', '{"email": "bob@example.com"}'::jsonb)`; + await orchestrator.drain(); + + expect(received).toEqual([ + { task: "normalize", payload: { email: "bob@example.com", retries: 3 } }, + ]); + }); + + test("async validators are awaited", async () => { + const { conductor, orchestrator, received } = await setup(); + + const valid = await conductor.invoke({ name: "redeem" }, { code: "valid" }); + const invalid = await conductor.invoke({ name: "redeem" }, { code: "expired" }); + await orchestrator.drain(); + + expect(await conductor.getExecution(valid)).toMatchObject({ status: "completed" }); + expect(await conductor.getExecution(invalid)).toMatchObject({ + status: "failed", + attempts: 1, + error: 'Invalid payload for task "redeem": code: unknown code', + }); + expect(received).toEqual([{ task: "redeem", payload: { code: "valid" } }]); + }); + + test("an invalid return value fails without retries", async () => { + const { conductor, orchestrator, received } = await setup(); + + const id = await conductor.invoke({ name: "sum" }, { a: 1.5 }); + await orchestrator.drain(); + + expect(await conductor.getExecution(id)).toMatchObject({ + status: "failed", + attempts: 1, + result: null, + error: + 'Invalid return value of task "sum": total: Invalid input: expected int, received number', + }); + expect(received).toHaveLength(1); + }); + + test("a validator that throws fails the attempt and retries", async () => { + const { db, conductor, orchestrator, received } = await setup(); + + const id = await conductor.invoke({ name: "unavailable" }, {}); + await orchestrator.drain(); + + const [row] = await db.sql<{ attempts: number; failed_at: Date | null; last_error: string }[]>` + select attempts, failed_at, last_error from pgconductor._private_executions where id = ${id} + `; + expect(row).toEqual({ attempts: 1, failed_at: null, last_error: "validator unavailable" }); + expect(received).toEqual([]); + }); + + test("batch tasks validate payloads and return values per item", async () => { + const { conductor, orchestrator, received } = await setup(); + + const [three, half, nine, four] = await conductor.invoke({ name: "batch-square" }, [ + { payload: { value: 3 } }, + { payload: { value: 0.5 } }, + { payload: { value: 9 } }, + { payload: { value: 4 } }, + ]); + await orchestrator.drain(); + + expect(received.map(({ payload }) => payload)).toHaveLength(3); + expect(received.map(({ payload }) => payload)).toEqual( + expect.arrayContaining([{ value: 3 }, { value: 9 }, { value: 4 }]), + ); + expect(await conductor.getExecution(three || "")).toMatchObject({ + status: "completed", + result: { square: 9 }, + }); + expect(await conductor.getExecution(half || "")).toMatchObject({ + status: "failed", + attempts: 1, + error: + 'Invalid payload for task "batch-square": value: Invalid input: expected int, received number', + }); + expect(await conductor.getExecution(nine || "")).toMatchObject({ + status: "failed", + attempts: 1, + error: + 'Invalid return value of task "batch-square": square: Too big: expected number to be <=50', + }); + expect(await conductor.getExecution(four || "")).toMatchObject({ + status: "completed", + result: { square: 16 }, + }); + }); + + test("type-only definitions are not validated", async () => { + const { db, orchestrator, received } = await setup(); + + await db.sql`select pgconductor.invoke('untyped', 'default', '{"amount": "ten"}'::jsonb)`; + await orchestrator.drain(); + + expect(received).toEqual([{ task: "untyped", payload: { amount: "ten" } }]); + }); +}); From 993144ba28785e08a3c22226f9e6b2f566ce1dc7 Mon Sep 17 00:00:00 2001 From: psteinroe Date: Tue, 20 Oct 2026 13:11:27 +0000 Subject: [PATCH 2/2] fix(signals): start signalWithStart executions with an empty payload --- docs/content/api/conductor.md | 2 +- docs/content/task-execution/signals.md | 2 +- packages/pgconductor-js/src/conductor.ts | 2 +- packages/pgconductor-js/src/task-context.ts | 2 +- 4 files changed, 4 insertions(+), 4 deletions(-) diff --git a/docs/content/api/conductor.md b/docs/content/api/conductor.md index c633ae60..782bce87 100644 --- a/docs/content/api/conductor.md +++ b/docs/content/api/conductor.md @@ -238,7 +238,7 @@ const executionId = await conductor.signalWithStart({ queue: "default", // optional key: "thread-1", signal: { name: "input", id: requestId, payload }, - input: {}, // used only when it starts an execution + input: {}, // used only when it starts an execution (default: {}) priority: 0, // optional metadata: {}, // optional }); diff --git a/docs/content/task-execution/signals.md b/docs/content/task-execution/signals.md index 3693a203..033823c2 100644 --- a/docs/content/task-execution/signals.md +++ b/docs/content/task-execution/signals.md @@ -53,7 +53,7 @@ const executionId = await conductor.signalWithStart({ task: "session", key: "slack:T1:C1:1712345678.000100", signal: { name: "input", id: requestId, payload }, - input: {}, // payload of the execution, used only when it starts one + input: {}, // payload of the execution, used only when it starts one (default: {}) }); ``` diff --git a/packages/pgconductor-js/src/conductor.ts b/packages/pgconductor-js/src/conductor.ts index f505461e..7bdd9fa2 100644 --- a/packages/pgconductor-js/src/conductor.ts +++ b/packages/pgconductor-js/src/conductor.ts @@ -489,7 +489,7 @@ export class Conductor< queue, key: options.key, signal: options.signal, - input: (options.input || null) as Payload | null, + input: (options.input || {}) as Payload, priority: options.priority, traceContext: this.telemetry.traceContext(), metadata, diff --git a/packages/pgconductor-js/src/task-context.ts b/packages/pgconductor-js/src/task-context.ts index 86caf921..68a9ccbb 100644 --- a/packages/pgconductor-js/src/task-context.ts +++ b/packages/pgconductor-js/src/task-context.ts @@ -745,7 +745,7 @@ export class TaskContext< queue: options.queue || "default", key: options.key, signal: options.signal, - input: (options.input || null) as Payload | null, + input: (options.input || {}) as Payload, priority: options.priority, traceContext: this.opts.telemetry.traceContext(), metadata: this.opts.execution.metadata,