diff --git a/docs/content/api/conductor.md b/docs/content/api/conductor.md index a8fbd7e..1956b1a 100644 --- a/docs/content/api/conductor.md +++ b/docs/content/api/conductor.md @@ -72,6 +72,7 @@ const task = conductor.createTask( removeOnComplete?: { days: number } | false; // Retention policy removeOnFail?: { days: number } | false; // Retention policy batch?: { size: number; timeoutMs: number }; // Batch processing config + deadLetter?: { queue: string; task?: Task }; // Final-failure destination } ``` diff --git a/docs/content/task-execution/dead-letter-queue.md b/docs/content/task-execution/dead-letter-queue.md new file mode 100644 index 0000000..24151e1 --- /dev/null +++ b/docs/content/task-execution/dead-letter-queue.md @@ -0,0 +1,24 @@ +# Dead-letter queues + +Configure a destination for executions that fail on their final attempt: + +```ts +const failedPayment = conductor.createTask( + { name: "failed-payment", queue: "payments-dlq" }, + { invocable: true }, + async (event) => {}, +); + +const chargeCard = conductor.createTask( + { + name: "charge-card", + deadLetter: { queue: "payments-dlq", task: failedPayment }, + }, + { invocable: true }, + async (event) => {}, +); +``` + +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. diff --git a/docs/zensical.toml b/docs/zensical.toml index 56b1789..7f9174b 100644 --- a/docs/zensical.toml +++ b/docs/zensical.toml @@ -73,6 +73,7 @@ nav = [ { "Cancellation" = "task-execution/cancellation.md" }, { "Priority" = "task-execution/priority.md" }, { "Concurrency" = "task-execution/concurrency.md" }, + { "Dead-letter queues" = "task-execution/dead-letter-queue.md" }, { "Deduplication" = "task-execution/deduplication.md" }, { "Rate Limiting" = "task-execution/rate-limiting.md" }, { "Batching" = "task-execution/batching.md" }, diff --git a/migrations/0000000001_setup.sql b/migrations/0000000001_setup.sql index dd65de7..01a65df 100644 --- a/migrations/0000000001_setup.sql +++ b/migrations/0000000001_setup.sql @@ -105,6 +105,14 @@ create table pgconductor._private_executions ( waiting_step_key text, parent_execution_id uuid, singleton_on timestamptz, + + -- Dead-letter metadata is denormalized so retained source rows are optional. + dead_letter_source_execution_id uuid, + dead_letter_source_queue text, + dead_letter_source_task_key text, + dead_letter_error text, + dead_letter_attempts integer, + dead_letter_failed_at timestamptz, primary key (id, queue), unique (task_key, dedupe_key, queue) ) partition by list (queue); @@ -147,10 +155,19 @@ create table pgconductor._private_tasks ( -- NULL means no limit (unlimited concurrency) concurrency_limit integer, group_concurrency_limit integer, + + -- Destination copied onto each source task registration. + dead_letter_queue text, + dead_letter_task_key text, + constraint positive_concurrency_limits check ( (concurrency_limit is null or concurrency_limit > 0) and (group_concurrency_limit is null or group_concurrency_limit > 0) ), + constraint dead_letter_not_self check ( + dead_letter_queue is null or dead_letter_queue <> queue or + (dead_letter_task_key is not null and dead_letter_task_key <> key) + ), primary key (queue, key) ); @@ -168,6 +185,10 @@ create table pgconductor._private_steps ( create index idx_steps_execution_id on pgconductor._private_steps (execution_id); +create unique index idx_executions_dead_letter_delivery + on pgconductor._private_executions (dead_letter_source_execution_id, queue, task_key) + where dead_letter_source_execution_id is not null; + -- Trigger function to manage executions partitions per queue -- Automatically creates partition when queue is inserted create or replace function pgconductor._private_manage_queue_partition() @@ -329,7 +350,9 @@ create type pgconductor.task_spec as ( window_start timetz, window_end timetz, concurrency_limit integer, - group_concurrency_limit integer + group_concurrency_limit integer, + dead_letter_queue text, + dead_letter_task_key text ); create type pgconductor._private_event_operation as enum ( @@ -367,8 +390,15 @@ begin values (p_queue_name) on conflict (name) do nothing; + -- Dead-letter destinations may not have a worker yet; create their partitions. + insert into pgconductor._private_queues (name) + select distinct spec.dead_letter_queue + from unnest(p_task_specs) as spec + where spec.dead_letter_queue is not null + on conflict (name) do nothing; + -- step 2: register/update tasks - insert into pgconductor._private_tasks (key, queue, max_attempts, remove_on_complete_days, remove_on_fail_days, window_start, window_end, concurrency_limit, group_concurrency_limit) + insert into pgconductor._private_tasks (key, queue, max_attempts, remove_on_complete_days, remove_on_fail_days, window_start, window_end, concurrency_limit, group_concurrency_limit, dead_letter_queue, dead_letter_task_key) select spec.key, coalesce(spec.queue, 'default'), @@ -378,7 +408,9 @@ begin spec.window_start, spec.window_end, spec.concurrency_limit, - spec.group_concurrency_limit + spec.group_concurrency_limit, + spec.dead_letter_queue, + spec.dead_letter_task_key from unnest(p_task_specs) as spec on conflict (queue, key) do update set @@ -389,7 +421,9 @@ begin window_start = excluded.window_start, window_end = excluded.window_end, concurrency_limit = excluded.concurrency_limit, - group_concurrency_limit = excluded.group_concurrency_limit; + group_concurrency_limit = excluded.group_concurrency_limit, + dead_letter_queue = excluded.dead_letter_queue, + dead_letter_task_key = excluded.dead_letter_task_key; -- step 3: insert scheduled cron executions insert into pgconductor._private_executions (task_key, queue, payload, run_at, dedupe_key, cron_expression, "group") diff --git a/packages/pgconductor-js/src/conductor.ts b/packages/pgconductor-js/src/conductor.ts index 26c6fb4..52116c6 100644 --- a/packages/pgconductor-js/src/conductor.ts +++ b/packages/pgconductor-js/src/conductor.ts @@ -8,6 +8,7 @@ import { type ValidateTasksQueue, type BatchConfig, type ExecuteFunction, + type ValidateDeadLetterConfiguration, } from "./task"; import type { TaskContext, BatchTaskContext } from "./task-context"; import { @@ -162,7 +163,7 @@ export class Conductor< }, const TTriggers extends object | readonly object[], >( - definition: TDef, + definition: TDef & ValidateDeadLetterConfiguration>, triggers: TTriggers & ValidateTriggers>, fn: TDef extends { readonly batch: BatchConfig } ? ResolvedReturns extends void @@ -198,7 +199,11 @@ export class Conductor< TaskContext & ExtraContext, TaskEventFromTriggers, Events, Database> >( - definition as TaskConfiguration>, + definition as TaskConfiguration< + TDef["name"], + ResolvedQueue, + ResolvedPayload + >, triggers as NonEmptyArray | Trigger, fn as ExecuteFunction< TaskEventFromTriggers, Events, Database>, diff --git a/packages/pgconductor-js/src/database-client.ts b/packages/pgconductor-js/src/database-client.ts index c2b523b..80d4f5f 100644 --- a/packages/pgconductor-js/src/database-client.ts +++ b/packages/pgconductor-js/src/database-client.ts @@ -48,8 +48,19 @@ export interface TaskSpec { window?: [string, string] | null; concurrency?: number | null; groupConcurrency?: number | null; + deadLetterQueue?: string | null; + deadLetterTaskKey?: string | null; } +export type DeadLetterMetadata = { + sourceExecutionId: string; + sourceQueue: string; + sourceTaskKey: string; + error: string | null; + attempts: number; + failedAt: Date; +}; + export interface Execution { id: string; task_key: string; @@ -63,6 +74,12 @@ export interface Execution { dedupe_key?: string | null; cron_expression?: string | null; group?: string | null; + dead_letter_source_execution_id?: string | null; + dead_letter_source_queue?: string | null; + dead_letter_source_task_key?: string | null; + dead_letter_error?: string | null; + dead_letter_attempts?: number | null; + dead_letter_failed_at?: Date | null; } // todo: move all of this to query-builder too or create new types.ts file diff --git a/packages/pgconductor-js/src/generated/sql.ts b/packages/pgconductor-js/src/generated/sql.ts index 97abfcd..199045a 100644 --- a/packages/pgconductor-js/src/generated/sql.ts +++ b/packages/pgconductor-js/src/generated/sql.ts @@ -121,6 +121,14 @@ create table pgconductor._private_executions ( waiting_step_key text, parent_execution_id uuid, singleton_on timestamptz, + + -- Dead-letter metadata is denormalized so retained source rows are optional. + dead_letter_source_execution_id uuid, + dead_letter_source_queue text, + dead_letter_source_task_key text, + dead_letter_error text, + dead_letter_attempts integer, + dead_letter_failed_at timestamptz, primary key (id, queue), unique (task_key, dedupe_key, queue) ) partition by list (queue); @@ -163,10 +171,19 @@ create table pgconductor._private_tasks ( -- NULL means no limit (unlimited concurrency) concurrency_limit integer, group_concurrency_limit integer, + + -- Destination copied onto each source task registration. + dead_letter_queue text, + dead_letter_task_key text, + constraint positive_concurrency_limits check ( (concurrency_limit is null or concurrency_limit > 0) and (group_concurrency_limit is null or group_concurrency_limit > 0) ), + constraint dead_letter_not_self check ( + dead_letter_queue is null or dead_letter_queue <> queue or + (dead_letter_task_key is not null and dead_letter_task_key <> key) + ), primary key (queue, key) ); @@ -184,6 +201,10 @@ create table pgconductor._private_steps ( create index idx_steps_execution_id on pgconductor._private_steps (execution_id); +create unique index idx_executions_dead_letter_delivery + on pgconductor._private_executions (dead_letter_source_execution_id, queue, task_key) + where dead_letter_source_execution_id is not null; + -- Trigger function to manage executions partitions per queue -- Automatically creates partition when queue is inserted create or replace function pgconductor._private_manage_queue_partition() @@ -345,7 +366,9 @@ create type pgconductor.task_spec as ( window_start timetz, window_end timetz, concurrency_limit integer, - group_concurrency_limit integer + group_concurrency_limit integer, + dead_letter_queue text, + dead_letter_task_key text ); create type pgconductor._private_event_operation as enum ( @@ -383,8 +406,15 @@ begin values (p_queue_name) on conflict (name) do nothing; + -- Dead-letter destinations may not have a worker yet; create their partitions. + insert into pgconductor._private_queues (name) + select distinct spec.dead_letter_queue + from unnest(p_task_specs) as spec + where spec.dead_letter_queue is not null + on conflict (name) do nothing; + -- step 2: register/update tasks - insert into pgconductor._private_tasks (key, queue, max_attempts, remove_on_complete_days, remove_on_fail_days, window_start, window_end, concurrency_limit, group_concurrency_limit) + insert into pgconductor._private_tasks (key, queue, max_attempts, remove_on_complete_days, remove_on_fail_days, window_start, window_end, concurrency_limit, group_concurrency_limit, dead_letter_queue, dead_letter_task_key) select spec.key, coalesce(spec.queue, 'default'), @@ -394,7 +424,9 @@ begin spec.window_start, spec.window_end, spec.concurrency_limit, - spec.group_concurrency_limit + spec.group_concurrency_limit, + spec.dead_letter_queue, + spec.dead_letter_task_key from unnest(p_task_specs) as spec on conflict (queue, key) do update set @@ -405,7 +437,9 @@ begin window_start = excluded.window_start, window_end = excluded.window_end, concurrency_limit = excluded.concurrency_limit, - group_concurrency_limit = excluded.group_concurrency_limit; + group_concurrency_limit = excluded.group_concurrency_limit, + dead_letter_queue = excluded.dead_letter_queue, + dead_letter_task_key = excluded.dead_letter_task_key; -- step 3: insert scheduled cron executions insert into pgconductor._private_executions (task_key, queue, payload, run_at, dedupe_key, cron_expression, "group") diff --git a/packages/pgconductor-js/src/query-builder.ts b/packages/pgconductor-js/src/query-builder.ts index 664872c..efee09b 100644 --- a/packages/pgconductor-js/src/query-builder.ts +++ b/packages/pgconductor-js/src/query-builder.ts @@ -373,10 +373,15 @@ export class QueryBuilder { where e.id = c.id and e.queue = ${queueName}::text and e.is_available = true returning e.id, e.task_key, e.queue, e.payload, e.waiting_on_execution_id, e.waiting_step_key, e.cancelled, e.last_error, e.dedupe_key, e.cron_expression, - e.locked_by, e."group", e.priority, e.run_at, e.created_at + e.locked_by, e."group", e.priority, e.run_at, e.created_at, + e.dead_letter_source_execution_id, e.dead_letter_source_queue, + e.dead_letter_source_task_key, e.dead_letter_error, + e.dead_letter_attempts, e.dead_letter_failed_at ) select id, task_key, queue, payload, waiting_on_execution_id, waiting_step_key, - cancelled, last_error, dedupe_key, cron_expression, locked_by, "group" + cancelled, last_error, dedupe_key, cron_expression, locked_by, "group", + dead_letter_source_execution_id, dead_letter_source_queue, + dead_letter_source_task_key, dead_letter_error, dead_letter_attempts, dead_letter_failed_at from claimed order by priority asc, run_at asc, created_at asc, id asc `; @@ -417,10 +422,13 @@ export class QueryBuilder { and e.locked_by = r.orchestrator_id for update of e )`); - ctes.push(this.sql`task_configs as ( - select queue, key, max_attempts, remove_on_complete_days, remove_on_fail_days - from pgconductor._private_tasks - where queue = any(${this.sql.array(Array.from(new Set(allResults.map((r) => r.queue))))}::text[]) + ctes.push(this.sql`task_configs as materialized ( + select t.queue, t.key, t.max_attempts, t.remove_on_complete_days, t.remove_on_fail_days, + t.dead_letter_queue, t.dead_letter_task_key + from ( + select distinct queue, task_key from valid_results + ) r + join pgconductor._private_tasks t on t.queue = r.queue and t.key = r.task_key )`); ctes.push(this.sql`completed_results as ( select * from valid_results where status = 'completed' and not execution_cancelled @@ -503,8 +511,13 @@ export class QueryBuilder { ctes.push(this.sql`permanently_failed_children as materialized ( select r.execution_id, r.queue, r.task_key, r.orchestrator_id, + r.execution_cancelled, coalesce(r.error, r.execution_last_error, 'unknown error') as child_error, - tc.remove_on_fail_days = 0 as should_remove + e."group" as execution_group, + e.payload as execution_payload, + e.attempts as execution_attempts, + tc.remove_on_fail_days = 0 as should_remove, + tc.dead_letter_queue, tc.dead_letter_task_key from failed_results r join pgconductor._private_executions e on e.id = r.execution_id and e.queue = r.queue join task_configs tc on tc.key = r.task_key and tc.queue = r.queue @@ -514,8 +527,13 @@ export class QueryBuilder { )`); ctes.push(this.sql`failed_parent_targets as materialized ( select p.execution_id as child_id, p.queue as child_queue, p.child_error, + p.execution_cancelled as child_cancelled, parent.id as parent_id, parent.queue as parent_queue, - pt.remove_on_fail_days = 0 as parent_should_remove + parent.task_key as parent_task_key, parent."group" as parent_group, + parent.payload as parent_payload, parent.attempts as parent_attempts, + pt.remove_on_fail_days = 0 as parent_should_remove, + pt.dead_letter_queue as parent_dead_letter_queue, + pt.dead_letter_task_key as parent_dead_letter_task_key from permanently_failed_children p join pgconductor._private_executions parent on parent.waiting_on_execution_id = p.execution_id @@ -524,6 +542,39 @@ export class QueryBuilder { where parent.completed_at is null and parent.failed_at is null and parent.locked_by is null for update of parent )`); + ctes.push(this.sql`terminal_failures as materialized ( + select p.execution_id, p.queue, p.task_key, p.execution_group, p.execution_payload as payload, + p.child_error as failure_error, p.execution_attempts as failure_attempts, + p.execution_cancelled, p.dead_letter_queue, p.dead_letter_task_key + from permanently_failed_children p + union all + select p.parent_id, p.parent_queue, p.parent_task_key, p.parent_group, p.parent_payload, + 'Child execution failed: ' || p.child_error, p.parent_attempts, p.child_cancelled, + p.parent_dead_letter_queue, p.parent_dead_letter_task_key + from failed_parent_targets p + )`); + ctes.push(this.sql`dead_lettered as materialized ( + insert into pgconductor._private_executions ( + task_key, queue, payload, run_at, "group", + dead_letter_source_execution_id, dead_letter_source_queue, + dead_letter_source_task_key, dead_letter_error, + dead_letter_attempts, dead_letter_failed_at + ) + select + coalesce(p.dead_letter_task_key, p.task_key), + coalesce(p.dead_letter_queue, p.queue), + p.payload, nt.ts, p.execution_group, + p.execution_id, p.queue, p.task_key, p.failure_error, + p.failure_attempts, nt.ts + from terminal_failures p + cross join now_ts nt + where p.dead_letter_queue is not null + and not p.execution_cancelled + on conflict (dead_letter_source_execution_id, queue, task_key) + where dead_letter_source_execution_id is not null + do update set dead_letter_source_execution_id = excluded.dead_letter_source_execution_id + returning id, dead_letter_source_execution_id + )`); ctes.push(this.sql`failed_updates as ( select p.execution_id as target_id, p.queue, p.child_error, true as is_child from permanently_failed_children p @@ -533,19 +584,30 @@ export class QueryBuilder { from failed_parent_targets p where p.parent_should_remove is not true )`); + ctes.push(this.sql`failed_delete_targets as materialized ( + select p.execution_id as target_id, p.queue, p.orchestrator_id as expected_locked_by + from permanently_failed_children p + where p.should_remove is true + and ( + p.execution_cancelled + or p.dead_letter_queue is null + or exists (select 1 from dead_lettered d where d.dead_letter_source_execution_id = p.execution_id) + ) + union all + select p.parent_id, p.parent_queue, null::uuid + from failed_parent_targets p + where p.parent_should_remove is true + and ( + p.child_cancelled + or p.parent_dead_letter_queue is null + or exists (select 1 from dead_lettered d where d.dead_letter_source_execution_id = p.parent_id) + ) + )`); ctes.push(this.sql`deleted_failed as ( delete from pgconductor._private_executions e - where exists ( - select 1 from permanently_failed_children p - where e.id = p.execution_id and e.queue = p.queue - and e.locked_by = p.orchestrator_id - and p.should_remove is true - ) - or exists ( - select 1 from failed_parent_targets p - where e.id = p.parent_id and e.queue = p.parent_queue - and p.parent_should_remove is true - ) + using failed_delete_targets f + where e.id = f.target_id and e.queue = f.queue + and e.locked_by is not distinct from f.expected_locked_by returning e.id )`); ctes.push(this.sql`updated_failed as ( @@ -668,6 +730,8 @@ export class QueryBuilder { window_end: spec.window?.[1] || null, concurrency_limit: spec.concurrency || null, group_concurrency_limit: spec.groupConcurrency || null, + dead_letter_queue: spec.deadLetterQueue || null, + dead_letter_task_key: spec.deadLetterTaskKey || null, })); const cronScheduleRows = cronSchedules.map((spec) => { diff --git a/packages/pgconductor-js/src/task-context.ts b/packages/pgconductor-js/src/task-context.ts index 539410e..14fc735 100644 --- a/packages/pgconductor-js/src/task-context.ts +++ b/packages/pgconductor-js/src/task-context.ts @@ -1,4 +1,10 @@ -import type { DatabaseClient, JsonValue, Execution, Payload } from "./database-client"; +import type { + DatabaseClient, + JsonValue, + Execution, + Payload, + DeadLetterMetadata, +} from "./database-client"; import { nextCronOccurrence } from "./lib/cron"; import type { Clock } from "./lib/clock"; import type { @@ -125,6 +131,28 @@ export class TaskContext< return this.opts.abortController.signal; } + /** Metadata describing the source execution when this is a DLQ delivery. */ + get deadLetter(): DeadLetterMetadata | null { + const execution = this.opts.execution; + if ( + !execution.dead_letter_source_execution_id || + !execution.dead_letter_source_queue || + !execution.dead_letter_source_task_key || + execution.dead_letter_attempts == null || + !execution.dead_letter_failed_at + ) { + return null; + } + return { + sourceExecutionId: execution.dead_letter_source_execution_id, + sourceQueue: execution.dead_letter_source_queue, + sourceTaskKey: execution.dead_letter_source_task_key, + error: execution.dead_letter_error ?? null, + attempts: execution.dead_letter_attempts, + failedAt: execution.dead_letter_failed_at, + }; + } + async step(name: string, fn: () => Promise | T): Promise { // Check abort signal if (this.signal.aborted) { diff --git a/packages/pgconductor-js/src/task.ts b/packages/pgconductor-js/src/task.ts index ae5c479..b611feb 100644 --- a/packages/pgconductor-js/src/task.ts +++ b/packages/pgconductor-js/src/task.ts @@ -25,14 +25,29 @@ export type TaskIdentifier>; + +function hasSameTaskIdentity( + left: QualifiedTaskIdentifier, + right: QualifiedTaskIdentifier, +): boolean { + return left.queue === right.queue && left.name === right.name; +} + export type BatchConfig = { size: number; timeoutMs: number; }; +export type DeadLetterConfiguration = { + readonly queue: string; + readonly task?: Task; +}; + export type TaskConfiguration< TName extends string = string, TQueue extends string = "default", + TPayload extends object = object, > = TaskIdentifier & { maxAttempts?: number; window?: [string, string]; @@ -41,8 +56,37 @@ export type TaskConfiguration< concurrency?: number; groupConcurrency?: number; batch?: BatchConfig; + deadLetter?: DeadLetterConfiguration; }; +/** Type-level validation for a dead-letter destination. */ +type ExactType = [TLeft] extends [TRight] + ? [TRight] extends [TLeft] + ? true + : false + : false; + +type ValidateDeadLetterTarget = + TTarget extends Task + ? TPayload extends TTargetPayload + ? ExactType extends true + ? unknown + : "deadLetter.queue must match deadLetter.task.queue" + : "deadLetter.task must accept the source task payload" + : "deadLetter.task must be a Task"; + +export type ValidateDeadLetterConfiguration = T extends { + readonly deadLetter: infer TDeadLetter; +} + ? TDeadLetter extends { readonly queue: infer TQueue } + ? TQueue extends string + ? TDeadLetter extends { readonly task: infer TTarget } + ? ValidateDeadLetterTarget + : unknown + : "deadLetter.queue must be a string" + : "deadLetter must include a queue string" + : unknown; + export type RetentionSettings = boolean | { days: number }; export type TaskEvent

= @@ -196,11 +240,12 @@ export class Task< public readonly concurrency?: number; public readonly groupConcurrency?: number; public readonly batch?: BatchConfig; + public readonly deadLetter?: DeadLetterConfiguration; public readonly triggers: NonEmptyArray; constructor( - definition: TaskConfiguration, + definition: TaskConfiguration, triggers: NonEmptyArray | Trigger, public readonly execute: ExecuteFunction, ) { @@ -215,6 +260,16 @@ export class Task< this.concurrency = assert.positiveInteger(config.concurrency, "concurrency"); this.groupConcurrency = assert.positiveInteger(config.groupConcurrency, "groupConcurrency"); this.batch = config.batch; + this.deadLetter = config.deadLetter; + if ( + this.deadLetter && + hasSameTaskIdentity(this, { + queue: this.deadLetter.queue, + name: this.deadLetter.task?.name ?? this.name, + }) + ) { + throw new Error("A task cannot dead-letter directly to itself"); + } this.triggers = Array.isArray(triggers) ? triggers : [triggers]; } @@ -227,7 +282,7 @@ export class Task< Context extends object, EventType, >( - definition: TaskConfiguration, + definition: TaskConfiguration, triggers: NonEmptyArray | Trigger, execute: ExecuteFunction, ): Task { diff --git a/packages/pgconductor-js/src/worker.ts b/packages/pgconductor-js/src/worker.ts index 49a410b..2274566 100644 --- a/packages/pgconductor-js/src/worker.ts +++ b/packages/pgconductor-js/src/worker.ts @@ -321,6 +321,8 @@ export class Worker< window: task.window, concurrency: task.concurrency, groupConcurrency: task.groupConcurrency, + deadLetterQueue: task.deadLetter?.queue, + deadLetterTaskKey: task.deadLetter?.task?.name, })); const allTasks = Array.from(this.tasks.values()); diff --git a/packages/pgconductor-js/tests/integration/dead-letter.test.ts b/packages/pgconductor-js/tests/integration/dead-letter.test.ts new file mode 100644 index 0000000..312758d --- /dev/null +++ b/packages/pgconductor-js/tests/integration/dead-letter.test.ts @@ -0,0 +1,577 @@ +import { afterAll, afterEach, beforeAll, describe, expect, test } from "bun:test"; +import { z } from "zod"; +import { Conductor } from "../../src/conductor"; +import { Orchestrator } from "../../src/orchestrator"; +import { TaskSchemas } from "../../src/schemas"; +import { defineTask } from "../../src/task-definition"; +import { TestDatabasePool, type TestDatabase } from "../fixtures/test-database"; +import { waitForCondition } from "../test-utils"; + +describe("dead-letter queues (Postgres integration)", () => { + let pool: TestDatabasePool; + const databases: TestDatabase[] = []; + + beforeAll(async () => { + pool = await TestDatabasePool.create(); + }, 60000); + + afterEach(async () => { + await Promise.all(databases.map((db) => db.destroy())); + databases.length = 0; + }); + + afterAll(async () => { + await pool?.destroy(); + }); + + test("retries, then delivers the final failure with payload and metadata", async () => { + const db = await pool.child(); + databases.push(db); + const payloadSchema = z.object({ value: z.string() }); + const sourceDefinition = defineTask({ name: "charge", payload: payloadSchema }); + const destinationDefinition = defineTask({ + name: "alternate-failure", + queue: "dlq", + payload: payloadSchema, + }); + const conductor = Conductor.create({ + sql: db.sql, + tasks: TaskSchemas.fromSchema([sourceDefinition, destinationDefinition]), + context: {}, + }); + const seen: Array<{ + value: string; + sourceTask: string; + attempts: number; + error: string | null; + }> = []; + const destination = conductor.createTask( + { name: "alternate-failure", queue: "dlq" }, + { invocable: true }, + async (event, ctx) => { + if (event.name === "pgconductor.invoke") { + if (!ctx.deadLetter) throw new Error("missing dead-letter metadata"); + seen.push({ + value: event.payload.value, + sourceTask: ctx.deadLetter.sourceTaskKey, + attempts: ctx.deadLetter.attempts, + error: ctx.deadLetter.error, + }); + } + }, + ); + const source = conductor.createTask( + { + name: "charge", + maxAttempts: 2, + removeOnFail: true, + deadLetter: { queue: "dlq", task: destination }, + }, + { invocable: true }, + async () => { + throw new Error("card declined"); + }, + ); + const sourceOrchestrator = Orchestrator.create({ + conductor, + tasks: [source], + defaultWorker: { pollIntervalMs: 10, flushIntervalMs: 10 }, + }); + await sourceOrchestrator.start(); + await conductor.invoke({ name: "charge" }, { value: "order-42" }); + await waitForCondition(async () => { + const rows = await db.sql<{ attempts: number; released: boolean }[]>` + select attempts, locked_by is null as released + from pgconductor._private_executions + where task_key = 'charge' + `; + return rows[0]?.attempts === 1 && rows[0].released; + }); + const [retry] = await db.sql<{ run_at: Date }[]>` + select run_at from pgconductor._private_executions + where task_key = 'charge' + `; + if (!retry) throw new Error("expected persisted retry"); + await db.client.setFakeTime({ date: new Date(retry.run_at.getTime() + 1) }); + await waitForCondition(async () => { + const rows = await db.sql<{ count: string }[]>` + select count(*)::text as count from pgconductor._private_executions + where queue = 'dlq' + and dead_letter_source_task_key = 'charge' + `; + return rows[0]?.count === "1"; + }); + const sourceRows = await db.sql<{ id: string }[]>` + select id from pgconductor._private_executions where task_key = 'charge' + `; + expect(sourceRows).toHaveLength(0); + await sourceOrchestrator.stop(); + + // The source registration created the destination partition, but the destination + // worker was intentionally started later. + const destinationOrchestrator = Orchestrator.create({ + conductor, + workers: [ + conductor.createWorker({ + queue: "dlq", + tasks: [destination], + config: { pollIntervalMs: 10, flushIntervalMs: 10 }, + }), + ], + }); + await destinationOrchestrator.start(); + await waitForCondition(async () => seen.length === 1); + await destinationOrchestrator.stop(); + expect(seen).toEqual([ + { value: "order-42", sourceTask: "charge", attempts: 2, error: "card declined" }, + ]); + }, 60_000); + + test("ignores duplicate settlements and wrong worker identity", async () => { + const db = await pool.child(); + databases.push(db); + const payloadSchema = z.object({ value: z.string() }); + const sourceDefinition = defineTask({ name: "settle-source", payload: payloadSchema }); + const destinationDefinition = defineTask({ + name: "settle-destination", + queue: "dlq", + payload: payloadSchema, + }); + const conductor = Conductor.create({ + sql: db.sql, + tasks: TaskSchemas.fromSchema([sourceDefinition, destinationDefinition]), + context: {}, + }); + await conductor.ensureInstalled(); + await db.client.registerWorker({ + queueName: "default", + taskSpecs: [ + { + key: "settle-source", + queue: "default", + maxAttempts: 1, + deadLetterQueue: "dlq", + deadLetterTaskKey: "settle-destination", + }, + ], + cronSchedules: [], + eventSubscriptions: [], + }); + const executionId = await db.client.invoke({ + task_key: "settle-source", + queue: "default", + payload: { value: "x" }, + }); + const orchestratorId = crypto.randomUUID(); + const claimed = await db.client.getExecutions({ + orchestratorId, + queueName: "default", + batchSize: 1, + filterTaskKeys: [], + }); + expect(claimed).toHaveLength(1); + const execution = claimed[0]!; + const result = { + execution_id: execution.id, + queue: execution.queue, + task_key: execution.task_key, + status: "permanently_failed" as const, + orchestrator_id: execution.locked_by, + error: "settlement failure", + }; + await db.client.returnExecutions({ + count: 1, + orchestratorId, + completed: [], + failed: [{ ...result }], + released: [], + invokeChild: [], + taskKeys: new Set([execution.task_key]), + }); + await db.client.returnExecutions({ + count: 1, + orchestratorId, + completed: [], + failed: [{ ...result }], + released: [], + invokeChild: [], + taskKeys: new Set([execution.task_key]), + }); + const rows = await db.sql<{ count: string; source: string | null }[]>` + select count(*)::text as count, min(dead_letter_source_execution_id::text) as source + from pgconductor._private_executions where queue = 'dlq' + `; + expect(rows[0]).toEqual({ count: "1", source: executionId }); + + // A result from a different worker is fenced out as well. + await db.client.returnExecutions({ + count: 1, + orchestratorId: crypto.randomUUID(), + completed: [], + failed: [{ ...result, orchestrator_id: crypto.randomUUID() }], + released: [], + invokeChild: [], + taskKeys: new Set([execution.task_key]), + }); + const count = await db.sql<{ count: string }[]>` + select count(*)::text as count from pgconductor._private_executions where queue = 'dlq' + `; + expect(count[0]?.count).toBe("1"); + }, 15000); + + test("cancellation never delivers to the DLQ", async () => { + const db = await pool.child(); + databases.push(db); + const sourceDefinition = defineTask({ name: "cancel-source", payload: z.object({}) }); + const destinationDefinition = defineTask({ + name: "cancel-destination", + queue: "dlq", + payload: z.object({}), + }); + const conductor = Conductor.create({ + sql: db.sql, + tasks: TaskSchemas.fromSchema([sourceDefinition, destinationDefinition]), + context: {}, + }); + await conductor.ensureInstalled(); + await db.client.registerWorker({ + queueName: "default", + taskSpecs: [ + { + key: "cancel-source", + queue: "default", + maxAttempts: 1, + deadLetterQueue: "dlq", + deadLetterTaskKey: "cancel-destination", + }, + ], + cronSchedules: [], + eventSubscriptions: [], + }); + const id = (await db.client.invoke({ + task_key: "cancel-source", + queue: "default", + payload: {}, + }))!; + expect(await db.client.cancelExecution(id)).toBe(true); + const rows = await db.sql<{ failed_at: Date | null; cancelled: boolean }[]>` + select failed_at, cancelled from pgconductor._private_executions where id = ${id}::uuid + `; + expect(rows[0]?.failed_at).not.toBeNull(); + expect(rows[0]?.cancelled).toBe(false); + const destinationRows = await db.sql<{ count: string }[]>` + select count(*)::text as count from pgconductor._private_executions where queue = 'dlq' + `; + expect(destinationRows[0]?.count).toBe("0"); + }, 15000); + + test("removes a claimed cancellation without delivering to the DLQ", async () => { + const db = await pool.child(); + databases.push(db); + const conductor = Conductor.create({ sql: db.sql, context: {} }); + await conductor.ensureInstalled(); + await db.client.registerWorker({ + queueName: "default", + taskSpecs: [ + { + key: "cancelled-running-source", + queue: "default", + maxAttempts: 1, + removeOnFailDays: 0, + deadLetterQueue: "dlq", + deadLetterTaskKey: "cancelled-running-destination", + }, + ], + cronSchedules: [], + eventSubscriptions: [], + }); + const executionId = await db.client.invoke({ + task_key: "cancelled-running-source", + queue: "default", + payload: {}, + }); + const orchestratorId = crypto.randomUUID(); + await db.client.orchestratorHeartbeat({ + orchestratorId, + version: "test", + migrationNumber: 1, + }); + const [execution] = await db.client.getExecutions({ + orchestratorId, + queueName: "default", + batchSize: 1, + filterTaskKeys: [], + }); + if (!execution || !executionId) throw new Error("expected claimed execution"); + + expect(await db.client.cancelExecution(executionId)).toBe(true); + await db.client.returnExecutions({ + count: 1, + orchestratorId, + completed: [], + failed: [ + { + execution_id: execution.id, + queue: execution.queue, + task_key: execution.task_key, + orchestrator_id: execution.locked_by, + status: "permanently_failed", + error: "Task was cancelled", + }, + ], + released: [], + invokeChild: [], + taskKeys: new Set([execution.task_key]), + }); + + const [counts] = await db.sql<{ source: string; destination: string }[]>` + select + count(*) filter (where queue = 'default')::text as source, + count(*) filter (where queue = 'dlq')::text as destination + from pgconductor._private_executions + `; + expect(counts).toEqual({ source: "0", destination: "0" }); + }, 15000); + + test("removes a parent failed by a claimed child cancellation without delivering to the DLQ", async () => { + const db = await pool.child(); + databases.push(db); + const conductor = Conductor.create({ sql: db.sql, context: {} }); + await conductor.ensureInstalled(); + await db.client.registerWorker({ + queueName: "default", + taskSpecs: [ + { + key: "cancelled-child-parent", + queue: "default", + removeOnFailDays: 0, + deadLetterQueue: "dlq", + deadLetterTaskKey: "cancelled-parent-destination", + }, + { + key: "cancelled-child", + queue: "default", + maxAttempts: 1, + removeOnFailDays: 0, + }, + ], + cronSchedules: [], + eventSubscriptions: [], + }); + const parentId = await db.client.invoke({ + task_key: "cancelled-child-parent", + queue: "default", + payload: {}, + }); + const parent = ( + await db.client.getExecutions({ + orchestratorId: crypto.randomUUID(), + queueName: "default", + batchSize: 1, + filterTaskKeys: [], + }) + )[0]; + if (!parent || !parentId) throw new Error("expected claimed parent execution"); + + await db.client.returnExecutions({ + count: 1, + orchestratorId: parent.locked_by, + completed: [], + failed: [], + released: [], + invokeChild: [ + { + execution_id: parent.id, + queue: parent.queue, + task_key: parent.task_key, + orchestrator_id: parent.locked_by, + status: "invoke_child", + timeout_ms: "infinity", + step_key: "child-step", + child_task_name: "cancelled-child", + child_task_queue: "default", + child_payload: {}, + }, + ], + taskKeys: new Set([parent.task_key]), + }); + const childOrchestratorId = crypto.randomUUID(); + await db.client.orchestratorHeartbeat({ + orchestratorId: childOrchestratorId, + version: "test", + migrationNumber: 1, + }); + const child = ( + await db.client.getExecutions({ + orchestratorId: childOrchestratorId, + queueName: "default", + batchSize: 1, + filterTaskKeys: [], + }) + )[0]; + if (!child) throw new Error("expected claimed child execution"); + + expect(await db.client.cancelExecution(child.id)).toBe(true); + await db.client.returnExecutions({ + count: 1, + orchestratorId: childOrchestratorId, + completed: [], + failed: [ + { + execution_id: child.id, + queue: child.queue, + task_key: child.task_key, + orchestrator_id: child.locked_by, + status: "permanently_failed", + error: "Task was cancelled", + }, + ], + released: [], + invokeChild: [], + taskKeys: new Set([child.task_key]), + }); + + const [counts] = await db.sql<{ source: string; destination: string }[]>` + select + count(*) filter (where queue = 'default')::text as source, + count(*) filter (where queue = 'dlq')::text as destination + from pgconductor._private_executions + where id = ${parentId}::uuid or queue = 'dlq' + `; + expect(counts).toEqual({ source: "0", destination: "0" }); + }, 15000); + + test("rejects direct self-targets in the database but permits cross-queue identity", async () => { + const db = await pool.child(); + databases.push(db); + const conductor = Conductor.create({ sql: db.sql, context: {} }); + await conductor.ensureInstalled(); + await expect( + db.client.registerWorker({ + queueName: "default", + taskSpecs: [ + { + key: "self", + queue: "default", + maxAttempts: 1, + deadLetterQueue: "default", + }, + ], + cronSchedules: [], + eventSubscriptions: [], + }), + ).rejects.toThrow(); + + await expect( + db.client.registerWorker({ + queueName: "default", + taskSpecs: [ + { + key: "same-name", + queue: "default", + maxAttempts: 1, + deadLetterQueue: "other-queue", + deadLetterTaskKey: "same-name", + }, + ], + cronSchedules: [], + eventSubscriptions: [], + }), + ).resolves.toBeUndefined(); + }, 15000); + + test("rolls back the source settlement when destination insertion fails", async () => { + const db = await pool.child(); + databases.push(db); + const payloadSchema = z.object({ value: z.string() }); + const sourceDefinition = defineTask({ name: "rollback-source", payload: payloadSchema }); + const destinationDefinition = defineTask({ + name: "rollback-destination", + queue: "dlq", + payload: payloadSchema, + }); + const conductor = Conductor.create({ + sql: db.sql, + tasks: TaskSchemas.fromSchema([sourceDefinition, destinationDefinition]), + context: {}, + }); + await conductor.ensureInstalled(); + await db.client.registerWorker({ + queueName: "default", + taskSpecs: [ + { + key: "rollback-source", + queue: "default", + maxAttempts: 1, + deadLetterQueue: "dlq", + deadLetterTaskKey: "rollback-destination", + }, + ], + cronSchedules: [], + eventSubscriptions: [], + }); + const id = await db.client.invoke({ + task_key: "rollback-source", + queue: "default", + payload: { value: "rollback" }, + }); + const claimed = await db.client.getExecutions({ + orchestratorId: crypto.randomUUID(), + queueName: "default", + batchSize: 1, + filterTaskKeys: [], + }); + const execution = claimed[0]!; + if (!execution.locked_by) throw new Error("execution was not claimed"); + const lockedBy = execution.locked_by; + await db.sql.unsafe(` + create function public.fail_dlq_insert() returns trigger language plpgsql as $$ + begin raise exception 'forced destination failure'; end; + $$; + create trigger fail_dlq_insert before insert on pgconductor.executions_dlq + for each row execute function public.fail_dlq_insert(); + `); + const settlement = { + execution_id: execution.id, + queue: execution.queue, + task_key: execution.task_key, + status: "permanently_failed" as const, + orchestrator_id: lockedBy, + error: "rollback failure", + }; + await expect( + db.client.returnExecutions({ + count: 1, + orchestratorId: lockedBy, + completed: [], + failed: [settlement], + released: [], + invokeChild: [], + taskKeys: new Set([execution.task_key]), + }), + ).rejects.toThrow("forced destination failure"); + const afterRollback = await db.sql< + { failed_at: Date | null; locked_by: string | null; attempts: number }[] + >`select failed_at, locked_by, attempts from pgconductor._private_executions where id = ${id}::uuid`; + expect(afterRollback[0]?.failed_at).toBeNull(); + expect(afterRollback[0]?.locked_by).toBe(lockedBy); + expect(afterRollback[0]?.attempts).toBe(1); + await db.sql.unsafe( + `drop trigger fail_dlq_insert on pgconductor.executions_dlq; drop function public.fail_dlq_insert();`, + ); + await db.client.returnExecutions({ + count: 1, + orchestratorId: lockedBy, + completed: [], + failed: [settlement], + released: [], + invokeChild: [], + taskKeys: new Set([execution.task_key]), + }); + const finalRows = await db.sql<{ source: string | null; source_failed: string }[]>` + select dead_letter_source_execution_id::text as source, (select failed_at is not null from pgconductor._private_executions where id = ${id}::uuid)::text as source_failed + from pgconductor._private_executions where queue = 'dlq' + `; + expect(Array.from(finalRows)).toEqual([{ source: id, source_failed: "true" }]); + }, 20000); +}); diff --git a/packages/pgconductor-js/tests/mocks/in-memory-database-client.ts b/packages/pgconductor-js/tests/mocks/in-memory-database-client.ts index 487bf85..73f11ce 100644 --- a/packages/pgconductor-js/tests/mocks/in-memory-database-client.ts +++ b/packages/pgconductor-js/tests/mocks/in-memory-database-client.ts @@ -60,6 +60,13 @@ interface StoredExecution { parent_step_key: string | null; created_at: Date; updated_at: Date; + failed_at: Date | null; + dead_letter_source_execution_id: string | null; + dead_letter_source_queue: string | null; + dead_letter_source_task_key: string | null; + dead_letter_error: string | null; + dead_letter_attempts: number | null; + dead_letter_failed_at: Date | null; } interface StoredStep { @@ -79,6 +86,8 @@ interface StoredTask { window_end: string | null; concurrency: number | null; group_concurrency: number | null; + dead_letter_queue: string | null; + dead_letter_task_key: string | null; } interface StoredCronSchedule { @@ -294,6 +303,8 @@ export class InMemoryDatabaseClient implements IDatabaseClient { window_end: taskSpec.window?.[1] || null, concurrency: taskSpec.concurrency || null, group_concurrency: taskSpec.groupConcurrency || null, + dead_letter_queue: taskSpec.deadLetterQueue || null, + dead_letter_task_key: taskSpec.deadLetterTaskKey || null, }; this.tasks.set(this.taskId(taskSpec.key, task.queue), task); } @@ -405,6 +416,12 @@ export class InMemoryDatabaseClient implements IDatabaseClient { dedupe_key: exec.dedupe_key || undefined, cron_expression: exec.cron_expression || undefined, group: exec.group, + dead_letter_source_execution_id: exec.dead_letter_source_execution_id, + dead_letter_source_queue: exec.dead_letter_source_queue, + dead_letter_source_task_key: exec.dead_letter_source_task_key, + dead_letter_error: exec.dead_letter_error, + dead_letter_attempts: exec.dead_letter_attempts, + dead_letter_failed_at: exec.dead_letter_failed_at, locked_by: exec.orchestrator_id || "", }); @@ -487,8 +504,11 @@ export class InMemoryDatabaseClient implements IDatabaseClient { if (exec.attempts >= maxAttempts) { // Permanently failed exec.state = "failed"; + exec.failed_at = now; + if (!exec.cancelled) this.deliverToDeadLetterQueue(exec, task, result.error, now); - // Fail parent if waiting + // Fail parent if waiting. A force-failed parent is itself terminal and + // follows its own DLQ and retention policy. if (exec.parent_execution_id) { const parent = this.executions.get(exec.parent_execution_id); if (parent && parent.waiting_on_execution_id === exec.id) { @@ -496,7 +516,14 @@ export class InMemoryDatabaseClient implements IDatabaseClient { parent.last_error = `Child execution failed: ${result.error}`; parent.waiting_on_execution_id = null; parent.waiting_step_key = null; - parent.waiting_timeout_at = null; + const parentTask = this.tasks.get(this.taskId(parent.task_key, parent.queue)); + if (!exec.cancelled) { + this.deliverToDeadLetterQueue(parent, parentTask, parent.last_error, now); + } + if (parentTask?.remove_on_fail_days != null) { + this.executions.delete(parent.id); + this.steps.delete(parent.id); + } } } @@ -535,10 +562,15 @@ export class InMemoryDatabaseClient implements IDatabaseClient { case "permanently_failed": { exec.state = "failed"; + exec.failed_at = now; exec.last_error = result.error; + + const task = this.tasks.get(this.taskId(exec.task_key, exec.queue)); + if (!exec.cancelled) this.deliverToDeadLetterQueue(exec, task, result.error, now); exec.orchestrator_id = null; - // Fail parent if waiting + // Fail parent if waiting. A force-failed parent is itself terminal and + // follows its own DLQ and retention policy. if (exec.parent_execution_id) { const parent = this.executions.get(exec.parent_execution_id); if (parent && parent.waiting_on_execution_id === exec.id) { @@ -547,10 +579,17 @@ export class InMemoryDatabaseClient implements IDatabaseClient { parent.waiting_on_execution_id = null; parent.waiting_step_key = null; parent.waiting_timeout_at = null; + const parentTask = this.tasks.get(this.taskId(parent.task_key, parent.queue)); + if (!exec.cancelled) { + this.deliverToDeadLetterQueue(parent, parentTask, parent.last_error, now); + } + if (parentTask?.remove_on_fail_days != null) { + this.executions.delete(parent.id); + this.steps.delete(parent.id); + } } } - const task = this.tasks.get(this.taskId(exec.task_key, exec.queue)); if (task && task.remove_on_fail_days != null) { this.executions.delete(exec.id); this.steps.delete(exec.id); @@ -700,6 +739,13 @@ export class InMemoryDatabaseClient implements IDatabaseClient { parent_step_key: spec.parent_step_key || null, created_at: now, updated_at: now, + failed_at: null, + dead_letter_source_execution_id: null, + dead_letter_source_queue: null, + dead_letter_source_task_key: null, + dead_letter_error: null, + dead_letter_attempts: null, + dead_letter_failed_at: null, }; this.executions.set(id, execution); @@ -1067,6 +1113,57 @@ export class InMemoryDatabaseClient implements IDatabaseClient { } // ============================================================================ + private deliverToDeadLetterQueue( + exec: StoredExecution, + task: StoredTask | undefined, + error: string, + now: Date, + ): void { + if (!task?.dead_letter_queue || exec.cancelled) return; + const destinationTaskKey = task.dead_letter_task_key || exec.task_key; + const duplicate = Array.from(this.executions.values()).some( + (destination) => + destination.dead_letter_source_execution_id === exec.id && + destination.queue === task.dead_letter_queue && + destination.task_key === destinationTaskKey, + ); + if (duplicate) return; + const id = this.generateId(); + this.executions.set(id, { + id, + task_key: destinationTaskKey, + queue: task.dead_letter_queue, + group: exec.group, + payload: structuredClone(exec.payload), + state: "pending", + run_at: now, + attempts: 0, + max_attempts: 3, + last_error: null, + result: null, + cancelled: false, + waiting_on_execution_id: null, + waiting_step_key: null, + waiting_timeout_at: null, + dedupe_key: null, + singleton_on: null, + cron_expression: null, + priority: 0, + orchestrator_id: null, + parent_execution_id: null, + parent_step_key: null, + created_at: now, + updated_at: now, + failed_at: null, + dead_letter_source_execution_id: exec.id, + dead_letter_source_queue: exec.queue, + dead_letter_source_task_key: exec.task_key, + dead_letter_error: error, + dead_letter_attempts: exec.attempts, + dead_letter_failed_at: now, + }); + } + // Helpers // ============================================================================ diff --git a/packages/pgconductor-js/tests/unit/dead-letter-queue-types.test.ts b/packages/pgconductor-js/tests/unit/dead-letter-queue-types.test.ts new file mode 100644 index 0000000..cc92147 --- /dev/null +++ b/packages/pgconductor-js/tests/unit/dead-letter-queue-types.test.ts @@ -0,0 +1,77 @@ +import { describe, expect, test } from "bun:test"; +import { z } from "zod"; +import { Conductor } from "../../src/conductor"; +import { TaskSchemas } from "../../src/schemas"; +import { defineTask } from "../../src/task-definition"; + +const sourceDefinition = defineTask({ + name: "source", + queue: "source-q", + payload: z.object({ value: z.string() }), +}); +const compatibleDefinition = defineTask({ + name: "compatible", + queue: "dlq", + payload: z.object({ value: z.string() }), +}); +const incompatibleDefinition = defineTask({ + name: "incompatible", + queue: "dlq", + payload: z.object({ other: z.number() }), +}); + +function conductor() { + return Conductor.create({ + sql: {} as any, + tasks: TaskSchemas.fromSchema([sourceDefinition, compatibleDefinition, incompatibleDefinition]), + context: {}, + }); +} + +describe("dead-letter task types and validation", () => { + test("accepts compatible targets and rejects incompatible payloads", () => { + const c = conductor(); + const compatible = c.createTask( + { name: "compatible", queue: "dlq" }, + { invocable: true }, + async () => {}, + ); + c.createTask( + { name: "source", queue: "source-q", deadLetter: { queue: "dlq", task: compatible } }, + { invocable: true }, + async () => {}, + ); + + const incompatible = c.createTask( + { name: "incompatible", queue: "dlq" }, + { invocable: true }, + async () => {}, + ); + if (false) + c.createTask( + // @ts-expect-error The DLQ handler must accept the source payload. + { name: "source", queue: "source-q", deadLetter: { queue: "dlq", task: incompatible } }, + { invocable: true }, + async () => {}, + ); + }); + + test("rejects a direct self-target but permits a cross-queue identity", () => { + const c = conductor(); + expect(() => + c.createTask( + { name: "source", queue: "source-q", deadLetter: { queue: "source-q" } }, + { invocable: true }, + async () => {}, + ), + ).toThrow("cannot dead-letter directly to itself"); + + expect(() => + c.createTask( + { name: "source", queue: "source-q", deadLetter: { queue: "other-q" } }, + { invocable: true }, + async () => {}, + ), + ).not.toThrow(); + }); +});