diff --git a/bun.lock b/bun.lock index e741119..aeb4ba2 100644 --- a/bun.lock +++ b/bun.lock @@ -24,6 +24,7 @@ }, "devDependencies": { "@opentelemetry/api": "catalog:", + "@opentelemetry/sdk-metrics": "catalog:", "@opentelemetry/sdk-trace-base": "catalog:", "@testcontainers/postgresql": "catalog:", "@types/bun": "catalog:", @@ -38,6 +39,7 @@ }, "catalog": { "@opentelemetry/api": "1.9.0", + "@opentelemetry/sdk-metrics": "1.30.1", "@opentelemetry/sdk-trace-base": "1.30.1", "@standard-schema/spec": "1.0.0", "@testcontainers/postgresql": "11.8.1", @@ -69,6 +71,8 @@ "@opentelemetry/resources": ["@opentelemetry/resources@1.30.1", "", { "dependencies": { "@opentelemetry/core": "1.30.1", "@opentelemetry/semantic-conventions": "1.28.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.0.0 <1.10.0" } }, "sha512-5UxZqiAgLYGFjS4s9qm5mBVo433u+dSPUFWVWXmLAD4wB65oMCoXaJP1KJa9DIYYMeHu3z4BZcStG3LC593cWA=="], + "@opentelemetry/sdk-metrics": ["@opentelemetry/sdk-metrics@1.30.1", "", { "dependencies": { "@opentelemetry/core": "1.30.1", "@opentelemetry/resources": "1.30.1" }, "peerDependencies": { "@opentelemetry/api": ">=1.3.0 <1.10.0" } }, "sha512-q9zcZ0Okl8jRgmy7eNW3Ku1XSgg3sDLa5evHZpCwjspw7E8Is4K/haRPDJrBcX3YSn/Y7gUvFnByNYEKQNbNog=="], + "@opentelemetry/sdk-trace-base": ["@opentelemetry/sdk-trace-base@1.30.1", "", { "dependencies": { "@opentelemetry/core": "1.30.1", "@opentelemetry/resources": "1.30.1", "@opentelemetry/semantic-conventions": "1.28.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.0.0 <1.10.0" } }, "sha512-jVPgBbH1gCy2Lb7X0AVQ8XAfgg0pJ4nvl8/IiQA6nxOsPvS+0zMJaFSs2ltXe0J6C8dqjcnpyqINDJmU30+uOg=="], "@opentelemetry/semantic-conventions": ["@opentelemetry/semantic-conventions@1.28.0", "", {}, "sha512-lp4qAiMTD4sNWW4DbKLBkfiMZ4jbAboJIGOQr5DvciMRI494OapieI9qiODpOt0XBr1LjIDy1xAGAnVs5supTA=="], diff --git a/docs/content/operations/opentelemetry.md b/docs/content/operations/opentelemetry.md new file mode 100644 index 0000000..d53c5fa --- /dev/null +++ b/docs/content/operations/opentelemetry.md @@ -0,0 +1,7 @@ +# OpenTelemetry + +Tracing and metrics are enabled by default and use the application's global OpenTelemetry API. A Node `NodeSDK` can therefore be installed before or after Conductor; the application owns exporters and shutdown. + +Set `telemetry: false` on the Conductor to opt out. Propagation is bounded W3C trace context only: payloads and baggage are never recorded or propagated. + +Producer, consumer, process, child, cron, event, wait-resume, and dead-letter boundaries preserve causal topology. Messaging attributes follow the current OpenTelemetry messaging convention (implementation status: experimental). Metrics use seconds and bounded queue/task/outcome dimensions. diff --git a/docs/zensical.toml b/docs/zensical.toml index 7f9174b..815f700 100644 --- a/docs/zensical.toml +++ b/docs/zensical.toml @@ -79,6 +79,7 @@ nav = [ { "Batching" = "task-execution/batching.md" }, ]}, { "Operations" = [ + { "OpenTelemetry" = "operations/opentelemetry.md" }, { "Horizontal Scaling" = "scaling/horizontal.md" }, { "Live Migrations" = "scaling/live-migrations.md" }, { "Maintenance Task" = "scaling/maintenance.md" }, diff --git a/migrations/0000000001_setup.sql b/migrations/0000000001_setup.sql index 4eb0f21..4881780 100644 --- a/migrations/0000000001_setup.sql +++ b/migrations/0000000001_setup.sql @@ -93,6 +93,7 @@ create table pgconductor._private_executions ( completed_at timestamptz, payload jsonb, trace_context jsonb, + trace_link_context jsonb, run_at timestamptz default pgconductor._private_current_time() not null, locked_at timestamptz, locked_by uuid, @@ -388,11 +389,13 @@ create or replace function pgconductor._private_register_worker( p_cron_schedules pgconductor.execution_spec[], p_event_subscriptions pgconductor.event_subscription_spec[] default array[]::pgconductor.event_subscription_spec[] ) -returns void +returns jsonb language plpgsql volatile set search_path to '' as $function$ +declare + v_cron_rows jsonb; begin -- Filter arrays are equality allowlists; an empty list is almost always a -- configuration mistake and must not silently match nothing. @@ -452,22 +455,24 @@ begin 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") - select - spec.task_key, - coalesce(spec.queue, 'default'), - coalesce(spec.payload, '{}'::jsonb), - coalesce(spec.run_at, pgconductor._private_current_time()), - spec.dedupe_key, - spec.cron_expression, - spec."group" - from unnest(p_cron_schedules) as spec - where spec.dedupe_key is not null - on conflict (task_key, dedupe_key, queue) do update set - payload = excluded.payload, - run_at = excluded.run_at, - cron_expression = excluded.cron_expression, - "group" = excluded."group"; + with inserted as ( + insert into pgconductor._private_executions (task_key, queue, payload, run_at, dedupe_key, cron_expression, "group", trace_context) + select spec.task_key, coalesce(spec.queue, 'default'), coalesce(spec.payload, '{}'::jsonb), + coalesce(spec.run_at, pgconductor._private_current_time()), spec.dedupe_key, + spec.cron_expression, spec."group", spec.trace_context + from unnest(p_cron_schedules) as spec + where spec.dedupe_key is not null + on conflict (task_key, dedupe_key, queue) do update set + payload = excluded.payload, run_at = excluded.run_at, + cron_expression = excluded.cron_expression, "group" = excluded."group", + trace_context = coalesce(excluded.trace_context, pgconductor._private_executions.trace_context) + returning id, task_key, queue, true as inserted + ) + select coalesce(jsonb_agg(jsonb_build_object( + 'id', id, 'task_key', task_key, 'queue', queue, + 'is_maintenance', task_key = 'pgconductor.maintenance', + 'inserted', inserted, 'authoritative', true + )), '[]'::jsonb) into v_cron_rows from inserted; -- step 4: clean up stale schedules for this queue -- delete future executions for schedules that no longer exist @@ -558,6 +563,7 @@ begin coalesce(array_to_string(source.column_names, ','), '') and target.filter is not distinct from source.filter ); + return v_cron_rows; end; $function$; diff --git a/migrations/0000000002_events.sql b/migrations/0000000002_events.sql index abe7032..4dcab1d 100644 --- a/migrations/0000000002_events.sql +++ b/migrations/0000000002_events.sql @@ -15,6 +15,7 @@ create table if not exists pgconductor._private_custom_events ( id uuid default pgconductor._private_portable_uuidv7() not null, event_key text not null, payload jsonb not null default '{}'::jsonb, + trace_context jsonb, event_position bigint not null default nextval('pgconductor._private_event_position_seq'), created_at timestamptz default pgconductor._private_current_time() not null, processed_at timestamptz, @@ -222,7 +223,7 @@ begin loop -- Look at persisted events regardless of processed status. This prevents -- concurrent task-event processors from changing wait delivery order. - select e.event_key, e.payload, e.event_position, e.created_at + select e.id, e.event_key, e.payload, e.trace_context, e.event_position, e.created_at into v_event from pgconductor._private_custom_events e where e.event_key = v_wait.event_key @@ -271,7 +272,8 @@ begin end if; update pgconductor._private_executions - set run_at = v_now, waiting_on_execution_id = null, waiting_step_key = null + set run_at = v_now, waiting_on_execution_id = null, waiting_step_key = null, + trace_link_context = case when v_has_event then v_event.trace_context else null end where id = v_wait.execution_id and queue = v_wait.queue; v_resolved := v_resolved + 1; end if; @@ -288,7 +290,7 @@ declare v_now timestamptz := pgconductor._private_current_time(); begin with candidates as materialized ( - select e.created_at, e.id, e.event_key, e.payload + select e.created_at, e.id, e.event_key, e.payload, e.trace_context from pgconductor._private_custom_events e where e.processed_at is null order by e.event_position, e.created_at, e.id @@ -298,7 +300,7 @@ begin select c.created_at as event_created_at, c.id as event_id, s.id as subscription_id, s.task_key, s.queue, pgconductor._private_extract_event_payload(s.payload_fields, c.payload) as selected_payload, - c.event_key + c.event_key, c.trace_context from candidates c join pgconductor._private_event_subscriptions s on s.event_key = c.event_key join pgconductor._private_tasks t on t.key = s.task_key and t.queue = s.queue @@ -324,11 +326,11 @@ begin returning event_created_at, event_id, subscription_id ), inserted_executions as ( insert into pgconductor._private_executions( - task_key, queue, payload, event_created_at, event_id, subscription_id + task_key, queue, payload, trace_context, event_created_at, event_id, subscription_id ) select m.task_key, m.queue, jsonb_build_object('event', m.event_key, 'payload', m.selected_payload), - d.event_created_at, d.event_id, d.subscription_id + m.trace_context, d.event_created_at, d.event_id, d.subscription_id from inserted_deliveries d join matches m using (event_created_at, event_id, subscription_id) on conflict (event_created_at, event_id, subscription_id, queue) @@ -397,13 +399,14 @@ $function$; create or replace function pgconductor.emit_event( p_event_key text, - p_payload jsonb default '{}'::jsonb + p_payload jsonb default '{}'::jsonb, + p_trace_context jsonb default null ) returns uuid language plpgsql volatile security definer set search_path to '' as $function$ declare v_id uuid; begin perform pg_advisory_xact_lock(hashtext('pgconductor:event-waits')); - insert into pgconductor._private_custom_events (event_key, payload) - values (p_event_key, p_payload) + insert into pgconductor._private_custom_events (event_key, payload, trace_context) + values (p_event_key, p_payload, p_trace_context) returning id into v_id; return v_id; end; @@ -578,7 +581,8 @@ create trigger sync_database_trigger create or replace function pgconductor.emit_event( p_event_key text, - p_payload jsonb default '{}'::jsonb + p_payload jsonb default '{}'::jsonb, + p_trace_context jsonb default null ) returns uuid language plpgsql @@ -589,8 +593,8 @@ as $_$ declare v_id uuid; begin perform pg_advisory_xact_lock(hashtext('pgconductor:event-waits')); - insert into pgconductor._private_custom_events (event_key, payload) - values (p_event_key, p_payload) + insert into pgconductor._private_custom_events (event_key, payload, trace_context) + values (p_event_key, p_payload, p_trace_context) returning id into v_id; return v_id; end; diff --git a/package.json b/package.json index bdc6c65..134a145 100644 --- a/package.json +++ b/package.json @@ -1,7 +1,9 @@ { "name": "pgconductor", "type": "module", - "workspaces": ["packages/pgconductor-js"], + "workspaces": [ + "packages/pgconductor-js" + ], "catalog": { "postgres": "3.4.7", "prettier": "3.4.2", @@ -13,6 +15,7 @@ "@standard-schema/spec": "1.0.0", "cron-parser": "5.4.0", "@opentelemetry/api": "1.9.0", + "@opentelemetry/sdk-metrics": "1.30.1", "@opentelemetry/sdk-trace-base": "1.30.1", "oxfmt": "0.15.0", "oxlint": "1.30.0", diff --git a/packages/pgconductor-js/package.json b/packages/pgconductor-js/package.json index b82d8ec..10892b2 100644 --- a/packages/pgconductor-js/package.json +++ b/packages/pgconductor-js/package.json @@ -42,6 +42,7 @@ }, "devDependencies": { "@opentelemetry/api": "catalog:", + "@opentelemetry/sdk-metrics": "catalog:", "@opentelemetry/sdk-trace-base": "catalog:", "@testcontainers/postgresql": "catalog:", "@types/bun": "catalog:", diff --git a/packages/pgconductor-js/src/conductor.ts b/packages/pgconductor-js/src/conductor.ts index 9412f9b..7cbc4d3 100644 --- a/packages/pgconductor-js/src/conductor.ts +++ b/packages/pgconductor-js/src/conductor.ts @@ -34,6 +34,8 @@ import { setSpanError, startSpan, runWithSpan, + recordSent, + recordOperationDuration, } from "./telemetry"; import { SpanKind } from "@opentelemetry/api"; import type { @@ -324,7 +326,7 @@ export class Conductor< : startSpan( `send ${queue}`, SpanKind.PRODUCER, - messagingAttributes(taskName, queue, "send"), + messagingAttributes(taskName, queue, "send", undefined, payloadOrItems.length), ); const carrier = producer ? carrierForContext(contextForSpan(producer)) : null; const specs = payloadOrItems.map((item) => ({ @@ -341,7 +343,12 @@ export class Conductor< trace_context: carrier, })); try { + const started = performance.now(); const ids = await runWithSpan(producer, () => this.db.invokeBatch(specs)); + if (this.options.telemetry !== false && ids.length > 0) { + recordSent(queue, ids.length, taskName); + recordOperationDuration(queue, performance.now() - started, "send", taskName); + } setSpanAttribute(producer, "messaging.batch.message_count", payloadOrItems.length); return ids; } catch (error) { @@ -361,6 +368,7 @@ export class Conductor< messagingAttributes(taskName, queue, "send"), ); const carrier = producer ? carrierForContext(contextForSpan(producer)) : null; + const started = performance.now(); try { const id = await runWithSpan(producer, () => this.db.invoke({ @@ -371,7 +379,12 @@ export class Conductor< trace_context: carrier, }), ); - if (id) setSpanAttribute(producer, "messaging.message.id", id); + if (id) { + if (this.options.telemetry !== false) recordSent(queue, 1, taskName); + setSpanAttribute(producer, "messaging.message.id", id); + } + if (this.options.telemetry !== false && id) + recordOperationDuration(queue, performance.now() - started, "send", taskName); return id; } catch (error) { setSpanError(producer, error); @@ -405,10 +418,42 @@ export class Conductor< payload = ((result as any)?.value ?? payload) as InferEventPayload; } - return this.db.emitEvent({ - eventKey: event, - payload: payload as any, - }); + const started = performance.now(); + const producer = + this.options.telemetry === false + ? null + : startSpan( + `send event ${String(event)}`, + SpanKind.PRODUCER, + messagingAttributes( + undefined, + String(event), + "send", + undefined, + undefined, + String(event), + ), + ); + try { + const id = await runWithSpan(producer, () => + this.db.emitEvent({ + eventKey: event, + payload: payload as any, + trace_context: producer ? carrierForContext(contextForSpan(producer)) : null, + }), + ); + if (this.options.telemetry !== false) { + recordSent(String(event), 1); + recordOperationDuration(String(event), performance.now() - started, "send"); + } + setSpanAttribute(producer, "messaging.message.id", id); + return id; + } catch (error) { + setSpanError(producer, error); + throw error; + } finally { + endSpan(producer); + } } async cancel(executionId: string, options?: { reason?: string }): Promise { diff --git a/packages/pgconductor-js/src/database-client.ts b/packages/pgconductor-js/src/database-client.ts index 3c22472..5d3b2fd 100644 --- a/packages/pgconductor-js/src/database-client.ts +++ b/packages/pgconductor-js/src/database-client.ts @@ -41,6 +41,7 @@ export interface ExecutionSpec { parent_step_key?: string | null; parent_timeout_ms?: number | null; trace_context?: TraceContextCarrier | null; + trace_link_context?: TraceContextCarrier | null; } export interface TaskSpec { @@ -73,6 +74,7 @@ export interface Execution { waiting_on_execution_id: string | null; waiting_step_key: string | null; locked_by: string; + attempts?: number; cancelled: boolean; last_error: string | null; dedupe_key?: string | null; @@ -85,6 +87,13 @@ export interface Execution { dead_letter_attempts?: number | null; dead_letter_failed_at?: Date | null; trace_context?: TraceContextCarrier | null; + /** A bounded event carrier linked when a durable wait resumes. */ + trace_link_context?: TraceContextCarrier | null; + parent_execution_id?: string | null; + parent_queue?: string | null; + parent_task_key?: string | null; + parent_dead_letter_queue?: string | null; + parent_dead_letter_task_key?: string | null; } // todo: move all of this to query-builder too or create new types.ts file @@ -96,6 +105,13 @@ export type ExecutionResult = | ExecutionPermamentlyFailed | ExecutionInvokeChild; +export type SettlementOutcome = { + queue: string; + task_key: string; + outcome: "retry" | "permanent_failure" | "cancellation" | "dead_letter"; + count: number; +}; + export type GroupedExecutionResults = { count: number; orchestratorId: string; @@ -106,6 +122,38 @@ export type GroupedExecutionResults = { taskKeys: Set; }; +export type DeadLetterDelivery = { + sourceExecutionId: string; + destinationExecutionId: string; + destinationQueue: string; + destinationTaskKey: string; +}; + +export type ReturnExecutionsRow = { + queue: string | null; + task_key: string | null; + outcome: SettlementOutcome["outcome"] | null; + count: number | null; + source_execution_id: string | null; + destination_execution_id: string | null; + destination_queue: string | null; + destination_task_key: string | null; +}; + +export type ReturnExecutionsResult = { + outcomes: SettlementOutcome[]; + deliveries: DeadLetterDelivery[]; +}; + +export type CronRegistration = { + id: string; + task_key: string; + queue: string; + is_maintenance: boolean; + inserted: boolean; + authoritative: boolean; +}; + export interface ExecutionCompleted { execution_id: string; queue: string; @@ -116,6 +164,11 @@ export interface ExecutionCompleted { } export interface ExecutionFailed { + /** Failed executions are retryable and therefore cannot be cancellations. */ + cancelled: false; + /** Carrier for the failed process / DLQ producer span. */ + trace_context?: TraceContextCarrier | null; + dead_letter_trace_contexts?: Record; execution_id: string; queue: string; orchestrator_id: string; @@ -135,6 +188,12 @@ export interface ExecutionReleased { } export interface ExecutionPermamentlyFailed { + /** True when this terminal result was caused by cancellation. */ + cancelled: boolean; + /** Carrier for each terminal DLQ producer span, keyed by source execution ID. */ + dead_letter_trace_contexts?: Record; + /** Carrier for the failed process / DLQ producer span. */ + trace_context?: TraceContextCarrier | null; execution_id: string; queue: string; orchestrator_id: string; @@ -144,6 +203,8 @@ export interface ExecutionPermamentlyFailed { } export interface ExecutionInvokeChild { + /** Carrier for the producer span created when the child is inserted. */ + trace_context?: TraceContextCarrier | null; group?: string | null; execution_id: string; queue: string; @@ -491,14 +552,33 @@ export class DatabaseClient { async returnExecutions( grouped: GroupedExecutionResults, opts?: QueryMethodOptions, - ): Promise { + ): Promise { const query = this.builder.buildReturnExecutions(grouped); - if (!query) { - return; - } + if (!query) return { outcomes: [], deliveries: [] }; - await this.query(() => query, { label: "returnExecutions", ...opts }); + const rows = await this.query(() => query, { label: "returnExecutions", ...opts }); + const outcomes = rows.flatMap((row): SettlementOutcome[] => + row.outcome !== null && row.queue !== null && row.task_key !== null && row.count !== null + ? [{ queue: row.queue, task_key: row.task_key, outcome: row.outcome, count: row.count }] + : [], + ); + const deliveries = rows.flatMap((row): DeadLetterDelivery[] => + row.source_execution_id !== null && + row.destination_execution_id !== null && + row.destination_queue !== null && + row.destination_task_key !== null + ? [ + { + sourceExecutionId: row.source_execution_id, + destinationExecutionId: row.destination_execution_id, + destinationQueue: row.destination_queue, + destinationTaskKey: row.destination_task_key, + }, + ] + : [], + ); + return { outcomes, deliveries }; } async removeExecutions(args: RemoveExecutionsArgs, opts?: QueryMethodOptions): Promise { @@ -522,11 +602,15 @@ export class DatabaseClient { return Number(result[0]?.deleted_count || 0) >= args.batchSize; } - async registerWorker(args: RegisterWorkerArgs, opts?: QueryMethodOptions): Promise { - await this.query(() => this.builder.buildRegisterWorker(args), { + async registerWorker( + args: RegisterWorkerArgs, + opts?: QueryMethodOptions, + ): Promise { + const result = await this.query(() => this.builder.buildRegisterWorker(args), { label: "registerWorker", ...opts, }); + return result[0]?.cron_rows ?? []; } async scheduleCronExecution( @@ -661,3 +745,7 @@ export class DatabaseClient { }); } } + +export type DatabaseClientLike = { + [K in keyof DatabaseClient as DatabaseClient[K] extends Function ? K : never]: DatabaseClient[K]; +}; diff --git a/packages/pgconductor-js/src/generated/sql.ts b/packages/pgconductor-js/src/generated/sql.ts index 85b49e7..7a8c537 100644 --- a/packages/pgconductor-js/src/generated/sql.ts +++ b/packages/pgconductor-js/src/generated/sql.ts @@ -109,6 +109,7 @@ create table pgconductor._private_executions ( completed_at timestamptz, payload jsonb, trace_context jsonb, + trace_link_context jsonb, run_at timestamptz default pgconductor._private_current_time() not null, locked_at timestamptz, locked_by uuid, @@ -404,11 +405,13 @@ create or replace function pgconductor._private_register_worker( p_cron_schedules pgconductor.execution_spec[], p_event_subscriptions pgconductor.event_subscription_spec[] default array[]::pgconductor.event_subscription_spec[] ) -returns void +returns jsonb language plpgsql volatile set search_path to '' as $function$ +declare + v_cron_rows jsonb; begin -- Filter arrays are equality allowlists; an empty list is almost always a -- configuration mistake and must not silently match nothing. @@ -468,22 +471,24 @@ begin 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") - select - spec.task_key, - coalesce(spec.queue, 'default'), - coalesce(spec.payload, '{}'::jsonb), - coalesce(spec.run_at, pgconductor._private_current_time()), - spec.dedupe_key, - spec.cron_expression, - spec."group" - from unnest(p_cron_schedules) as spec - where spec.dedupe_key is not null - on conflict (task_key, dedupe_key, queue) do update set - payload = excluded.payload, - run_at = excluded.run_at, - cron_expression = excluded.cron_expression, - "group" = excluded."group"; + with inserted as ( + insert into pgconductor._private_executions (task_key, queue, payload, run_at, dedupe_key, cron_expression, "group", trace_context) + select spec.task_key, coalesce(spec.queue, 'default'), coalesce(spec.payload, '{}'::jsonb), + coalesce(spec.run_at, pgconductor._private_current_time()), spec.dedupe_key, + spec.cron_expression, spec."group", spec.trace_context + from unnest(p_cron_schedules) as spec + where spec.dedupe_key is not null + on conflict (task_key, dedupe_key, queue) do update set + payload = excluded.payload, run_at = excluded.run_at, + cron_expression = excluded.cron_expression, "group" = excluded."group", + trace_context = coalesce(excluded.trace_context, pgconductor._private_executions.trace_context) + returning id, task_key, queue, true as inserted + ) + select coalesce(jsonb_agg(jsonb_build_object( + 'id', id, 'task_key', task_key, 'queue', queue, + 'is_maintenance', task_key = 'pgconductor.maintenance', + 'inserted', inserted, 'authoritative', true + )), '[]'::jsonb) into v_cron_rows from inserted; -- step 4: clean up stale schedules for this queue -- delete future executions for schedules that no longer exist @@ -574,6 +579,7 @@ begin coalesce(array_to_string(source.column_names, ','), '') and target.filter is not distinct from source.filter ); + return v_cron_rows; end; $function$; @@ -987,6 +993,7 @@ create table if not exists pgconductor._private_custom_events ( id uuid default pgconductor._private_portable_uuidv7() not null, event_key text not null, payload jsonb not null default '{}'::jsonb, + trace_context jsonb, event_position bigint not null default nextval('pgconductor._private_event_position_seq'), created_at timestamptz default pgconductor._private_current_time() not null, processed_at timestamptz, @@ -1194,7 +1201,7 @@ begin loop -- Look at persisted events regardless of processed status. This prevents -- concurrent task-event processors from changing wait delivery order. - select e.event_key, e.payload, e.event_position, e.created_at + select e.id, e.event_key, e.payload, e.trace_context, e.event_position, e.created_at into v_event from pgconductor._private_custom_events e where e.event_key = v_wait.event_key @@ -1243,7 +1250,8 @@ begin end if; update pgconductor._private_executions - set run_at = v_now, waiting_on_execution_id = null, waiting_step_key = null + set run_at = v_now, waiting_on_execution_id = null, waiting_step_key = null, + trace_link_context = case when v_has_event then v_event.trace_context else null end where id = v_wait.execution_id and queue = v_wait.queue; v_resolved := v_resolved + 1; end if; @@ -1260,7 +1268,7 @@ declare v_now timestamptz := pgconductor._private_current_time(); begin with candidates as materialized ( - select e.created_at, e.id, e.event_key, e.payload + select e.created_at, e.id, e.event_key, e.payload, e.trace_context from pgconductor._private_custom_events e where e.processed_at is null order by e.event_position, e.created_at, e.id @@ -1270,7 +1278,7 @@ begin select c.created_at as event_created_at, c.id as event_id, s.id as subscription_id, s.task_key, s.queue, pgconductor._private_extract_event_payload(s.payload_fields, c.payload) as selected_payload, - c.event_key + c.event_key, c.trace_context from candidates c join pgconductor._private_event_subscriptions s on s.event_key = c.event_key join pgconductor._private_tasks t on t.key = s.task_key and t.queue = s.queue @@ -1296,11 +1304,11 @@ begin returning event_created_at, event_id, subscription_id ), inserted_executions as ( insert into pgconductor._private_executions( - task_key, queue, payload, event_created_at, event_id, subscription_id + task_key, queue, payload, trace_context, event_created_at, event_id, subscription_id ) select m.task_key, m.queue, jsonb_build_object('event', m.event_key, 'payload', m.selected_payload), - d.event_created_at, d.event_id, d.subscription_id + m.trace_context, d.event_created_at, d.event_id, d.subscription_id from inserted_deliveries d join matches m using (event_created_at, event_id, subscription_id) on conflict (event_created_at, event_id, subscription_id, queue) @@ -1369,13 +1377,14 @@ $function$; create or replace function pgconductor.emit_event( p_event_key text, - p_payload jsonb default '{}'::jsonb + p_payload jsonb default '{}'::jsonb, + p_trace_context jsonb default null ) returns uuid language plpgsql volatile security definer set search_path to '' as $function$ declare v_id uuid; begin perform pg_advisory_xact_lock(hashtext('pgconductor:event-waits')); - insert into pgconductor._private_custom_events (event_key, payload) - values (p_event_key, p_payload) + insert into pgconductor._private_custom_events (event_key, payload, trace_context) + values (p_event_key, p_payload, p_trace_context) returning id into v_id; return v_id; end; @@ -1550,7 +1559,8 @@ create trigger sync_database_trigger create or replace function pgconductor.emit_event( p_event_key text, - p_payload jsonb default '{}'::jsonb + p_payload jsonb default '{}'::jsonb, + p_trace_context jsonb default null ) returns uuid language plpgsql @@ -1561,8 +1571,8 @@ as $_$ declare v_id uuid; begin perform pg_advisory_xact_lock(hashtext('pgconductor:event-waits')); - insert into pgconductor._private_custom_events (event_key, payload) - values (p_event_key, p_payload) + insert into pgconductor._private_custom_events (event_key, payload, trace_context) + values (p_event_key, p_payload, p_trace_context) returning id into v_id; return v_id; end; diff --git a/packages/pgconductor-js/src/maintenance-task.ts b/packages/pgconductor-js/src/maintenance-task.ts index c41be59..6fd2431 100644 --- a/packages/pgconductor-js/src/maintenance-task.ts +++ b/packages/pgconductor-js/src/maintenance-task.ts @@ -1,5 +1,5 @@ import { Task, type AnyTask } from "./task"; -import type { DatabaseClient } from "./database-client"; +import type { DatabaseClientLike } from "./database-client"; import crypto from "crypto"; import type { TaskContext } from "./task-context"; @@ -24,7 +24,7 @@ export const createMaintenanceTask = (queue: Q Queue, object, void, - { db: DatabaseClient; tasks: Map } & TaskContext, + { db: DatabaseClientLike; tasks: Map } & TaskContext, { name: "pgconductor.maintenance" } >( { diff --git a/packages/pgconductor-js/src/query-builder.ts b/packages/pgconductor-js/src/query-builder.ts index b4f4b6a..3198c32 100644 --- a/packages/pgconductor-js/src/query-builder.ts +++ b/packages/pgconductor-js/src/query-builder.ts @@ -1,5 +1,9 @@ import type { PendingQuery, Row, RowList, Sql } from "postgres"; -import type { GroupedExecutionResults } from "./database-client"; +import type { + CronRegistration, + GroupedExecutionResults, + ReturnExecutionsRow, +} from "./database-client"; import { boundedCarrier } from "./telemetry"; import * as assert from "./lib/assert"; import type { @@ -93,6 +97,7 @@ export type ClearWaitingStateArgs = { export type EmitEventArgs = { eventKey: string; payload?: JsonValue; + trace_context?: import("./internal-types").TraceContextCarrier | null; }; export type RegisterEventWaitArgs = { @@ -341,7 +346,7 @@ export class QueryBuilder { -- Keep a bounded candidate pool so SKIP LOCKED can backfill a batch. limit greatest(${batchSize}::integer * 4, ${batchSize}::integer) ), locked_candidates as ( - select e.id + select e.id, e.trace_link_context from pgconductor._private_executions e join eligible c on c.id = e.id where e.queue = ${queueName}::text @@ -353,25 +358,37 @@ export class QueryBuilder { set attempts = e.attempts + 1, locked_by = ${orchestratorId}::uuid, - locked_at = pgconductor._private_current_time() + locked_at = pgconductor._private_current_time(), + trace_link_context = null from locked_candidates c 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.trace_context, e.dead_letter_source_execution_id, + e.waiting_step_key, e.attempts, e.cancelled, e.last_error, e.dedupe_key, + e.cron_expression, + e.locked_by, e."group", e.trace_context, c.trace_link_context, + 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, e.priority, e.run_at, e.created_at + e.dead_letter_attempts, e.dead_letter_failed_at, e.parent_execution_id, + e.priority, e.run_at, e.created_at ) - select id, task_key, queue, payload, waiting_on_execution_id, waiting_step_key, - cancelled, last_error, dedupe_key, cron_expression, locked_by, "group", trace_context, - 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 + select c.id, c.task_key, c.queue, c.payload, c.waiting_on_execution_id, + c.waiting_step_key, c.attempts, c.cancelled, c.last_error, c.dedupe_key, + c.cron_expression, c.locked_by, c."group", c.trace_context, c.trace_link_context, + c.dead_letter_source_execution_id, c.dead_letter_source_queue, + c.dead_letter_source_task_key, c.dead_letter_error, c.dead_letter_attempts, + c.dead_letter_failed_at, c.parent_execution_id, p.queue as parent_queue, + p.task_key as parent_task_key, pt.dead_letter_queue as parent_dead_letter_queue, + pt.dead_letter_task_key as parent_dead_letter_task_key + from claimed c + left join pgconductor._private_executions p on p.id = c.parent_execution_id + left join pgconductor._private_tasks pt on pt.queue = p.queue and pt.key = p.task_key + order by c.priority asc, c.run_at asc, c.created_at asc, c.id asc `; } - buildReturnExecutions(grouped: GroupedExecutionResults): PendingQuery | null { + buildReturnExecutions( + grouped: GroupedExecutionResults, + ): PendingQuery | null { const allResults = [ ...grouped.completed, ...grouped.failed, @@ -381,7 +398,7 @@ export class QueryBuilder { if (allResults.length === 0) return null; - const ctes: PendingQuery[] = []; + const ctes: PendingQuery>[] = []; ctes.push(this.sql`now_ts as (select pgconductor._private_current_time() as ts)`); ctes.push(this.sql`result_data as ( select * from jsonb_to_recordset(${this.sql.json(JSON.parse(JSON.stringify(allResults)))}::jsonb) @@ -390,7 +407,7 @@ export class QueryBuilder { orchestrator_id uuid, result jsonb, error text, reschedule_in_ms text, step_key text, timeout_ms text, child_task_name text, child_task_queue text, child_payload jsonb, - "group" text + "group" text, trace_context jsonb, dead_letter_trace_contexts jsonb ) )`); // Lock the claimed rows for the whole statement. This prevents recovery or a @@ -418,10 +435,10 @@ export class QueryBuilder { ctes.push(this.sql`failed_results as ( select * from valid_results where status in ('failed', 'permanently_failed') - or (status = 'completed' and execution_cancelled) + or (status in ('completed', 'released') and execution_cancelled) )`); ctes.push(this.sql`released_results as ( - select * from valid_results where status = 'released' + select * from valid_results where status = 'released' and not execution_cancelled )`); ctes.push(this.sql`invoke_child_data as ( select * from valid_results where status = 'invoke_child' @@ -488,12 +505,12 @@ export class QueryBuilder { and tc.key = r.task_key and tc.queue = r.queue and (tc.remove_on_complete_days is null or tc.remove_on_complete_days != 0) and not exists (select 1 from orphaned_children oc where oc.id = e.id) - returning e.id + returning e.id, e.queue )`); ctes.push(this.sql`permanently_failed_children as materialized ( select r.execution_id, r.queue, r.task_key, r.orchestrator_id, - r.execution_cancelled, + r.trace_context, r.dead_letter_trace_contexts, r.execution_cancelled, coalesce(r.error, r.execution_last_error, 'unknown error') as child_error, e."group" as execution_group, e.attempts as execution_attempts, @@ -507,7 +524,7 @@ 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, + p.dead_letter_trace_contexts, p.execution_cancelled as child_cancelled, parent.id as parent_id, parent.queue as parent_queue, parent.task_key as parent_task_key, parent."group" as parent_group, parent.payload as parent_payload, parent.attempts as parent_attempts, @@ -522,6 +539,7 @@ export class QueryBuilder { )`); ctes.push(this.sql`terminal_failures as materialized ( select p.execution_id, p.queue, p.task_key, p.execution_group, e.payload, + p.dead_letter_trace_contexts -> p.execution_id::text as dead_letter_trace_context, p.child_error as failure_error, p.execution_attempts as failure_attempts, p.execution_cancelled, tc.dead_letter_queue, tc.dead_letter_task_key from permanently_failed_children p @@ -530,6 +548,7 @@ export class QueryBuilder { join task_configs tc on tc.key = p.task_key and tc.queue = p.queue union all select p.parent_id, p.parent_queue, p.parent_task_key, p.parent_group, p.parent_payload, + p.dead_letter_trace_contexts -> p.parent_id::text as dead_letter_trace_context, 'Child execution failed: ' || p.child_error, p.parent_attempts, p.child_cancelled, pt.dead_letter_queue, pt.dead_letter_task_key from failed_parent_targets p @@ -541,14 +560,14 @@ export class QueryBuilder { 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 + dead_letter_attempts, dead_letter_failed_at, trace_context ) 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 + p.failure_attempts, nt.ts, p.dead_letter_trace_context from terminal_failures p cross join now_ts nt where p.dead_letter_queue is not null @@ -556,7 +575,7 @@ export class QueryBuilder { 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 + returning id, dead_letter_source_execution_id, queue, task_key )`); ctes.push(this.sql`failed_updates as ( select p.execution_id as target_id, p.queue, p.child_error, true as is_child @@ -575,8 +594,15 @@ export class QueryBuilder { and e.locked_by = p.orchestrator_id and p.should_remove is true and ( - not exists (select 1 from task_configs tc where tc.key = p.task_key and tc.queue = p.queue and tc.dead_letter_queue is not null) - or exists (select 1 from dead_lettered d where d.dead_letter_source_execution_id = p.execution_id) + not exists ( + select 1 from task_configs tc + where tc.key = p.task_key and tc.queue = p.queue + and tc.dead_letter_queue is not null + ) + or exists ( + select 1 from dead_lettered d + where d.dead_letter_source_execution_id = p.execution_id + ) ) ) or exists ( @@ -584,8 +610,15 @@ export class QueryBuilder { where e.id = p.parent_id and e.queue = p.parent_queue and p.parent_should_remove is true and ( - not exists (select 1 from pgconductor._private_tasks pt where pt.key = p.parent_task_key and pt.queue = p.parent_queue and pt.dead_letter_queue is not null) - or exists (select 1 from dead_lettered d where d.dead_letter_source_execution_id = p.parent_id) + not exists ( + select 1 from pgconductor._private_tasks pt + where pt.key = p.parent_task_key and pt.queue = p.parent_queue + and pt.dead_letter_queue is not null + ) + or exists ( + select 1 from dead_lettered d + where d.dead_letter_source_execution_id = p.parent_id + ) ) ) returning e.id @@ -599,8 +632,9 @@ export class QueryBuilder { locked_by = null, locked_at = null from now_ts nt, failed_updates f where e.id = f.target_id and e.queue = f.queue - returning e.id + returning e.id, e.queue )`); + ctes.push(this.sql`retried as ( update pgconductor._private_executions e set last_error = coalesce(r.error, 'unknown error'), @@ -614,7 +648,7 @@ export class QueryBuilder { and r.status <> 'permanently_failed' and not r.execution_cancelled and e.attempts < tc.max_attempts - returning e.id + returning e.id, e.queue, e.task_key )`); ctes.push(this.sql`released_steps as ( @@ -639,8 +673,8 @@ export class QueryBuilder { )`); ctes.push(this.sql`inserted_children as ( - insert into pgconductor._private_executions (id, task_key, queue, payload, run_at, parent_execution_id, "group") - select pgconductor._private_portable_uuidv7(), r.child_task_name, r.child_task_queue, r.child_payload, nt.ts, r.execution_id, r."group" + insert into pgconductor._private_executions (id, task_key, queue, payload, run_at, parent_execution_id, "group", trace_context) + select pgconductor._private_portable_uuidv7(), r.child_task_name, r.child_task_queue, r.child_payload, nt.ts, r.execution_id, r."group", r.trace_context from invoke_child_data r, now_ts nt where exists ( select 1 from pgconductor._private_executions parent @@ -664,8 +698,32 @@ export class QueryBuilder { returning e.id )`); + ctes.push(this.sql`settlement_outcomes as ( + select p.queue, p.task_key, + case when p.execution_cancelled then 'cancellation' else 'permanent_failure' end as outcome + from permanently_failed_children p + union all + select p.parent_queue, p.parent_task_key, + case when p.child_cancelled then 'cancellation' else 'permanent_failure' end + from failed_parent_targets p + union all + select queue, task_key, 'retry' from retried + union all + select t.queue, t.task_key, 'dead_letter' + from dead_lettered d + join terminal_failures t on t.execution_id = d.dead_letter_source_execution_id + )`); + const combined = ctes.reduce((acc, cte, i) => (i === 0 ? cte : this.sql`${acc}, ${cte}`)); - return this.sql<[{ result: number }]>`with ${combined} select 1 as result`; + return this.sql`with ${combined} + select queue, task_key, outcome, count(*)::integer as count, + null::uuid as source_execution_id, null::uuid as destination_execution_id, + null::text as destination_queue, null::text as destination_task_key + from settlement_outcomes group by queue, task_key, outcome + union all + select null::text, null::text, null::text, null::integer, + d.dead_letter_source_execution_id, d.id, d.queue, d.task_key + from dead_lettered d`; } buildRemoveExecutions({ queueName, @@ -710,7 +768,7 @@ export class QueryBuilder { taskSpecs, cronSchedules, eventSubscriptions, - }: RegisterWorkerArgs): PendingQuery> { + }: RegisterWorkerArgs): PendingQuery<{ cron_rows: CronRegistration[] }[]> { const taskSpecRows = taskSpecs.map((spec) => ({ key: spec.key, queue: spec.queue || null, @@ -737,6 +795,7 @@ export class QueryBuilder { cron_expression: spec.cron_expression, priority: spec.priority || null, group: spec.group || null, + trace_context: boundedCarrier(spec.trace_context), }; }); @@ -753,7 +812,7 @@ export class QueryBuilder { filter: spec.filter, })); - return this.sql` + return this.sql<{ cron_rows: CronRegistration[] }[]>` select pgconductor._private_register_worker( p_queue_name := ${queueName}::text, p_task_specs := array( @@ -765,7 +824,7 @@ export class QueryBuilder { p_event_subscriptions := array( select json_populate_recordset(null::pgconductor.event_subscription_spec, ${this.sql.json(eventSubscriptionRows)}::json)::pgconductor.event_subscription_spec ) - ) + ) as cron_rows `; } @@ -834,6 +893,7 @@ export class QueryBuilder { p_dedupe_next_slot := false::boolean, p_cron_expression := ${cronExpression}::text, p_priority := ${spec.priority || 0}::integer, + p_trace_context := ${boundedCarrier(spec.trace_context) ? this.sql.json(boundedCarrier(spec.trace_context)) : null}::jsonb, p_group := ${spec.group || null}::text ) `; @@ -1028,6 +1088,7 @@ export class QueryBuilder { ) update pgconductor._private_executions e set waiting_on_execution_id = null, waiting_step_key = ${stepKey}::text, + trace_link_context = null, run_at = 'infinity'::timestamptz, locked_by = null, locked_at = null from claimed c @@ -1069,7 +1130,7 @@ export class QueryBuilder { and ci.child_locked_by is null -- not currently executing and e.completed_at is null and e.failed_at is null - returning e.id + returning e.id, e.queue ), -- Signal executing (locked) children to cancel signaled_executing_child as ( @@ -1112,11 +1173,16 @@ export class QueryBuilder { `; } - buildEmitEvent({ eventKey, payload }: EmitEventArgs): PendingQuery<{ id: string }[]> { + buildEmitEvent({ + eventKey, + payload, + trace_context, + }: EmitEventArgs): PendingQuery<{ id: string }[]> { return this.sql<{ id: string }[]>` select pgconductor.emit_event( ${eventKey}::text, - ${this.sql.json(payload || {})}::jsonb + ${this.sql.json(payload || {})}::jsonb, + ${boundedCarrier(trace_context) ? this.sql.json(boundedCarrier(trace_context)) : null}::jsonb ) as id `; } diff --git a/packages/pgconductor-js/src/task-context.ts b/packages/pgconductor-js/src/task-context.ts index bc2f3f5..6206f98 100644 --- a/packages/pgconductor-js/src/task-context.ts +++ b/packages/pgconductor-js/src/task-context.ts @@ -1,6 +1,6 @@ import CronExpressionParser from "cron-parser"; import type { - DatabaseClient, + DatabaseClientLike, JsonValue, Execution, Payload, @@ -19,7 +19,19 @@ import { WindowChecker } from "./lib/window-checker"; import { TypedAbortController } from "./lib/typed-abort-controller"; import { parseDuration, type DurationInput } from "./lib/duration"; import { SpanKind } from "@opentelemetry/api"; -import { endSpan, stepAttributes, runWithSpan, setSpanError, startSpan } from "./telemetry"; +import { + endSpan, + stepAttributes, + runWithSpan, + setSpanError, + startSpan, + carrierForContext, + contextForSpan, + messagingAttributes, + setSpanAttribute, + recordOperationDuration, + recordSent, +} from "./telemetry"; import type { EventDefinition, EventName, @@ -95,7 +107,7 @@ export class WaitForEventTimeoutError extends Error { export type TaskContextOptions = { abortController: TypedAbortController; - db: DatabaseClient; + db: DatabaseClientLike; execution: Execution; logger: Logger; window?: [string, string]; @@ -433,21 +445,45 @@ export class TaskContext< const nextTimestamp = interval.next().toDate(); const queue = task.queue || "default"; - await this.opts.db.scheduleCronExecution( - { - spec: { - task_key: task.name, - queue, - payload, - run_at: nextTimestamp, - cron_expression: options.cron, - priority: options.priority || null, - group: options.group || null, - }, - scheduleName, - }, - { signal: this.signal }, - ); + const producer = + this.opts.telemetry === false + ? null + : startSpan( + `send ${queue}`, + SpanKind.PRODUCER, + messagingAttributes(task.name, queue, "send"), + ); + const started = performance.now(); + try { + const id = await runWithSpan(producer, () => + this.opts.db.scheduleCronExecution( + { + spec: { + task_key: task.name, + queue, + payload, + run_at: nextTimestamp, + cron_expression: options.cron, + priority: options.priority || null, + group: options.group || null, + trace_context: producer ? carrierForContext(contextForSpan(producer)) : null, + }, + scheduleName, + }, + { signal: this.signal }, + ), + ); + if (id && producer) { + setSpanAttribute(producer, "messaging.message.id", id); + recordSent(queue, 1, task.name); + recordOperationDuration(queue, performance.now() - started, "send", task.name); + } + } catch (error) { + setSpanError(producer, error); + throw error; + } finally { + endSpan(producer); + } } async unschedule, TQueue extends string = "default">( @@ -497,13 +533,40 @@ export class TaskContext< TName extends EventName, TDef extends FindEventByIdentifier = FindEventByIdentifier, >(event: TName, payload: InferEventPayload): Promise { - return this.opts.db.emitEvent( - { - eventKey: event, - payload: payload as any, - }, - { signal: this.signal }, - ); + const producer = + this.opts.telemetry === false + ? null + : startSpan(`send event ${String(event)}`, SpanKind.PRODUCER, { + "messaging.system": "postgres_conductor", + "messaging.destination.name": String(event), + "messaging.operation.name": "send", + "messaging.operation.type": "send", + "pgconductor.event.name": String(event), + }); + const started = performance.now(); + try { + const id = await runWithSpan(producer, () => + this.opts.db.emitEvent( + { + eventKey: event, + payload: payload as any, + trace_context: producer ? carrierForContext(contextForSpan(producer)) : null, + }, + { signal: this.signal }, + ), + ); + setSpanAttribute(producer, "messaging.message.id", id); + if (producer) { + recordSent(String(event), 1); + recordOperationDuration(String(event), performance.now() - started, "send"); + } + return id; + } catch (error) { + setSpanError(producer, error); + throw error; + } finally { + endSpan(producer); + } } /** diff --git a/packages/pgconductor-js/src/telemetry.ts b/packages/pgconductor-js/src/telemetry.ts index c734fa3..df9b79e 100644 --- a/packages/pgconductor-js/src/telemetry.ts +++ b/packages/pgconductor-js/src/telemetry.ts @@ -5,6 +5,9 @@ import { SpanKind, SpanStatusCode, trace, + metrics, + type Counter, + type Histogram, type Context, type Link, type Span, @@ -12,6 +15,9 @@ import { } from "@opentelemetry/api"; import type { TraceContextCarrier } from "./internal-types"; +const meter = () => + safe(() => metrics.getMeter(INSTRUMENTATION_NAME, INSTRUMENTATION_VERSION), null); + export type { TraceContextCarrier } from "./internal-types"; const TRACECONTEXT_SIZE_LIMIT = 1024; @@ -27,6 +33,117 @@ const ZERO_SPAN_ID = "0000000000000000"; type MessagingAttributes = Record; +let instruments: { + sent: Counter; + consumed: Counter; + processDuration: Histogram; + operationDuration: Histogram; + retry: Counter; + permanentFailure: Counter; + cancellation: Counter; + deadLetter: Counter; +} | null = null; +let instrumentProvider: unknown = null; + +function getInstruments() { + const provider = safe(() => metrics.getMeterProvider(), null); + // The API returns a no-op provider before an SDK is installed. Do not cache + // instruments created from it: applications commonly install their provider + // after constructing a Conductor. + if (instruments && instrumentProvider === provider) return instruments; + const m = meter(); + if (!m) return null; + instrumentProvider = provider; + return (instruments = { + sent: m.createCounter("messaging.client.sent.messages", { unit: "{message}" }), + consumed: m.createCounter("messaging.client.consumed.messages", { unit: "{message}" }), + processDuration: m.createHistogram("messaging.process.duration", { unit: "s" }), + operationDuration: m.createHistogram("messaging.client.operation.duration", { unit: "s" }), + retry: m.createCounter("pgconductor.execution.retries", { unit: "{execution}" }), + permanentFailure: m.createCounter("pgconductor.execution.permanent_failures", { + unit: "{execution}", + }), + cancellation: m.createCounter("pgconductor.execution.cancellations", { unit: "{execution}" }), + deadLetter: m.createCounter("pgconductor.execution.dead_letters", { unit: "{execution}" }), + }); +} + +const OUTCOMES = new Set(["retry", "permanent_failure", "cancellation", "dead_letter"]); +const boundedDimension = (value: string | undefined): string | undefined => + value === undefined ? undefined : value.length <= 128 ? value : value.slice(0, 128); + +function metricAttributes(queue: string, task?: string, operation?: string, outcome?: string) { + return { + "messaging.system": "postgres_conductor", + "messaging.destination.name": boundedDimension(queue) || "unknown", + "messaging.operation.name": boundedDimension(operation) || "unknown", + "messaging.operation.type": boundedDimension(operation) || "unknown", + ...(task === undefined || boundedDimension(task) === undefined + ? {} + : { "pgconductor.task.name": boundedDimension(task)! }), + ...(outcome !== undefined && OUTCOMES.has(outcome) ? { "pgconductor.outcome": outcome } : {}), + }; +} + +export function recordSent(queue: string, count = 1, task?: string): void { + safe(() => getInstruments()?.sent.add(count, metricAttributes(queue, task, "send")), undefined); +} +export function recordConsumed(queue: string, task?: string, count = 1): void { + safe( + () => getInstruments()?.consumed.add(count, metricAttributes(queue, task, "process")), + undefined, + ); +} +export function recordProcessDuration(queue: string, durationMs: number, task?: string): void { + safe( + () => + getInstruments()?.processDuration.record( + durationMs / 1000, + metricAttributes(queue, task, "process"), + ), + undefined, + ); +} +export function recordOperationDuration( + queue: string, + durationMs: number, + operation: string, + task?: string, +): void { + safe( + () => + getInstruments()?.operationDuration.record( + durationMs / 1000, + metricAttributes(queue, task, operation), + ), + undefined, + ); +} +export function recordLifecycle( + queue: string, + outcome: "retry" | "permanent_failure" | "cancellation" | "dead_letter", + task?: string, + count = 1, +): void { + safe(() => { + const i = getInstruments(); + const counter = + outcome === "retry" + ? i?.retry + : outcome === "permanent_failure" + ? i?.permanentFailure + : outcome === "cancellation" + ? i?.cancellation + : i?.deadLetter; + counter?.add(count, metricAttributes(queue, task, "settle", outcome)); + }, undefined); +} + +export function resetMetricsForTests(): void { + instruments = null; + instrumentProvider = null; +} + type MessagingOperation = "send" | "process" | "settle"; export const messagingAttributes = ( @@ -35,6 +152,7 @@ export const messagingAttributes = ( operation: MessagingOperation, messageId?: string, batchMessageCount?: number, + eventKey?: string, ): MessagingAttributes => ({ "messaging.system": "postgres_conductor", "messaging.destination.name": queue, @@ -45,6 +163,7 @@ export const messagingAttributes = ( ...(batchMessageCount === undefined ? {} : { "messaging.batch.message_count": batchMessageCount }), + ...(eventKey === undefined ? {} : { "pgconductor.event.name": boundedDimension(eventKey) }), "pgconductor.queue": queue, }); @@ -185,6 +304,15 @@ export function linkContext(ctx: Context | null): Link | null { : null; } +export function linkFromCarrier(value: unknown): Link | null { + return linkContext(extractCarrier(value)); +} + +export function linksFromCarrier(value: unknown): Link[] { + const link = linkFromCarrier(value); + return link ? [link] : []; +} + export function spanContext(span: Span | null): SpanContext | null { if (!span) return null; return safe(() => { @@ -197,6 +325,10 @@ export function contextForSpan(span: Span | null, parent: Context = context.acti return span ? safe(() => trace.setSpan(parent, span), parent) : parent; } +export function contextForSpanContext(value: SpanContext | null): Context { + return value ? safe(() => trace.setSpanContext(ROOT_CONTEXT, value), ROOT_CONTEXT) : ROOT_CONTEXT; +} + export function runWithSpan(span: Span | null, fn: () => T): T { if (!span) return fn(); const active = safe(() => context.active(), ROOT_CONTEXT); diff --git a/packages/pgconductor-js/src/worker.ts b/packages/pgconductor-js/src/worker.ts index b20e780..4965ba5 100644 --- a/packages/pgconductor-js/src/worker.ts +++ b/packages/pgconductor-js/src/worker.ts @@ -1,5 +1,6 @@ import type { - DatabaseClient, + CronRegistration, + DatabaseClientLike, EventSubscriptionSpec, Execution, ExecutionResult, @@ -31,17 +32,25 @@ import { makeChildLogger, type Logger } from "./lib/logger"; import type { EventDefinition } from "./event-definition"; import { coerceError } from "./lib/coerce-error"; import type { TypedAbortController } from "./lib/typed-abort-controller"; -import { ROOT_CONTEXT, SpanKind, type SpanContext } from "@opentelemetry/api"; +import { ROOT_CONTEXT, SpanKind, type Span, type SpanContext } from "@opentelemetry/api"; import { endSpan, extractCarrier, - linkContext, + linksFromCarrier, + carrierForContext, + contextForSpan, + contextForSpanContext, messagingAttributes, runWithSpan, setSpanAttribute, setSpanError, spanContext, startSpan, + recordConsumed, + recordProcessDuration, + recordOperationDuration, + recordLifecycle, + recordSent, } from "./telemetry"; /** @@ -82,6 +91,8 @@ function isRetryableEventError(error: unknown): boolean { return code !== undefined && RETRYABLE_EVENT_ERROR_CODES.has(code); } +const MAINTENANCE_TASK_NAME = "pgconductor.maintenance"; + const DEFAULT_WORKER_CONFIG: WorkerConfig = { concurrency: 1, flushBatchSize: 2, @@ -182,6 +193,13 @@ export class Worker< private _runningTasks = new Map>(); private eventProcessingGate: Promise = Promise.resolve(); private readonly processSpanContexts = new Map(); + private readonly pendingDurableProducers = new Map(); + private readonly parentDeadLetterTargets = new Map< + string, + { sourceExecutionId: string; queue: string; task: string } + >(); + private flushInFlight: Promise = Promise.resolve(); + private hasRegisteredCronSchedules = false; /** Used by Orchestrator to prevent local event fan-out before every worker registers. */ setEventProcessingGate(gate: Promise): void { @@ -191,7 +209,7 @@ export class Worker< constructor( public readonly queueName: string, tasks: readonly AnyTask[], - private readonly db: DatabaseClient, + private readonly db: DatabaseClientLike, private readonly logger: Logger, config: Partial = {}, private readonly extraContext: object = {}, @@ -328,6 +346,7 @@ export class Worker< this._stopDeferred?.reject(error); } finally { queue.close(); + await this.flushInFlight; if (this._stopDeferred && !this._stopDeferred.isSettled) this._stopDeferred.resolve(); this.resetLifecycle(); } @@ -378,6 +397,7 @@ export class Worker< this.logger.error("Worker pipeline error:", error); this._stopDeferred?.reject(error); } finally { + await this.flushInFlight; if (this._stopDeferred && !this._stopDeferred.isSettled) this._stopDeferred.resolve(); this.resetLifecycle(); } @@ -390,6 +410,8 @@ export class Worker< this.orchestratorId = null; this.eventProcessingGate = Promise.resolve(); this.processSpanContexts.clear(); + this.pendingDurableProducers.clear(); + this.parentDeadLetterTargets.clear(); } private async processEventBatches({ runOnce }: { runOnce: boolean }): Promise { @@ -514,11 +536,12 @@ export class Worker< const allTasks = Array.from(this.tasks.values()); + const currentTime = await this.db.getCurrentTime({ signal: this.signal }); const cronSchedules: ExecutionSpec[] = allTasks.flatMap((task) => task.triggers .filter((t): t is { cron: string; name: string; group?: string } => "cron" in t) .map((trigger) => { - const interval = CronExpressionParser.parse(trigger.cron); + const interval = CronExpressionParser.parse(trigger.cron, { currentDate: currentTime }); const nextTimestamp = interval.next().toDate(); const timestampSeconds = Math.floor(nextTimestamp.getTime() / 1000); return { @@ -580,15 +603,48 @@ export class Worker< return [...customEvents, ...dbEvents]; }); - await this.db.registerWorker( - { - queueName: this.queueName, - taskSpecs, - cronSchedules, - eventSubscriptions, - }, - { signal: this.signal }, + const userCronSchedules = cronSchedules.filter( + (schedule) => schedule.task_key !== MAINTENANCE_TASK_NAME, ); + const cronProducer = + this.telemetry && userCronSchedules.length && !this.hasRegisteredCronSchedules + ? startSpan( + `send ${this.queueName}`, + SpanKind.PRODUCER, + messagingAttributes( + userCronSchedules.length === 1 ? userCronSchedules[0]?.task_key : undefined, + this.queueName, + "send", + undefined, + userCronSchedules.length, + ), + ) + : null; + const cronCarrier = cronProducer ? carrierForContext(contextForSpan(cronProducer)) : null; + for (const schedule of userCronSchedules) schedule.trace_context = cronCarrier; + const registrationStarted = performance.now(); + try { + const registrations: CronRegistration[] = await runWithSpan(cronProducer, () => + this.db.registerWorker( + { queueName: this.queueName, taskSpecs, cronSchedules, eventSubscriptions }, + { signal: this.signal }, + ), + ); + const authoritativeUserRows = registrations.filter( + (row) => row.authoritative && !row.is_maintenance, + ); + setSpanAttribute(cronProducer, "messaging.batch.message_count", authoritativeUserRows.length); + if (cronProducer && authoritativeUserRows.length > 0 && this.telemetry) { + recordSent(this.queueName, authoritativeUserRows.length); + recordOperationDuration(this.queueName, performance.now() - registrationStarted, "send"); + } + this.hasRegisteredCronSchedules = true; + } catch (error) { + setSpanError(cronProducer, error); + throw error; + } finally { + endSpan(cronProducer); + } } // --- Stage 1: Fetch executions from database --- @@ -599,12 +655,7 @@ export class Worker< let fetched = 0; assert.ok(this.orchestratorId, "orchestratorId must be set when starting the pipeline"); - // Pre-compute task metadata once const allTasks = Array.from(this.tasks.values()); - const taskMaxAttempts: Record = {}; - for (const task of allTasks) { - taskMaxAttempts[task.name] = task.maxAttempts || 3; - } // Check if any tasks have windows - only then do we need time-based filtering const tasksWithWindows = allTasks.filter((task) => task.window); @@ -692,23 +743,23 @@ export class Worker< orchestrator_id: exec.locked_by, task_key: taskKey, status: "failed", + cancelled: false, error: `Task not found: ${taskKey}`, })) as ExecutionResult[]; } // Safety check: don't execute already-cancelled tasks const cancelledExecs = executions.filter((e) => e.cancelled); - if (cancelledExecs.length === executions.length) { - // All cancelled - return failures for all - return executions.map((exec) => ({ - execution_id: exec.id, - orchestrator_id: exec.locked_by, - queue: exec.queue, - task_key: taskKey, - status: "permanently_failed", - error: exec.last_error || "Execution was cancelled", - })) as ExecutionResult[]; - } + const cancelledResults = cancelledExecs.map((exec) => ({ + execution_id: exec.id, + orchestrator_id: exec.locked_by, + queue: exec.queue, + task_key: taskKey, + status: "permanently_failed" as const, + cancelled: true as const, + error: exec.last_error || "Execution was cancelled", + })); + if (cancelledExecs.length === executions.length) return cancelledResults; // Note: We intentionally do NOT check this.signal.aborted here. // When shutdown occurs, fetchExecutions closes the queue which flushes @@ -727,13 +778,19 @@ export class Worker< // If task has batch config, always use batch execution (even for single items) if (task.batch) { - return this.executeBatchTask(task, taskKey, activeExecs); + return [ + ...cancelledResults, + ...(await this.executeBatchTask(task, taskKey, activeExecs)), + ]; } // Execute single (non-batched tasks) const singleExec = activeExecs[0]; assert.ok(singleExec, "activeExecs must have at least one item"); - return this.executeSingleTask(task, singleExec); + const activeResult = await this.executeSingleTask(task, singleExec); + return cancelledResults.length && activeResult + ? [...cancelledResults, activeResult] + : activeResult; }, )) { // Waiting executions are released atomically by registerEventWait and @@ -762,19 +819,28 @@ export class Worker< resolve(taskAbortController.signal.reason); }); }); - const consumer = this.telemetry + const processTelemetry = this.telemetry && exec.task_key !== MAINTENANCE_TASK_NAME; + if (exec.parent_execution_id && exec.parent_dead_letter_queue) + this.parentDeadLetterTargets.set(exec.id, { + sourceExecutionId: exec.parent_execution_id, + queue: exec.parent_dead_letter_queue, + task: exec.parent_dead_letter_task_key || exec.parent_task_key || exec.task_key, + }); + const processStarted = performance.now(); + const consumer = processTelemetry ? startSpan( `process ${exec.queue}`, SpanKind.CONSUMER, messagingAttributes(exec.task_key, exec.queue, "process", exec.id), extractCarrier(exec.trace_context) || ROOT_CONTEXT, + linksFromCarrier(exec.trace_link_context), ) : null; const consumerContext = spanContext(consumer); if (consumerContext) this.processSpanContexts.set(exec.id, consumerContext); try { - await this.scheduleNextExecution(exec); + await this.scheduleNextExecution(exec, consumer, processTelemetry); // Determine event type based on execution data let taskEvent: any; @@ -820,7 +886,7 @@ export class Worker< queue: exec.queue, }), window: task.window, - telemetry: this.telemetry ? undefined : false, + telemetry: processTelemetry ? undefined : false, }, extraContext, ), @@ -855,6 +921,7 @@ export class Worker< queue: exec.queue, task_key: exec.task_key, status: "permanently_failed", + cancelled: true, error: exec.last_error || "Task was cancelled", } as const; case "released": @@ -888,10 +955,18 @@ export class Worker< orchestrator_id: exec.locked_by, queue: exec.queue, task_key: exec.task_key, - status: "failed", + status: + exec.attempts !== undefined && exec.attempts >= (task.maxAttempts ?? 3) + ? "permanently_failed" + : "failed", + cancelled: false, error: coerceError(err).message, } as const; } finally { + if (processTelemetry) { + recordConsumed(exec.queue, exec.task_key); + recordProcessDuration(exec.queue, performance.now() - processStarted, exec.task_key); + } endSpan(consumer); // Clean up running task tracking this._runningTasks.delete(exec.id); @@ -910,10 +985,21 @@ export class Worker< taskKey: string, executions: Execution[], ): Promise { - const links = this.telemetry - ? executions.flatMap((exec) => linkContext(extractCarrier(exec.trace_context)) || []) + const processTelemetry = this.telemetry && taskKey !== MAINTENANCE_TASK_NAME; + for (const exec of executions) + if (exec.parent_execution_id && exec.parent_dead_letter_queue) + this.parentDeadLetterTargets.set(exec.id, { + sourceExecutionId: exec.parent_execution_id, + queue: exec.parent_dead_letter_queue, + task: exec.parent_dead_letter_task_key || exec.parent_task_key || exec.task_key, + }); + const links = processTelemetry + ? executions.flatMap((exec) => [ + ...linksFromCarrier(exec.trace_context), + ...linksFromCarrier(exec.trace_link_context), + ]) : []; - const consumer = this.telemetry + const consumer = processTelemetry ? startSpan( `process ${this.queueName}`, SpanKind.CONSUMER, @@ -964,9 +1050,12 @@ export class Worker< }); }); + const processStarted = performance.now(); try { // Schedule next executions for cron tasks - await Promise.all(executions.map((exec) => this.scheduleNextExecution(exec))); + await Promise.all( + executions.map((exec) => this.scheduleNextExecution(exec, consumer, processTelemetry)), + ); const result = await runWithSpan(consumer, () => Promise.race([task.execute(events, batchContext), abortPromise]), @@ -994,6 +1083,7 @@ export class Worker< queue: exec.queue, task_key: taskKey, status: "failed" as const, + cancelled: false, error: `Task aborted: ${result.reason}`, })); } @@ -1040,14 +1130,23 @@ export class Worker< queue: exec.queue, task_key: taskKey, status: "failed" as const, + cancelled: false, error: errorMsg, })); } finally { + if (processTelemetry) { + recordConsumed(this.queueName, taskKey, executions.length); + recordProcessDuration(this.queueName, performance.now() - processStarted, taskKey); + } endSpan(consumer); } } - private async scheduleNextExecution(execution: Execution): Promise { + private async scheduleNextExecution( + execution: Execution, + parent: Span | null = null, + telemetry = this.telemetry, + ): Promise { if (!execution.cron_expression) { return; } @@ -1063,22 +1162,57 @@ export class Worker< } const scheduleName = parts[1]; - const interval = CronExpressionParser.parse(execution.cron_expression); + const currentTime = await this.db.getCurrentTime({ signal: this.signal }); + const interval = CronExpressionParser.parse(execution.cron_expression, { + currentDate: currentTime, + }); const nextTimestamp = interval.next().toDate(); const timestampSeconds = Math.floor(nextTimestamp.getTime() / 1000); const nextDedupeKey = `scheduled::${scheduleName}::${timestampSeconds}`; - await this.db.invoke( - { - task_key: execution.task_key, - queue: execution.queue, - run_at: nextTimestamp, - dedupe_key: nextDedupeKey, - cron_expression: execution.cron_expression, - group: execution.group || null, - }, - { signal: this.signal }, - ); + const producer = telemetry + ? startSpan( + `send ${execution.queue}`, + SpanKind.PRODUCER, + messagingAttributes(execution.task_key, execution.queue, "send"), + parent ? contextForSpanContext(spanContext(parent)) : ROOT_CONTEXT, + ) + : null; + const started = performance.now(); + try { + const id = await runWithSpan(producer, () => + this.db.invoke( + { + task_key: execution.task_key, + queue: execution.queue, + run_at: nextTimestamp, + dedupe_key: nextDedupeKey, + cron_expression: execution.cron_expression, + group: execution.group || null, + trace_context: producer + ? carrierForContext(contextForSpanContext(spanContext(producer))) + : null, + }, + { signal: this.signal }, + ), + ); + if (id) { + setSpanAttribute(producer, "messaging.message.id", id); + if (telemetry) recordSent(execution.queue, 1, execution.task_key); + if (telemetry) + recordOperationDuration( + execution.queue, + performance.now() - started, + "send", + execution.task_key, + ); + } + } catch (error) { + setSpanError(producer, error); + throw error; + } finally { + endSpan(producer); + } } // --- Stage 3: Flush results to database --- @@ -1097,6 +1231,70 @@ export class Worker< flushTimer = null; } + if (this.telemetry) { + for (const result of [ + ...batch.invokeChild, + ...batch.failed.filter((item) => { + if (item.status !== "permanently_failed") return false; + const task = this.tasks.get(item.task_key); + return Boolean(task?.deadLetter?.queue && !item.cancelled); + }), + ]) { + const existing = this.pendingDurableProducers.get(result.execution_id); + const parent = this.processSpanContexts.get(result.execution_id); + const deadLetter = + result.status === "invoke_child" + ? null + : this.tasks.get(result.task_key)?.deadLetter || null; + const targets = + result.status === "invoke_child" + ? [ + { + sourceExecutionId: result.execution_id, + queue: result.child_task_queue, + task: result.child_task_name, + }, + ] + : [ + ...(deadLetter?.queue && !result.cancelled + ? [ + { + sourceExecutionId: result.execution_id, + queue: deadLetter.queue, + task: deadLetter.task?.name || result.task_key, + }, + ] + : []), + ...(result.status === "permanently_failed" && !result.cancelled + ? (() => { + const target = this.parentDeadLetterTargets.get(result.execution_id); + return target ? [target] : []; + })() + : []), + ]; + for (const target of targets) { + const producer = + (target.sourceExecutionId === result.execution_id ? existing : undefined) || + startSpan( + `send ${target.queue}`, + SpanKind.PRODUCER, + messagingAttributes(target.task, target.queue, "send"), + parent ? contextForSpanContext(parent) : ROOT_CONTEXT, + ); + if (producer) { + this.pendingDurableProducers.set(target.sourceExecutionId, producer); + const carrier = carrierForContext(contextForSpanContext(spanContext(producer))); + if (carrier) { + if (result.status === "invoke_child") result.trace_context = carrier; + else { + result.dead_letter_trace_contexts ||= {}; + result.dead_letter_trace_contexts[target.sourceExecutionId] = carrier; + } + } + } + } + } + } const links = this.telemetry ? Array.from( new Set( @@ -1117,9 +1315,39 @@ export class Worker< ) : null; let settled = false; + const settleStarted = performance.now(); try { batch.orchestratorId = this.orchestratorId || batch.orchestratorId; - await runWithSpan(settle, () => this.db.returnExecutions(batch, { signal: this.signal })); + const committed = await runWithSpan(settle, () => + this.db.returnExecutions(batch, { signal: this.signal }), + ); + const hasUserResults = Array.from(batch.taskKeys).some( + (key) => key !== MAINTENANCE_TASK_NAME, + ); + if (this.telemetry) { + for (const delivery of committed?.deliveries || []) { + const producer = this.pendingDurableProducers.get(delivery.sourceExecutionId); + setSpanAttribute( + producer || null, + "messaging.message.id", + delivery.destinationExecutionId, + ); + endSpan(producer || null); + this.pendingDurableProducers.delete(delivery.sourceExecutionId); + } + } + if (this.telemetry && hasUserResults) { + recordOperationDuration(this.queueName, performance.now() - settleStarted, "settle"); + for (const outcome of committed?.outcomes || []) { + if (outcome.task_key === MAINTENANCE_TASK_NAME) continue; + recordLifecycle( + outcome.queue, + outcome.outcome, + outcome.task_key, + Number(outcome.count), + ); + } + } settled = true; setSpanAttribute(settle, "pgconductor.db.commit.status", "success"); } catch (err) { @@ -1129,22 +1357,46 @@ export class Worker< buffer.restore(batch); } } finally { + if (settled || isCleanup) { + const pending = isCleanup + ? [...this.pendingDurableProducers.entries()] + : [...batch.completed, ...batch.failed, ...batch.released, ...batch.invokeChild].map( + (result) => + [ + result.execution_id, + this.pendingDurableProducers.get(result.execution_id), + ] as const, + ); + for (const [executionId, producer] of pending) { + endSpan(producer || null); + this.pendingDurableProducers.delete(executionId); + } + } endSpan(settle); - if (settled || isCleanup) + if (settled || isCleanup) { for (const result of [ ...batch.completed, ...batch.failed, ...batch.released, ...batch.invokeChild, - ]) + ]) { this.processSpanContexts.delete(result.execution_id); + this.parentDeadLetterTargets.delete(result.execution_id); + } + } } }; + const runFlush = (isCleanup = false): Promise => { + const next = this.flushInFlight.then(() => flushNow(isCleanup)); + this.flushInFlight = next.catch(() => {}); + return next; + }; + const scheduleFlush = () => { if (flushTimer) clearTimeout(flushTimer); - flushTimer = setTimeout(async () => { - await flushNow(); + flushTimer = setTimeout(() => { + void runFlush(); }, this.flushIntervalMs); }; @@ -1156,13 +1408,13 @@ export class Worker< if (!flushTimer) scheduleFlush(); if (buffer.count >= this.flushBatchSize) { - await flushNow(); + await runFlush(); scheduleFlush(); } } } finally { if (flushTimer) clearTimeout(flushTimer); - await flushNow(true); + await runFlush(true); } } diff --git a/packages/pgconductor-js/tests/integration/cron-scheduling.test.ts b/packages/pgconductor-js/tests/integration/cron-scheduling.test.ts index 409e7ab..18e5a20 100644 --- a/packages/pgconductor-js/tests/integration/cron-scheduling.test.ts +++ b/packages/pgconductor-js/tests/integration/cron-scheduling.test.ts @@ -6,17 +6,7 @@ import { defineTask } from "../../src/task-definition"; import { TestDatabasePool } from "../fixtures/test-database"; import { waitFor } from "../../src/lib/wait-for"; import { TaskSchemas } from "../../src/schemas"; - -async function waitForCondition( - condition: () => boolean | Promise, - timeoutMs = 5000, -): Promise { - const deadline = Date.now() + timeoutMs; - while (!(await condition())) { - if (Date.now() >= deadline) throw new Error("condition was not met before timeout"); - await waitFor(50); - } -} +import { waitForCondition } from "../test-utils"; describe("Cron Scheduling", () => { let pool: TestDatabasePool; diff --git a/packages/pgconductor-js/tests/integration/dead-letter.test.ts b/packages/pgconductor-js/tests/integration/dead-letter.test.ts index 3f513cb..785898f 100644 --- a/packages/pgconductor-js/tests/integration/dead-letter.test.ts +++ b/packages/pgconductor-js/tests/integration/dead-letter.test.ts @@ -5,17 +5,7 @@ import { Orchestrator } from "../../src/orchestrator"; import { TaskSchemas } from "../../src/schemas"; import { defineTask } from "../../src/task-definition"; import { TestDatabasePool, type TestDatabase } from "../fixtures/test-database"; - -const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); - -async function eventually(check: () => Promise, timeoutMs = 20_000): Promise { - const deadline = Date.now() + timeoutMs; - while (Date.now() < deadline) { - if (await check()) return; - await sleep(25); - } - throw new Error(`condition was not met within ${timeoutMs}ms`); -} +import { waitForCondition } from "../test-utils"; describe("dead-letter queues (Postgres integration)", () => { let pool: TestDatabasePool; @@ -89,7 +79,7 @@ describe("dead-letter queues (Postgres integration)", () => { }); await sourceOrchestrator.start(); await conductor.invoke({ name: "charge" }, { value: "order-42" }); - await eventually(async () => { + await waitForCondition(async () => { const rows = await db.sql<{ attempts: number; released: boolean }[]>` select attempts, locked_by is null as released from pgconductor._private_executions @@ -103,7 +93,7 @@ describe("dead-letter queues (Postgres integration)", () => { `; if (!retry) throw new Error("expected persisted retry"); await db.client.setFakeTime({ date: new Date(retry.run_at.getTime() + 1) }); - await eventually(async () => { + await waitForCondition(async () => { const rows = await db.sql<{ count: string }[]>` select count(*)::text as count from pgconductor._private_executions where queue = 'dlq' @@ -130,7 +120,7 @@ describe("dead-letter queues (Postgres integration)", () => { ], }); await destinationOrchestrator.start(); - await eventually(async () => seen.length === 1); + await waitForCondition(async () => seen.length === 1); await destinationOrchestrator.stop(); expect(seen).toEqual([ { value: "order-42", sourceTask: "charge", attempts: 2, error: "card declined" }, @@ -186,6 +176,7 @@ describe("dead-letter queues (Postgres integration)", () => { queue: execution.queue, task_key: execution.task_key, status: "permanently_failed" as const, + cancelled: false, orchestrator_id: execution.locked_by, error: "settlement failure", }; @@ -311,7 +302,7 @@ describe("dead-letter queues (Postgres integration)", () => { cronSchedules: [], eventSubscriptions: [], }), - ).resolves.toBeUndefined(); + ).resolves.toEqual([]); }, 15000); test("rolls back the source settlement when destination insertion fails", async () => { @@ -370,6 +361,7 @@ describe("dead-letter queues (Postgres integration)", () => { queue: execution.queue, task_key: execution.task_key, status: "permanently_failed" as const, + cancelled: false, orchestrator_id: lockedBy, error: "rollback failure", }; diff --git a/packages/pgconductor-js/tests/integration/event-pipeline.test.ts b/packages/pgconductor-js/tests/integration/event-pipeline.test.ts index a50306f..143561e 100644 --- a/packages/pgconductor-js/tests/integration/event-pipeline.test.ts +++ b/packages/pgconductor-js/tests/integration/event-pipeline.test.ts @@ -9,6 +9,7 @@ import { defineTask } from "../../src/task-definition"; import { EventSchemas, TaskSchemas } from "../../src/schemas"; import { TestDatabasePool } from "../fixtures/test-database"; import type { TestDatabase } from "../fixtures/test-database"; +import { waitForCondition } from "../test-utils"; import postgres from "postgres"; describe("event pipeline", () => { @@ -35,15 +36,6 @@ describe("event pipeline", () => { return db; } - async function waitUntil(check: () => Promise, timeout = 5000): Promise { - const deadline = Date.now() + timeout; - while (Date.now() < deadline) { - if (await check()) return; - await Bun.sleep(10); - } - throw new Error("condition was not met before timeout"); - } - async function subscription( db: TestDatabase, taskKey: string, @@ -95,7 +87,7 @@ describe("event pipeline", () => { await conductor.emit("pipeline.order", { status: "trial", region: "us" }); await conductor.emit("pipeline.order", { status: "paid", region: "eu" }); await conductor.emit("pipeline.order", { status: "cancelled", region: "us" }); - await waitUntil(async () => received.length === 2); + await waitForCondition(async () => received.length === 2); expect(received.sort()).toEqual(["paid", "trial"]); } finally { await orchestrator.stop(); diff --git a/packages/pgconductor-js/tests/integration/execution-foundations.test.ts b/packages/pgconductor-js/tests/integration/execution-foundations.test.ts index 8668e9f..432c91d 100644 --- a/packages/pgconductor-js/tests/integration/execution-foundations.test.ts +++ b/packages/pgconductor-js/tests/integration/execution-foundations.test.ts @@ -213,6 +213,7 @@ describe("execution foundations", () => { orchestrator_id: child.locked_by, task_key: child.task_key, status: "permanently_failed", + cancelled: false, error: "child failed", }, ], @@ -305,6 +306,7 @@ describe("execution foundations", () => { orchestrator_id: child.locked_by, task_key: child.task_key, status: "permanently_failed", + cancelled: false, error: "child failed", }, ], @@ -532,7 +534,7 @@ describe("execution foundations", () => { count: 3, orchestratorId: oldOrchestrator, completed: [{ ...staleBase, status: "completed" }], - failed: [{ ...staleBase, status: "failed", error: "stale" }], + failed: [{ ...staleBase, status: "failed", cancelled: false, error: "stale" }], released: [{ ...staleBase, status: "released", reschedule_in_ms: 0 }], invokeChild: [], taskKeys: new Set(["fenced-task"]), diff --git a/packages/pgconductor-js/tests/integration/group-concurrency.test.ts b/packages/pgconductor-js/tests/integration/group-concurrency.test.ts index 9dd3022..ad96891 100644 --- a/packages/pgconductor-js/tests/integration/group-concurrency.test.ts +++ b/packages/pgconductor-js/tests/integration/group-concurrency.test.ts @@ -7,6 +7,7 @@ import { TaskSchemas } from "../../src/schemas"; import { Deferred } from "../../src/lib/deferred"; import { TestDatabasePool } from "../fixtures/test-database"; import type { TestDatabase } from "../fixtures/test-database"; +import { waitForCondition } from "../test-utils"; const workerConfig = { concurrency: 10, @@ -19,16 +20,6 @@ const workerConfig = { // Group limits are intentionally soft: concurrent claim transactions may race. // These tests use blockers and avoid asserting exact global bounds across workers. -async function waitUntil(predicate: () => boolean, timeoutMs = 20_000): Promise { - const deadline = Date.now() + timeoutMs; - while (!predicate()) { - if (Date.now() >= deadline) { - throw new Error("Timed out waiting for condition"); - } - await new Promise((resolve) => setTimeout(resolve, 10)); - } -} - describe("Group concurrency", () => { let pool: TestDatabasePool; const databases: TestDatabase[] = []; @@ -82,13 +73,13 @@ describe("Group concurrency", () => { try { await conductor.invoke({ name: "same-group" }, { id: 1 }, { group: "tenant-a" }); await conductor.invoke({ name: "same-group" }, { id: 2 }, { group: "tenant-a" }); - await waitUntil(() => started.length === 1); + await waitForCondition(() => started.length === 1); await new Promise((resolve) => setTimeout(resolve, 100)); const first = started[0]; expect(first === 1 || first === 2).toBe(true); blockers.get(first!)?.resolve(); - await waitUntil(() => started.length === 2); + await waitForCondition(() => started.length === 2); expect(new Set(started)).toEqual(new Set([1, 2])); } finally { blockers.forEach((blocker) => blocker.resolve()); @@ -127,9 +118,9 @@ describe("Group concurrency", () => { await orchestrator.start(); try { await conductor.invoke({ name: "different-groups" }, { id: 1 }, { group: "tenant-a" }); - await waitUntil(() => started.includes(1)); + await waitForCondition(() => started.includes(1)); await conductor.invoke({ name: "different-groups" }, { id: 2 }, { group: "tenant-b" }); - await waitUntil(() => started.length === 2); + await waitForCondition(() => started.length === 2); expect(new Set(started)).toEqual(new Set([1, 2])); } finally { blocker.resolve(); @@ -169,7 +160,7 @@ describe("Group concurrency", () => { try { await conductor.invoke({ name: "ungrouped" }, { id: 1 }, { group: "tenant-a" }); await conductor.invoke({ name: "ungrouped" }, { id: 2 }); - await waitUntil(() => started.length === 2); + await waitForCondition(() => started.length === 2); expect(new Set(started)).toEqual(new Set([1, 2])); } finally { blocker.resolve(); @@ -215,13 +206,15 @@ describe("Group concurrency", () => { await conductor.invoke({ name: "composed-limits" }, { id: 1 }, { group: "tenant-a" }); await conductor.invoke({ name: "composed-limits" }, { id: 3 }, { group: "tenant-b" }); await conductor.invoke({ name: "composed-limits" }, { id: 2 }, { group: "tenant-a" }); - await waitUntil(() => started.includes(3) && started.some((id) => id === 1 || id === 2)); + await waitForCondition( + () => started.includes(3) && started.some((id) => id === 1 || id === 2), + ); await new Promise((resolve) => setTimeout(resolve, 100)); expect(started).toContain(3); expect(started.filter((id) => id === 1 || id === 2)).toHaveLength(1); blockers.forEach((blocker) => blocker.resolve()); - await waitUntil(() => started.length === 3); + await waitForCondition(() => started.length === 3); } finally { blockers.forEach((blocker) => blocker.resolve()); await orchestrator.stop(); @@ -282,7 +275,7 @@ describe("Group concurrency", () => { await conductor.invoke({ name: "scope-a" }, {}, { group: "shared" }); await conductor.invoke({ name: "scope-b" }, {}, { group: "shared" }); await conductor.invoke({ name: "scope-queue", queue: "other" }, {}, { group: "shared" }); - await waitUntil(() => started.size === 3); + await waitForCondition(() => started.size === 3); expect(started).toEqual(new Set(["task-a", "task-b", "queue"])); } finally { blocker.resolve(); @@ -336,12 +329,12 @@ describe("Group concurrency", () => { `; expect(rows.map((row) => row.group)).toEqual(["batch-tenant", "batch-tenant"]); - await waitUntil(() => started.length === 1); + await waitForCondition(() => started.length === 1); await new Promise((resolve) => setTimeout(resolve, 100)); const first = started[0]; expect(first === 1 || first === 2).toBe(true); blockers.get(first!)?.resolve(); - await waitUntil(() => started.length === 2); + await waitForCondition(() => started.length === 2); expect(new Set(started)).toEqual(new Set([1, 2])); } finally { blockers.forEach((blocker) => blocker.resolve()); diff --git a/packages/pgconductor-js/tests/integration/invoke-support.test.ts b/packages/pgconductor-js/tests/integration/invoke-support.test.ts index 45add61..c25f0ea 100644 --- a/packages/pgconductor-js/tests/integration/invoke-support.test.ts +++ b/packages/pgconductor-js/tests/integration/invoke-support.test.ts @@ -7,15 +7,7 @@ import { TestDatabasePool } from "../fixtures/test-database"; import type { TestDatabase } from "../fixtures/test-database"; import { TaskSchemas } from "../../src/schemas"; import { Deferred } from "../../src/lib/deferred"; - -async function waitForCondition(check: () => Promise, timeoutMs = 20_000): Promise { - const deadline = Date.now() + timeoutMs; - while (Date.now() < deadline) { - if (await check()) return; - await new Promise((resolve) => setTimeout(resolve, 25)); - } - throw new Error(`condition was not met within ${timeoutMs}ms`); -} +import { waitForCondition } from "../test-utils"; describe("Invoke Support", () => { let pool: TestDatabasePool; diff --git a/packages/pgconductor-js/tests/integration/subscription-lifecycle.test.ts b/packages/pgconductor-js/tests/integration/subscription-lifecycle.test.ts index d5708d2..0e9d6a4 100644 --- a/packages/pgconductor-js/tests/integration/subscription-lifecycle.test.ts +++ b/packages/pgconductor-js/tests/integration/subscription-lifecycle.test.ts @@ -9,6 +9,7 @@ import { DatabaseSchema } from "../../src/schemas"; import { z } from "zod"; import { TestDatabasePool, TestDatabase } from "../fixtures/test-database"; import type { Database } from "../database.types"; +import { waitForCondition } from "../test-utils"; describe("Event Subscription Lifecycle", () => { let pool: TestDatabasePool; @@ -30,15 +31,6 @@ describe("Event Subscription Lifecycle", () => { await pool?.destroy(); }); - async function waitUntil(check: () => Promise, timeout = 5000): Promise { - const deadline = Date.now() + timeout; - while (Date.now() < deadline) { - if (await check()) return; - await Bun.sleep(10); - } - throw new Error("condition was not met before timeout"); - } - test("custom event subscriptions are persisted and processed asynchronously", async () => { const db = await pool.child(); databases.push(db); @@ -95,7 +87,7 @@ describe("Event Subscription Lifecycle", () => { expect(compiledFilters.count).toBe("0"); await conductor.emit("user.created", { userId: "user-123" }); - await waitUntil(async () => taskFn.mock.calls.length === 1); + await waitForCondition(async () => taskFn.mock.calls.length === 1); const [processedEvent] = await db.sql<[{ processed_at: Date | null }]>` select processed_at diff --git a/packages/pgconductor-js/tests/integration/telemetry-features.test.ts b/packages/pgconductor-js/tests/integration/telemetry-features.test.ts new file mode 100644 index 0000000..2f6797f --- /dev/null +++ b/packages/pgconductor-js/tests/integration/telemetry-features.test.ts @@ -0,0 +1,645 @@ +import { afterAll, afterEach, beforeAll, describe, expect, test } from "bun:test"; +import { AsyncLocalStorage } from "node:async_hooks"; +import { + context, + ROOT_CONTEXT, + SpanKind, + trace, + type Context, + type ContextManager, +} from "@opentelemetry/api"; +import { + BasicTracerProvider, + InMemorySpanExporter, + SimpleSpanProcessor, +} from "@opentelemetry/sdk-trace-base"; +import type { Sql } from "postgres"; +import { Conductor } from "../../src/conductor"; +import { EventSchemas, TaskSchemas } from "../../src/schemas"; +import { Task } from "../../src/task"; +import { Orchestrator } from "../../src/orchestrator"; +import { Worker } from "../../src/worker"; +import { DefaultLogger } from "../../src/lib/logger"; +import { + carrierForContext, + contextForSpan, + endSpan, + messagingAttributes, + runWithSpan, + startSpan, +} from "../../src/telemetry"; +import { InMemoryDatabaseClient } from "../mocks/in-memory-database-client"; +import { TestDatabasePool, type TestDatabase } from "../fixtures/test-database"; +import { defineEvent } from "../../src/event-definition"; +import { defineTask } from "../../src/task-definition"; +import { z } from "zod"; +import { waitForCondition } from "../test-utils"; + +const logger = new DefaultLogger(); +const fakeSql = Object.assign((async () => []) as unknown as Sql, { + json: (value: unknown) => JSON.stringify(value), +}); + +class TestContextManager implements ContextManager { + private readonly storage = new AsyncLocalStorage(); + active(): Context { + return this.storage.getStore() || ROOT_CONTEXT; + } + with ReturnType>( + ctx: Context, + fn: F, + thisArg?: ThisParameterType, + ...args: A + ): ReturnType { + return this.storage.run(ctx, () => fn.apply(thisArg, args)); + } + bind(_: Context, target: T): T { + return target; + } + enable(): this { + return this; + } + disable(): this { + this.storage.disable(); + return this; + } +} + +const activeProviders = new Set(); +const activeOrchestrators = new Set(); + +function installProvider() { + context.disable(); + trace.disable(); + const exporter = new InMemorySpanExporter(); + const provider = new BasicTracerProvider(); + provider.addSpanProcessor(new SimpleSpanProcessor(exporter)); + provider.register(); + context.setGlobalContextManager(new TestContextManager()); + activeProviders.add(provider); + return { exporter, provider }; +} + +async function cleanupProvider(provider: BasicTracerProvider) { + activeProviders.delete(provider); + await provider.shutdown(); + context.disable(); + trace.disable(); +} + +function namedSpans(exporter: InMemorySpanExporter, name: string) { + return exporter.getFinishedSpans().filter((span) => span.name === name); +} + +function makeTask( + name: string, + execute: (event: any, ctx: any) => Promise, + definition: Record = {}, + trigger: Record = { invocable: true }, +) { + return Task.create({ name, ...definition } as any, trigger as any, execute as any); +} + +function makeWorker( + db: InMemoryDatabaseClient, + tasks: any[], + telemetry = true, + eventDefinitions: any[] = [], +) { + return new Worker( + "default", + tasks, + db, + logger, + { pollIntervalMs: 1, flushIntervalMs: 1, fetchBatchSize: 10, flushBatchSize: 10 }, + {}, + eventDefinitions, + telemetry, + ); +} + +function externalProducer(name: string, destination = "default") { + const span = startSpan( + name, + SpanKind.PRODUCER, + messagingAttributes(undefined, destination, "send"), + ); + const carrier = carrierForContext(contextForSpan(span)); + return { span, carrier }; +} + +describe.serial("OpenTelemetry trace propagation feature paths", () => { + test("Conductor.emit producer becomes the event-trigger consumer parent", async () => { + const { exporter, provider } = installProvider(); + const db = new InMemoryDatabaseClient(); + const event = defineEvent({ name: "feature.event", payload: z.object({ id: z.string() }) }); + const conductor = Conductor.create({ + sql: fakeSql, + events: EventSchemas.fromSchema([event]), + context: {}, + }); + (conductor as any).db = db; + const task = makeTask("event-consumer", async () => undefined, {}, { event: "feature.event" }); + const worker = makeWorker(db, [task], true, [event]); + await worker.drain("event-worker-register"); + + const root = externalProducer("event-root"); + await runWithSpan(root.span, () => conductor.emit("feature.event", { id: "1" })); + endSpan(root.span); + await worker.drain("event-worker"); + + const send = namedSpans(exporter, "send event feature.event")[0]!; + const process = namedSpans(exporter, "process default")[0]!; + expect(send.parentSpanId).toBe(root.span?.spanContext().spanId); + expect(process.parentSpanId).toBe(send.spanContext().spanId); + expect(process.spanContext().traceId).toBe(send.spanContext().traceId); + await cleanupProvider(provider); + }); + + test("waitForEvent keeps the task producer parent and links the event producer (real Postgres)", async () => { + const db = await postgresDatabases.child(); + postgresChildren.push(db); + const event = defineEvent({ name: "feature.wait", payload: z.object({ id: z.string() }) }); + const taskDefinition = defineTask({ name: "feature.waiter", payload: z.object({}) }); + const { exporter, provider } = installProvider(); + const conductor = Conductor.create({ + sql: db.sql, + tasks: TaskSchemas.fromSchema([taskDefinition]), + events: EventSchemas.fromSchema([event]), + context: {}, + }); + const task = conductor.createTask( + { name: "feature.waiter" }, + { invocable: true }, + async (_event, ctx) => { + await ctx.waitForEvent("feature-wait", { event }); + }, + ); + const orchestrator = Orchestrator.create({ + conductor, + tasks: [task], + defaultWorker: { pollIntervalMs: 10, flushIntervalMs: 10 }, + }); + activeOrchestrators.add(orchestrator); + await orchestrator.start(); + const invocation = await conductor.invoke({ name: "feature.waiter" }, {}); + await waitForCondition(async () => { + const rows = await db.sql<{ n: number }[]>` + select count(*)::int as n from pgconductor._private_event_subscriptions + where kind = 'execution_wait' + `; + return Number(rows[0]?.n || 0) === 1; + }); + const eventId = await conductor.emit("feature.wait", { id: "event" }); + await waitForCondition(async () => { + const rows = await db.sql<{ completed_at: Date | null }[]>` + select completed_at from pgconductor._private_executions where id = ${invocation} + `; + return Boolean(rows[0]?.completed_at); + }); + + const invocationSend = namedSpans(exporter, "send default").find( + (span) => span.attributes["pgconductor.task.name"] === "feature.waiter", + )!; + const eventSend = namedSpans(exporter, "send event feature.wait")[0]!; + const processes = namedSpans(exporter, "process default"); + const resumed = processes.find((process) => process.links.length > 0)!; + expect(resumed).toBeTruthy(); + expect(resumed.parentSpanId).toBe(invocationSend.spanContext().spanId); + expect(resumed.spanContext().traceId).toBe(invocationSend.spanContext().traceId); + expect(resumed.links.map((link) => link.context.spanId)).toContain( + eventSend.spanContext().spanId, + ); + expect(eventId).toBeTruthy(); + await cleanupProvider(provider); + }); + + test("a matched wait followed by a timeout has no stale event link (real Postgres)", async () => { + const db = await postgresDatabases.child(); + postgresChildren.push(db); + const event = defineEvent({ + name: "feature.match-timeout", + payload: z.object({ id: z.string() }), + }); + const taskDefinition = defineTask({ + name: "feature.match-timeout-task", + payload: z.object({}), + }); + const { exporter, provider } = installProvider(); + const conductor = Conductor.create({ + sql: db.sql, + tasks: TaskSchemas.fromSchema([taskDefinition]), + events: EventSchemas.fromSchema([event]), + context: {}, + }); + const task = conductor.createTask( + { name: "feature.match-timeout-task" }, + { invocable: true }, + async (_taskEvent, ctx) => { + await ctx.waitForEvent("matched", { event }); + try { + await ctx.waitForEvent("timed-out", { event, timeout: "20ms" }); + } catch { + // The second wait intentionally times out. + } + }, + ); + const orchestrator = Orchestrator.create({ + conductor, + tasks: [task], + defaultWorker: { pollIntervalMs: 10, flushIntervalMs: 10 }, + }); + activeOrchestrators.add(orchestrator); + await orchestrator.start(); + const invocation = await conductor.invoke({ name: task.name }, {}); + await waitForCondition(async () => { + const rows = await db.sql<{ n: number }[]>` + select count(*)::int as n from pgconductor._private_event_subscriptions + where execution_id = ${invocation}::uuid + `; + return Number(rows[0]?.n || 0) === 1; + }); + await conductor.emit(event.name, { id: "matched" }); + await waitForCondition(async () => { + const rows = await db.sql<{ step_key: string }[]>` + select step_key from pgconductor._private_event_subscriptions + where execution_id = ${invocation}::uuid + `; + return rows[0]?.step_key === "timed-out"; + }); + await waitForCondition(async () => { + const rows = await db.sql<{ completed_at: Date | null }[]>` + select completed_at from pgconductor._private_executions where id = ${invocation}::uuid + `; + return Boolean(rows[0]?.completed_at); + }); + + const processes = namedSpans(exporter, "process default").filter( + (span) => span.attributes["pgconductor.task.name"] === task.name, + ); + const eventSend = namedSpans(exporter, "send event feature.match-timeout")[0]!; + const resumed = processes.find((span) => + span.links.some((link) => link.context.spanId === eventSend.spanContext().spanId), + )!; + expect(resumed).toBeTruthy(); + expect(processes.at(-1)?.links).toHaveLength(0); + await cleanupProvider(provider); + }, 60_000); + + test("an event link is consumed by the first resumed attempt", async () => { + const { exporter, provider } = installProvider(); + const db = new InMemoryDatabaseClient(); + const event = defineEvent({ + name: "feature.retry-wait", + payload: z.object({ id: z.string() }), + }); + let resumes = 0; + const task = makeTask( + "feature.retry-waiter", + async (_event, ctx) => { + await ctx.waitForEvent("event", { event }); + if (++resumes === 1) throw new Error("retry after event"); + }, + { maxAttempts: 2 }, + ); + + const invocation = externalProducer("retry-wait invocation"); + await db.invoke({ + task_key: task.name, + queue: "default", + payload: {}, + trace_context: invocation.carrier, + }); + endSpan(invocation.span); + await makeWorker(db, [task], true, [event]).drain("register-wait"); + + const eventProducer = externalProducer("retry-wait event"); + await db.emitEvent({ + eventKey: event.name, + payload: { id: "event" }, + trace_context: eventProducer.carrier, + }); + endSpan(eventProducer.span); + await makeWorker(db, [task], true, [event]).drain("resume-wait"); + + const execution = db.getAllExecutions().find((item) => item.task_key === task.name)!; + await db.setFakeTime({ date: new Date(execution.run_at.getTime() + 1) }); + await makeWorker(db, [task], true, [event]).drain("retry-wait"); + + const processes = namedSpans(exporter, "process default").filter( + (span) => span.attributes["pgconductor.task.name"] === task.name, + ); + const eventSpanId = eventProducer.span?.spanContext().spanId; + expect( + processes.filter((span) => span.links.some((link) => link.context.spanId === eventSpanId)), + ).toHaveLength(1); + expect(processes.at(-1)?.links).toHaveLength(0); + await cleanupProvider(provider); + }); + + test("child invocation creates producer-parented child process spans", async () => { + const { exporter, provider } = installProvider(); + const db = new InMemoryDatabaseClient(); + const child = makeTask("child-feature", async () => undefined); + const parent = makeTask("parent-feature", async (_event, ctx) => + ctx.invoke("child", child, {}), + ); + await db.invoke({ task_key: parent.name, queue: "default", payload: {} }); + await makeWorker(db, [parent, child]).drain("child-first"); + await makeWorker(db, [parent, child]).drain("child-second"); + + const parentProcess = namedSpans(exporter, "process default")[0]!; + const childSend = namedSpans(exporter, "send default").find( + (span) => span.attributes["pgconductor.task.name"] === child.name, + )!; + const childProcess = namedSpans(exporter, "process default").find( + (span) => span.attributes["pgconductor.task.name"] === child.name, + )!; + expect(childSend.parentSpanId).toBe(parentProcess.spanContext().spanId); + expect(childProcess.parentSpanId).toBe(childSend.spanContext().spanId); + expect(childProcess.spanContext().traceId).toBe(parentProcess.spanContext().traceId); + await cleanupProvider(provider); + }); + + test("child and parent dead letters preserve distinct producer carriers (real Postgres)", async () => { + const db = await postgresDatabases.child(); + postgresChildren.push(db); + const payload = z.object({}); + const definitions = [ + defineTask({ name: "feature.dlq-parent", payload }), + defineTask({ name: "feature.dlq-child", payload }), + defineTask({ name: "feature.dlq-child-target", queue: "feature.child-dlq", payload }), + defineTask({ name: "feature.dlq-parent-target", queue: "feature.parent-dlq", payload }), + ]; + const { exporter, provider } = installProvider(); + const conductor = Conductor.create({ + sql: db.sql, + tasks: TaskSchemas.fromSchema(definitions), + context: {}, + }); + const childTarget = conductor.createTask( + { name: "feature.dlq-child-target", queue: "feature.child-dlq" }, + { invocable: true }, + async () => undefined, + ); + const parentTarget = conductor.createTask( + { name: "feature.dlq-parent-target", queue: "feature.parent-dlq" }, + { invocable: true }, + async () => undefined, + ); + const child = conductor.createTask( + { + name: "feature.dlq-child", + maxAttempts: 1, + deadLetter: { queue: "feature.child-dlq", task: childTarget }, + }, + { invocable: true }, + async () => { + throw new Error("child terminal failure"); + }, + ); + const parent = conductor.createTask( + { + name: "feature.dlq-parent", + maxAttempts: 1, + deadLetter: { queue: "feature.parent-dlq", task: parentTarget }, + }, + { invocable: true }, + async (_event, ctx) => ctx.invoke("child", child, {}), + ); + const sourceOrchestrator = Orchestrator.create({ + conductor, + tasks: [parent, child], + defaultWorker: { pollIntervalMs: 10, flushIntervalMs: 10 }, + }); + activeOrchestrators.add(sourceOrchestrator); + await sourceOrchestrator.start(); + await conductor.invoke({ name: parent.name }, {}); + await waitForCondition(async () => { + const rows = await db.sql<{ queue: string; count: string }[]>` + select queue, count(*)::text as count from pgconductor._private_executions + where queue in ('feature.child-dlq', 'feature.parent-dlq') group by queue + `; + return rows.length === 2 && rows.every((row) => row.count === "1"); + }); + await sourceOrchestrator.stop(); + + const destinationOrchestrator = Orchestrator.create({ + conductor, + workers: [ + conductor.createWorker({ + queue: "feature.child-dlq", + tasks: [childTarget], + config: { pollIntervalMs: 10, flushIntervalMs: 10 }, + }), + conductor.createWorker({ + queue: "feature.parent-dlq", + tasks: [parentTarget], + config: { pollIntervalMs: 10, flushIntervalMs: 10 }, + }), + ], + }); + activeOrchestrators.add(destinationOrchestrator); + await destinationOrchestrator.start(); + await waitForCondition(async () => { + const rows = await db.sql<{ count: string }[]>` + select count(*)::text as count from pgconductor._private_executions + where queue in ('feature.child-dlq', 'feature.parent-dlq') and completed_at is not null + `; + return rows[0]?.count === "2"; + }); + + const sends = namedSpans(exporter, "send feature.child-dlq").filter( + (span) => span.attributes["pgconductor.task.name"] === childTarget.name, + ); + const parentSends = namedSpans(exporter, "send feature.parent-dlq").filter( + (span) => span.attributes["pgconductor.task.name"] === parentTarget.name, + ); + const childProcess = namedSpans(exporter, "process feature.child-dlq").find( + (span) => span.attributes["pgconductor.task.name"] === childTarget.name, + ); + const parentProcess = namedSpans(exporter, "process feature.parent-dlq").find( + (span) => span.attributes["pgconductor.task.name"] === parentTarget.name, + ); + expect(sends).toHaveLength(1); + expect(parentSends).toHaveLength(1); + expect(childProcess?.parentSpanId).toBe(sends[0]?.spanContext().spanId); + expect(parentProcess?.parentSpanId).toBe(parentSends[0]?.spanContext().spanId); + expect(childProcess?.parentSpanId).not.toBe(parentProcess?.parentSpanId); + await cleanupProvider(provider); + }); + + test("initial and recurring cron executions use producer then consumer carriers", async () => { + const { exporter, provider } = installProvider(); + const now = new Date("2024-01-01T00:00:00.000Z"); + const db = new InMemoryDatabaseClient(now); + const cron = makeTask( + "cron-feature", + async () => undefined, + {}, + { cron: "* * * * *", name: "minute" }, + ); + const worker = makeWorker(db, [cron]); + await worker.drain("cron-register"); + db.advanceTime(60_000); + await worker.drain("cron-first"); + db.advanceTime(60_000); + await worker.drain("cron-second"); + + const sends = namedSpans(exporter, "send default").filter( + (span) => span.attributes["pgconductor.task.name"] === cron.name, + ); + const processes = namedSpans(exporter, "process default").filter( + (span) => span.attributes["pgconductor.task.name"] === cron.name, + ); + expect(processes.length).toBeGreaterThanOrEqual(2); + expect(sends.length).toBeGreaterThanOrEqual(3); + expect(processes[0]!.parentSpanId).toBe(sends[0]!.spanContext().spanId); + expect(processes[1]!.parentSpanId).toBe(sends[1]!.spanContext().spanId); + await cleanupProvider(provider); + }); + + test("final DLQ has a destination producer and consumer, but retry/cancel/no-DLQ do not", async () => { + const { exporter, provider } = installProvider(); + const db = new InMemoryDatabaseClient(); + const dlq = makeTask("feature-dlq", async () => undefined); + const final = makeTask( + "feature-final", + async () => { + throw new Error("terminal"); + }, + { maxAttempts: 1, deadLetter: { queue: "default", task: dlq } }, + ); + const retry = makeTask( + "feature-retry", + async () => { + throw new Error("retry"); + }, + { maxAttempts: 2 }, + ); + const cancel = makeTask("feature-cancel", async () => undefined, { maxAttempts: 1 }); + const noDlq = makeTask( + "feature-no-dlq", + async () => { + throw new Error("failure"); + }, + { maxAttempts: 1 }, + ); + const cancelledId = await db.invoke({ task_key: cancel.name, queue: "default", payload: {} }); + await db.cancelExecution(cancelledId!, { reason: "cancelled" }); + await db.invoke({ task_key: final.name, queue: "default", payload: {} }); + await db.invoke({ task_key: retry.name, queue: "default", payload: {} }); + await db.invoke({ task_key: noDlq.name, queue: "default", payload: {} }); + const worker = makeWorker(db, [dlq, final, retry, cancel, noDlq]); + await worker.drain("lifecycle"); + + const dlqSend = namedSpans(exporter, "send default").find( + (span) => span.attributes["pgconductor.task.name"] === dlq.name, + ); + const dlqProcess = namedSpans(exporter, "process default").find( + (span) => span.attributes["pgconductor.task.name"] === dlq.name, + ); + expect(dlqSend).toBeTruthy(); + expect(dlqProcess).toBeTruthy(); + expect(dlqProcess?.parentSpanId).toBe(dlqSend?.spanContext().spanId); + expect( + namedSpans(exporter, "send default").filter((span) => + [retry.name, cancel.name, noDlq.name].includes( + String(span.attributes["pgconductor.task.name"]), + ), + ), + ).toHaveLength(0); + await cleanupProvider(provider); + }); + + test("dedupe replacement propagates the latest accepted producer context", async () => { + const { exporter, provider } = installProvider(); + const db = new InMemoryDatabaseClient(); + const task = makeTask("feature-dedupe", async () => undefined); + const first = externalProducer("accepted-first"); + const firstId = await runWithSpan(first.span, () => + db.invoke({ + task_key: task.name, + queue: "default", + payload: { value: 1 }, + dedupe_key: "same", + trace_context: first.carrier, + }), + ); + endSpan(first.span); + const second = externalProducer("accepted-latest"); + await runWithSpan(second.span, () => + db.invoke({ + task_key: task.name, + queue: "default", + payload: { value: 2 }, + dedupe_key: "same", + trace_context: second.carrier, + }), + ); + endSpan(second.span); + await makeWorker(db, [task]).drain("dedupe"); + const process = namedSpans(exporter, "process default")[0]!; + expect(process.parentSpanId).toBe(second.span?.spanContext().spanId); + expect(process.parentSpanId).not.toBe(first.span?.spanContext().spanId); + expect(db.getExecution(firstId!)?.payload).toEqual({ value: 2 }); + await cleanupProvider(provider); + }); + + test("telemetry false suppresses event, child, cron, DLQ, and carrier spans", async () => { + const { exporter, provider } = installProvider(); + const db = new InMemoryDatabaseClient(new Date("2024-01-01T00:00:00.000Z")); + const child = makeTask("disabled-child", async () => undefined); + const parent = makeTask("disabled-parent", async (_event, ctx) => + ctx.invoke("child", child, {}), + ); + const bad = makeTask( + "disabled-bad", + async () => { + throw new Error("bad"); + }, + { maxAttempts: 1, deadLetter: { queue: "default", task: child } }, + ); + await db.invoke({ task_key: parent.name, queue: "default", payload: {} }); + await db.invoke({ task_key: bad.name, queue: "default", payload: {} }); + await makeWorker(db, [parent, child, bad], false).drain("disabled"); + expect(exporter.getFinishedSpans()).toHaveLength(0); + expect(db.getAllExecutions().every((execution) => execution.trace_context == null)).toBe(true); + await cleanupProvider(provider); + }); + + test("settlement failure still closes durable producer spans", async () => { + const { exporter, provider } = installProvider(); + const db = new InMemoryDatabaseClient(); + const child = makeTask("settle-child", async () => undefined); + const parent = makeTask("settle-parent", async (_event, ctx) => ctx.invoke("child", child, {})); + await db.invoke({ task_key: parent.name, queue: "default", payload: {} }); + (db as any).returnExecutions = async () => { + throw new Error("settlement failed"); + }; + await makeWorker(db, [parent, child]).drain("settle-failure"); + expect( + namedSpans(exporter, "send default").some( + (span) => span.attributes["pgconductor.task.name"] === child.name, + ), + ).toBe(true); + expect(exporter.getFinishedSpans().every((span) => span.endTime[0] !== 0)).toBe(true); + await cleanupProvider(provider); + }); +}); + +let postgresDatabases: TestDatabasePool; +const postgresChildren: TestDatabase[] = []; +beforeAll(async () => { + postgresDatabases = await TestDatabasePool.create(); +}, 60000); +afterEach(async () => { + await Promise.all([...activeOrchestrators].map((orchestrator) => orchestrator.stop())); + activeOrchestrators.clear(); + await Promise.all([...activeProviders].map((provider) => cleanupProvider(provider))); + await Promise.all(postgresChildren.map((db) => db.destroy())); + postgresChildren.length = 0; +}); +afterAll(async () => { + await postgresDatabases?.destroy(); +}); diff --git a/packages/pgconductor-js/tests/integration/telemetry-metrics.test.ts b/packages/pgconductor-js/tests/integration/telemetry-metrics.test.ts new file mode 100644 index 0000000..9bb7ba8 --- /dev/null +++ b/packages/pgconductor-js/tests/integration/telemetry-metrics.test.ts @@ -0,0 +1,294 @@ +import { afterEach, describe, expect, test } from "bun:test"; +import { metrics } from "@opentelemetry/api"; +import { MeterProvider, MetricReader } from "@opentelemetry/sdk-metrics"; +import { Conductor } from "../../src/conductor"; +import { Task } from "../../src/task"; +import { Worker } from "../../src/worker"; +import { DefaultLogger } from "../../src/lib/logger"; +import { InMemoryDatabaseClient } from "../mocks/in-memory-database-client"; +import { resetMetricsForTests } from "../../src/telemetry"; + +const logger = new DefaultLogger(); + +// sdk-metrics 1.x calls this MetricReader; keep the test's reader manual and +// synchronous while remaining compatible with the version supported here. +class ManualMetricReader extends MetricReader { + protected async onShutdown(): Promise {} + protected async onForceFlush(): Promise {} +} + +const fakeSql = Object.assign((async () => []) as any, { + json: (value: unknown) => JSON.stringify(value), +}); + +let provider: MeterProvider | undefined; + +function installMetrics() { + metrics.disable(); + resetMetricsForTests(); + const reader = new ManualMetricReader(); + provider = new MeterProvider({ readers: [reader] }); + expect(metrics.setGlobalMeterProvider(provider)).toBe(true); + return reader; +} + +async function collect(reader: ManualMetricReader) { + await provider?.forceFlush(); + return (await reader.collect()).resourceMetrics; +} + +async function cleanupMetrics() { + await provider?.shutdown(); + provider = undefined; + metrics.disable(); + resetMetricsForTests(); +} + +afterEach(async () => { + await cleanupMetrics(); +}); + +function makeTask( + name: string, + execute: (event: any, context: any) => Promise, + config: Record = {}, +) { + return Task.create({ name, ...config } as any, { invocable: true } as any, execute as any); +} + +function makeWorker(db: InMemoryDatabaseClient, tasks: any[], telemetry = true) { + return new Worker( + "default", + tasks, + db, + logger, + { + pollIntervalMs: 1, + flushIntervalMs: 1, + fetchBatchSize: 10, + flushBatchSize: 10, + }, + {}, + [], + telemetry, + ); +} + +function allMetrics(resourceMetrics: any) { + return resourceMetrics.scopeMetrics.flatMap((scope: any) => scope.metrics); +} + +function metric(resourceMetrics: any, name: string): any { + return allMetrics(resourceMetrics).find((candidate: any) => candidate.descriptor.name === name); +} + +function metricValue(resourceMetrics: any, name: string, attributes: Record = {}) { + const candidate = metric(resourceMetrics, name); + if (!candidate) return 0; + return candidate.dataPoints + .filter((point: any) => + Object.entries(attributes).every(([key, value]) => point.attributes[key] === value), + ) + .reduce( + (total: number, point: any) => + total + + Number( + typeof point.value === "number" ? point.value : (point.value?.sum ?? point.sum ?? 0), + ), + 0, + ); +} + +function lifecycleTotal(resourceMetrics: any) { + return [ + "pgconductor.execution.retries", + "pgconductor.execution.permanent_failures", + "pgconductor.execution.cancellations", + "pgconductor.execution.dead_letters", + ].reduce((total, name) => total + metricValue(resourceMetrics, name), 0); +} + +describe.serial("OpenTelemetry metrics instrumentation", () => { + test("counts accepted invokes but not throttled no-ops", async () => { + const reader = installMetrics(); + const db = new InMemoryDatabaseClient(); + const conductor = Conductor.create({ sql: fakeSql, context: {} }); + (conductor as any).db = db; + + const task = { name: "invoke-metrics" }; + expect(await (conductor as any).invoke(task, {}, { throttle: { seconds: 60 } })).toBeTruthy(); + expect(await (conductor as any).invoke(task, {}, { throttle: { seconds: 60 } })).toBeNull(); + + const resourceMetrics = await collect(reader); + expect(metricValue(resourceMetrics, "messaging.client.sent.messages")).toBe(1); + }); + + test("records worker consumption, handler/process duration, and settle duration", async () => { + const reader = installMetrics(); + const db = new InMemoryDatabaseClient(); + const task = makeTask("worker-metrics", async () => undefined); + await db.invoke({ task_key: task.name, queue: "default", payload: {} }); + await makeWorker(db, [task]).drain("metrics-worker"); + + const resourceMetrics = await collect(reader); + expect(metricValue(resourceMetrics, "messaging.client.consumed.messages")).toBe(1); + expect(metricValue(resourceMetrics, "messaging.process.duration")).toBeGreaterThan(0); + expect( + metricValue(resourceMetrics, "messaging.client.operation.duration", { + "messaging.operation.name": "settle", + }), + ).toBeGreaterThan(0); + }); + + test("records retry and permanent failure only after returnExecutions commits", async () => { + const reader = installMetrics(); + const db = new InMemoryDatabaseClient(); + const task = makeTask( + "lifecycle-metrics", + async () => { + throw new Error("always fails"); + }, + { maxAttempts: 2 }, + ); + await db.invoke({ task_key: task.name, queue: "default", payload: {} }); + + let entered!: () => void; + const returnEntered = new Promise((resolve) => (entered = resolve)); + let release!: () => void; + const commit = new Promise((resolve) => (release = resolve)); + const originalReturn = db.returnExecutions.bind(db); + (db as any).returnExecutions = async (results: any, options: any) => { + entered(); + await commit; + return originalReturn(results, options); + }; + + const firstDrain = makeWorker(db, [task]).drain("metrics-worker"); + await returnEntered; + let resourceMetrics = await collect(reader); + expect(lifecycleTotal(resourceMetrics)).toBe(0); + + release(); + await firstDrain; + resourceMetrics = await collect(reader); + expect(metricValue(resourceMetrics, "pgconductor.execution.retries")).toBe(1); + + db.advanceTime(16_000); + await makeWorker(db, [task]).drain("metrics-worker-2"); + resourceMetrics = await collect(reader); + expect(metricValue(resourceMetrics, "pgconductor.execution.retries")).toBe(1); + expect(metricValue(resourceMetrics, "pgconductor.execution.permanent_failures")).toBe(1); + }); + + test("failed returnExecutions records no lifecycle metrics", async () => { + const reader = installMetrics(); + const db = new InMemoryDatabaseClient(); + const task = makeTask("settlement-failure", async () => undefined); + await db.invoke({ task_key: task.name, queue: "default", payload: {} }); + (db as any).returnExecutions = async () => { + throw new Error("settlement failed"); + }; + + await makeWorker(db, [task]).drain("metrics-worker"); + const resourceMetrics = await collect(reader); + expect(lifecycleTotal(resourceMetrics)).toBe(0); + }); + + test("records cancellation and final failure plus DLQ lifecycle counts", async () => { + const reader = installMetrics(); + const db = new InMemoryDatabaseClient(); + const dlqTask = makeTask("metrics-dlq", async () => undefined); + const cancelledTask = makeTask("metrics-cancelled", async () => undefined, { + maxAttempts: 1, + }); + const failedTask = makeTask( + "metrics-dead-letter", + async () => { + throw new Error("terminal failure"); + }, + { maxAttempts: 1, deadLetter: { queue: "default", task: dlqTask } }, + ); + const cancelledId = await db.invoke({ + task_key: cancelledTask.name, + queue: "default", + payload: {}, + }); + await db.invoke({ + task_key: failedTask.name, + queue: "default", + payload: {}, + }); + await db.cancelExecution(cancelledId!, { reason: "cancelled for metrics" }); + + await makeWorker(db, [cancelledTask, failedTask, dlqTask]).drain("metrics-worker"); + const resourceMetrics = await collect(reader); + expect(metricValue(resourceMetrics, "pgconductor.execution.cancellations")).toBe(1); + expect(metricValue(resourceMetrics, "pgconductor.execution.permanent_failures")).toBe(1); + expect(metricValue(resourceMetrics, "pgconductor.execution.dead_letters")).toBe(1); + expect( + metric(resourceMetrics, "pgconductor.execution.permanent_failures")?.dataPoints.some( + (point: any) => + point.attributes["pgconductor.task.name"] === failedTask.name && + point.attributes["pgconductor.outcome"] === "permanent_failure", + ), + ).toBe(true); + }); + + test("telemetry false records no metrics for conductor or worker", async () => { + const reader = installMetrics(); + const db = new InMemoryDatabaseClient(); + const conductor = Conductor.create({ + sql: fakeSql, + context: {}, + telemetry: false, + }); + (conductor as any).db = db; + const task = makeTask("telemetry-disabled", async () => undefined); + const id = await conductor.invoke(task, {} as any); + expect(id).toBeTruthy(); + await makeWorker(db, [task], false).drain("metrics-worker"); + + expect(allMetrics(await collect(reader))).toHaveLength(0); + }); + + test("uses seconds for plausible duration values and safe metric attributes", async () => { + const reader = installMetrics(); + const db = new InMemoryDatabaseClient(); + const task = makeTask("attribute-metrics", async () => undefined); + await db.invoke({ + task_key: task.name, + queue: "default", + payload: { secret: "no metric" }, + }); + await makeWorker(db, [task]).drain("metrics-worker"); + const resourceMetrics = await collect(reader); + const allowed = new Set([ + "messaging.system", + "messaging.destination.name", + "messaging.operation.name", + "messaging.operation.type", + "pgconductor.task.name", + "pgconductor.outcome", + ]); + + for (const candidate of allMetrics(resourceMetrics)) { + for (const point of candidate.dataPoints) { + expect(Object.keys(point.attributes).every((key) => allowed.has(key))).toBe(true); + expect(point.attributes["messaging.system"]).toBe("postgres_conductor"); + expect(point.attributes["messaging.destination.name"]).toBe("default"); + expect(point.attributes["messaging.operation.name"]).toBeTruthy(); + if (candidate.descriptor.name.includes("duration")) { + expect(candidate.descriptor.unit).toBe("s"); + const duration = Number( + typeof point.value === "number" ? point.value : (point.value?.sum ?? point.sum ?? 0), + ); + expect(duration).toBeGreaterThan(0); + expect(duration).toBeLessThan(10); + } + } + } + const processPoint = metric(resourceMetrics, "messaging.process.duration")?.dataPoints[0]; + expect(processPoint.attributes["pgconductor.task.name"]).toBe(task.name); + expect(JSON.stringify(resourceMetrics)).not.toContain("secret"); + }); +}); diff --git a/packages/pgconductor-js/tests/integration/telemetry.test.ts b/packages/pgconductor-js/tests/integration/telemetry.test.ts index 5ada2e7..7786d90 100644 --- a/packages/pgconductor-js/tests/integration/telemetry.test.ts +++ b/packages/pgconductor-js/tests/integration/telemetry.test.ts @@ -23,7 +23,6 @@ import { Task } from "../../src/task"; import { Worker } from "../../src/worker"; import { DefaultLogger } from "../../src/lib/logger"; import { InMemoryDatabaseClient } from "../mocks/in-memory-database-client"; -import type { DatabaseClient } from "../../src/database-client"; import { boundedCarrier, carrierForContext, @@ -100,7 +99,7 @@ function makeWorker(db: InMemoryDatabaseClient, task: any, telemetry = true) { return new Worker( "default", [task], - db as unknown as DatabaseClient, + db, logger, { pollIntervalMs: 1, flushIntervalMs: 1, fetchBatchSize: 10, flushBatchSize: 10 }, {}, @@ -307,7 +306,7 @@ describe.serial("OpenTelemetry instrumentation", () => { const eventWorker = new Worker( "default", [makeTask("root-cron", async () => undefined), makeTask("root-event", async () => undefined)], - db as unknown as DatabaseClient, + db, logger, { pollIntervalMs: 1, flushIntervalMs: 1 }, {}, diff --git a/packages/pgconductor-js/tests/integration/wait-for-event.test.ts b/packages/pgconductor-js/tests/integration/wait-for-event.test.ts index 93734b4..dff8eb7 100644 --- a/packages/pgconductor-js/tests/integration/wait-for-event.test.ts +++ b/packages/pgconductor-js/tests/integration/wait-for-event.test.ts @@ -7,6 +7,7 @@ import { defineTask } from "../../src/task-definition"; import { EventSchemas, TaskSchemas } from "../../src/schemas"; import { WaitForEventTimeoutError } from "../../src/index"; import { TestDatabasePool, type TestDatabase } from "../fixtures/test-database"; +import { waitForCondition } from "../test-utils"; const event = defineEvent({ name: "wait.order", @@ -17,15 +18,6 @@ const taskDefinition = defineTask({ name: "wait.task", payload: z.object({ id: z type Handler = (id: string, ctx: any) => Promise; -async function until(check: () => Promise, timeout = 20_000) { - const end = Date.now() + timeout; - while (Date.now() < end) { - if (await check()) return; - await Bun.sleep(10); - } - throw new Error("condition was not met"); -} - async function setup(db: TestDatabase, fn: Handler, orchestrators: Orchestrator[] = []) { const conductor = Conductor.create({ sql: db.sql, @@ -78,7 +70,7 @@ describe.serial("waitForEvent", () => { } async function waiting(database: TestDatabase, count = 1) { - await until( + await waitForCondition( async () => Number( ( @@ -119,7 +111,9 @@ describe.serial("waitForEvent", () => { await Bun.sleep(100); expect(entries).toEqual(["entered:one", "entered:two"]); await conductor.emit("wait.order", { id: "yes", kind: "match" }); - await until(async () => entries.filter((entry) => entry.startsWith("replay:")).length === 2); + await waitForCondition( + async () => entries.filter((entry) => entry.startsWith("replay:")).length === 2, + ); expect(entries).toEqual([ "entered:one", "entered:two", @@ -152,7 +146,7 @@ describe.serial("waitForEvent", () => { ); await conductor.invoke({ name: "wait.task" }, { id: "one" }); - await until(async () => errors.length === 1); + await waitForCondition(async () => errors.length === 1); expect(entered).toEqual(["one", "one"]); expect(errors).toHaveLength(1); expect(errors[0]).toBeInstanceOf(WaitForEventTimeoutError); @@ -173,7 +167,7 @@ describe.serial("waitForEvent", () => { await conductor.invoke({ name: "wait.task" }, { id: "race" }); await waiting(database); await conductor.emit("wait.order", { id: "race", kind: "match" }); - await until(async () => result.length === 1); + await waitForCondition(async () => result.length === 1); expect(result).toEqual(["race"]); }); @@ -196,7 +190,7 @@ describe.serial("waitForEvent", () => { await first.orchestrator.stop(); const second = await setup(database, handler, orchestrators); await second.conductor.emit("wait.order", { id: "new", kind: "match" }); - await until(async () => result.length === 1); + await waitForCondition(async () => result.length === 1); expect(result).toEqual(["new"]); expect(executionId).toBeTruthy(); }); @@ -216,7 +210,7 @@ describe.serial("waitForEvent", () => { await first.orchestrator.stop(); const second = await setup(database, handler, orchestrators); await second.conductor.emit("wait.order", { id: "restart", kind: "match" }); - await until(async () => result.length === 1); + await waitForCondition(async () => result.length === 1); expect(entered).toEqual(["restart", "restart"]); expect(result).toEqual(["restart"]); }); @@ -242,7 +236,7 @@ describe.serial("waitForEvent", () => { await conductor.emit("wait.order", { id: "second", kind: "match" }); await Bun.sleep(5); await conductor.emit("wait.order", { id: "third", kind: "match" }); - await until(async () => result.length === 1); + await waitForCondition(async () => result.length === 1); expect(result).toEqual(["first"]); }); @@ -353,7 +347,7 @@ describe.serial("waitForEvent", () => { await conductor.invoke({ name: "wait.task" }, { id: "outcome" }); await waiting(database); await conductor.emit("wait.order", { id: "outcome", kind: "match" }); - await until(async () => outcomes.length === 1); + await waitForCondition(async () => outcomes.length === 1); await Bun.sleep(150); expect(outcomes).toEqual(["match:outcome"]); expect( @@ -415,7 +409,7 @@ describe.serial("waitForEvent", () => { ), ).toBe(1); await conductor.emit("wait.order", { id: "once", kind: "match" }); - await until(async () => result.length === 1); + await waitForCondition(async () => result.length === 1); await Bun.sleep(100); expect(entered).toEqual(["once", "once"]); expect(result).toEqual(["once"]); diff --git a/packages/pgconductor-js/tests/mocks/database-client.mock.ts b/packages/pgconductor-js/tests/mocks/database-client.mock.ts index 7b4e000..70f1731 100644 --- a/packages/pgconductor-js/tests/mocks/database-client.mock.ts +++ b/packages/pgconductor-js/tests/mocks/database-client.mock.ts @@ -1,11 +1,11 @@ import { mock } from "bun:test"; -import type { DatabaseClient } from "../../src/database-client"; +import type { + CronRegistration, + DatabaseClientLike, + ReturnExecutionsResult, +} from "../../src/database-client"; -type PublicMethodsOf = { - [K in keyof T as T[K] extends Function ? K : never]: T[K]; -}; - -type IDatabaseClient = PublicMethodsOf; +type IDatabaseClient = DatabaseClientLike; export class MockDatabaseClient implements IDatabaseClient { close = mock(async () => {}); @@ -17,10 +17,21 @@ export class MockDatabaseClient implements IDatabaseClient { countActiveOrchestratorsBelow = mock(async () => 0); orchestratorShutdown = mock(async () => {}); getExecutions = mock(async () => []); - returnExecutions = mock(async (_results) => {}); + returnExecutions = mock( + async ( + _results: Parameters[0], + ): Promise => ({ + outcomes: [], + deliveries: [], + }), + ); removeExecutions = mock(async () => false); removeProcessedEvents = mock(async () => false); - registerWorker = mock(async () => {}); + registerWorker = mock( + async ( + _args: Parameters[0], + ): Promise => [], + ); scheduleCronExecution = mock(async () => "mock-cron-id"); unscheduleCronExecution = mock(async () => {}); invoke = mock(async () => "mock-id"); 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 67b5a28..25bf70b 100644 --- a/packages/pgconductor-js/tests/mocks/in-memory-database-client.ts +++ b/packages/pgconductor-js/tests/mocks/in-memory-database-client.ts @@ -1,5 +1,7 @@ import type { + CronRegistration, DatabaseClient, + DeadLetterDelivery, Execution, ExecutionResult, ExecutionSpec, @@ -7,6 +9,7 @@ import type { Payload, JsonValue, SetFakeTimeArgs, + SettlementOutcome, // EventSubscriptionSpec, } from "../../src/database-client"; import { DatabaseClient as RealDatabaseClient } from "../../src/database-client"; @@ -25,7 +28,7 @@ import type { ClearWaitingStateArgs, RegisterEventWaitArgs, OrchestratorShutdownArgs, - // EmitEventArgs, + EmitEventArgs, } from "../../src/query-builder"; import type { Migration } from "../../src/migration-store"; import type { Logger } from "../../src/lib/logger"; @@ -70,6 +73,7 @@ interface StoredExecution { dead_letter_attempts: number | null; dead_letter_failed_at: Date | null; trace_context: Execution["trace_context"]; + trace_link_context: Execution["trace_link_context"]; } interface StoredStep { @@ -125,6 +129,7 @@ interface StoredCustomEvent { id: string; event_key: string; payload: JsonValue; + trace_context: Execution["trace_context"]; created_at: Date; processed_at: Date | null; event_position: number; @@ -323,7 +328,10 @@ export class InMemoryDatabaseClient implements IDatabaseClient { // Worker Registration // ============================================================================ - async registerWorker(args: RegisterWorkerArgs, _opts?: { signal?: AbortSignal }): Promise { + async registerWorker( + args: RegisterWorkerArgs, + _opts?: { signal?: AbortSignal }, + ): Promise { // Register tasks for (const taskSpec of args.taskSpecs) { const task: StoredTask = { @@ -344,8 +352,10 @@ export class InMemoryDatabaseClient implements IDatabaseClient { // Registration is authoritative for this queue, just like PostgreSQL. for (const id of [...this.eventSubscriptions.keys()]) { - if (this.eventSubscriptions.get(id)?.queue === args.queueName) + const subscription = this.eventSubscriptions.get(id); + if (subscription?.queue === args.queueName && !subscription.execution_id) { this.eventSubscriptions.delete(id); + } } for (const spec of args.eventSubscriptions || []) { const id = this.generateId(); @@ -358,19 +368,35 @@ export class InMemoryDatabaseClient implements IDatabaseClient { }); } - // Register cron schedules (ExecutionSpec[]) + const registrations: CronRegistration[] = []; for (const cronSpec of args.cronSchedules || []) { - if (cronSpec.cron_expression) { - const key = `${cronSpec.task_key}:${cronSpec.cron_expression}`; - this.cronSchedules.set(key, { - task_key: cronSpec.task_key, - queue: cronSpec.queue, - schedule_name: cronSpec.cron_expression, - cron_expression: cronSpec.cron_expression, - last_execution_id: null, - }); - } + if (!cronSpec.cron_expression || !cronSpec.dedupe_key) continue; + const key = `${cronSpec.task_key}:${cronSpec.cron_expression}`; + this.cronSchedules.set(key, { + task_key: cronSpec.task_key, + queue: cronSpec.queue, + schedule_name: cronSpec.cron_expression, + cron_expression: cronSpec.cron_expression, + last_execution_id: null, + }); + const existing = [...this.executions.values()].find( + (execution) => + execution.task_key === cronSpec.task_key && + execution.queue === cronSpec.queue && + execution.dedupe_key === cronSpec.dedupe_key, + ); + const id = existing?.id ?? (await this.invoke(cronSpec)); + if (!id) continue; + registrations.push({ + id, + task_key: cronSpec.task_key, + queue: cronSpec.queue, + is_maintenance: cronSpec.task_key === "pgconductor.maintenance", + inserted: !existing, + authoritative: !existing, + }); } + return registrations; } // ============================================================================ @@ -440,10 +466,12 @@ export class InMemoryDatabaseClient implements IDatabaseClient { if (activeGroup >= task.group_concurrency) continue; } - // Claim execution. + // Claim execution and consume any one-shot event link. + const traceLinkContext = exec.trace_link_context; exec.state = "running"; exec.attempts += 1; exec.orchestrator_id = args.orchestratorId; + exec.trace_link_context = null; // Update concurrency count if (task?.concurrency != null) { @@ -453,6 +481,12 @@ export class InMemoryDatabaseClient implements IDatabaseClient { ); } + const parent = exec.parent_execution_id + ? this.executions.get(exec.parent_execution_id) + : undefined; + const parentTask = parent + ? this.tasks.get(this.taskId(parent.task_key, parent.queue)) + : undefined; results.push({ id: exec.id, task_key: exec.task_key, @@ -460,6 +494,7 @@ export class InMemoryDatabaseClient implements IDatabaseClient { payload: exec.payload, waiting_on_execution_id: exec.waiting_on_execution_id, waiting_step_key: exec.waiting_step_key, + attempts: exec.attempts, cancelled: exec.cancelled, last_error: exec.last_error, dedupe_key: exec.dedupe_key || undefined, @@ -472,6 +507,12 @@ export class InMemoryDatabaseClient implements IDatabaseClient { dead_letter_attempts: exec.dead_letter_attempts, dead_letter_failed_at: exec.dead_letter_failed_at, trace_context: exec.trace_context, + trace_link_context: traceLinkContext, + parent_execution_id: exec.parent_execution_id, + parent_queue: parent?.queue, + parent_task_key: parent?.task_key, + parent_dead_letter_queue: parentTask?.dead_letter_queue, + parent_dead_letter_task_key: parentTask?.dead_letter_task_key, locked_by: exec.orchestrator_id || "", }); @@ -486,7 +527,7 @@ export class InMemoryDatabaseClient implements IDatabaseClient { | ExecutionResult[] | import("../../src/database-client").GroupedExecutionResults, _opts?: { signal?: AbortSignal }, - ): Promise { + ): Promise { // Handle both old array format (for testing) and new grouped format const results: ExecutionResult[] = Array.isArray(resultsOrGrouped) ? resultsOrGrouped @@ -499,6 +540,10 @@ export class InMemoryDatabaseClient implements IDatabaseClient { // ...resultsOrGrouped.waitForDbEvent, ]; const now = this.getInternalTime(); + const deliveries: DeadLetterDelivery[] = []; + const outcomes: SettlementOutcome[] = []; + const recordOutcome = (execution: StoredExecution, outcome: SettlementOutcome["outcome"]) => + outcomes.push({ queue: execution.queue, task_key: execution.task_key, outcome, count: 1 }); for (const result of results) { const exec = this.executions.get(result.execution_id); @@ -522,6 +567,9 @@ export class InMemoryDatabaseClient implements IDatabaseClient { if (exec.parent_execution_id) { const parent = this.executions.get(exec.parent_execution_id); if (parent && parent.waiting_on_execution_id === exec.id) { + if (exec.parent_step_key) { + this.saveWaitResult(parent, exec.parent_step_key, result.result || {}); + } parent.waiting_on_execution_id = null; parent.waiting_step_key = null; parent.waiting_timeout_at = null; @@ -553,22 +601,46 @@ export class InMemoryDatabaseClient implements IDatabaseClient { if (exec.attempts >= maxAttempts) { // Permanently failed + recordOutcome(exec, exec.cancelled ? "cancellation" : "permanent_failure"); exec.state = "failed"; exec.failed_at = now; - if (!exec.cancelled) this.deliverToDeadLetterQueue(exec, task, result.error, now); + if (!exec.cancelled) { + const delivery = this.deliverToDeadLetterQueue( + exec, + task, + result.error, + now, + result.dead_letter_trace_contexts?.[exec.id], + ); + if (delivery) { + deliveries.push(delivery); + recordOutcome(exec, "dead_letter"); + } + } // 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) { + recordOutcome(parent, exec.cancelled ? "cancellation" : "permanent_failure"); parent.state = "failed"; parent.last_error = `Child execution failed: ${result.error}`; parent.waiting_on_execution_id = null; parent.waiting_step_key = null; const parentTask = this.tasks.get(this.taskId(parent.task_key, parent.queue)); if (!exec.cancelled) { - this.deliverToDeadLetterQueue(parent, parentTask, parent.last_error, now); + const delivery = this.deliverToDeadLetterQueue( + parent, + parentTask, + parent.last_error, + now, + result.dead_letter_trace_contexts?.[parent.id], + ); + if (delivery) { + deliveries.push(delivery); + recordOutcome(parent, "dead_letter"); + } } if (parentTask?.remove_on_fail_days != null) { this.executions.delete(parent.id); @@ -584,6 +656,7 @@ export class InMemoryDatabaseClient implements IDatabaseClient { } } else { // Retry with backoff + recordOutcome(exec, "retry"); exec.state = "pending"; const backoffSeconds = this.calculateBackoff(exec.attempts); exec.run_at = new Date(now.getTime() + backoffSeconds * 1000); @@ -611,12 +684,25 @@ export class InMemoryDatabaseClient implements IDatabaseClient { } case "permanently_failed": { + recordOutcome(exec, exec.cancelled ? "cancellation" : "permanent_failure"); 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); + if (!exec.cancelled) { + const delivery = this.deliverToDeadLetterQueue( + exec, + task, + result.error, + now, + result.dead_letter_trace_contexts?.[exec.id], + ); + if (delivery) { + deliveries.push(delivery); + recordOutcome(exec, "dead_letter"); + } + } exec.orchestrator_id = null; // Fail parent if waiting. A force-failed parent is itself terminal and @@ -624,6 +710,7 @@ export class InMemoryDatabaseClient implements IDatabaseClient { if (exec.parent_execution_id) { const parent = this.executions.get(exec.parent_execution_id); if (parent && parent.waiting_on_execution_id === exec.id) { + recordOutcome(parent, exec.cancelled ? "cancellation" : "permanent_failure"); parent.state = "failed"; parent.last_error = `Child execution failed: ${result.error}`; parent.waiting_on_execution_id = null; @@ -631,7 +718,17 @@ export class InMemoryDatabaseClient implements IDatabaseClient { 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); + const delivery = this.deliverToDeadLetterQueue( + parent, + parentTask, + parent.last_error, + now, + result.dead_letter_trace_contexts?.[parent.id], + ); + if (delivery) { + deliveries.push(delivery); + recordOutcome(parent, "dead_letter"); + } } if (parentTask?.remove_on_fail_days != null) { this.executions.delete(parent.id); @@ -656,6 +753,7 @@ export class InMemoryDatabaseClient implements IDatabaseClient { group: result.group, parent_execution_id: exec.id, parent_step_key: result.step_key, + trace_context: result.trace_context, }); // Set parent to wait @@ -720,6 +818,7 @@ export class InMemoryDatabaseClient implements IDatabaseClient { exec.updated_at = now; } + return { outcomes, deliveries }; } async removeExecutions( @@ -797,6 +896,7 @@ export class InMemoryDatabaseClient implements IDatabaseClient { dead_letter_attempts: null, dead_letter_failed_at: null, trace_context: spec.trace_context || null, + trace_link_context: null, }; this.executions.set(id, execution); @@ -880,6 +980,7 @@ export class InMemoryDatabaseClient implements IDatabaseClient { exec.run_at = spec.run_at || now; exec.priority = spec.priority || 0; exec.cron_expression = spec.cron_expression || null; + exec.trace_context = spec.trace_context || null; exec.updated_at = now; return exec.id; } @@ -935,6 +1036,7 @@ export class InMemoryDatabaseClient implements IDatabaseClient { exec.priority = spec.priority || 0; exec.singleton_on = singletonOn; exec.cron_expression = spec.cron_expression || null; + exec.trace_context = spec.trace_context || null; exec.updated_at = now; ids.push(exec.id); foundExisting = true; @@ -1139,6 +1241,7 @@ export class InMemoryDatabaseClient implements IDatabaseClient { args.timeoutMs == null ? null : new Date(this.getInternalTime().getTime() + args.timeoutMs), }); exec.state = "pending"; + exec.attempts = Math.max(0, exec.attempts - 1); exec.run_at = new Date(8640000000000000); exec.waiting_on_execution_id = null; exec.waiting_step_key = args.stepKey; @@ -1203,12 +1306,13 @@ export class InMemoryDatabaseClient implements IDatabaseClient { return this.generateId(); } - async emitEvent(args: { eventKey: string; payload?: JsonValue }): Promise { + async emitEvent(args: EmitEventArgs): Promise { const id = this.generateId(); this.customEvents.set(id, { id, event_key: args.eventKey, payload: args.payload ?? {}, + trace_context: args.trace_context ?? null, created_at: this.getInternalTime(), processed_at: null, event_position: ++this.lastEventPosition, @@ -1278,6 +1382,7 @@ export class InMemoryDatabaseClient implements IDatabaseClient { exec.run_at = now; exec.waiting_step_key = null; exec.waiting_timeout_at = null; + exec.trace_link_context = event?.trace_context ?? null; this.eventSubscriptions.delete(subscription.id); count++; } @@ -1305,6 +1410,7 @@ export class InMemoryDatabaseClient implements IDatabaseClient { task_key: subscription.task_key, queue: subscription.queue, payload: { event: event.event_key, payload: structuredClone(event.payload) as Payload }, + trace_context: event.trace_context, }); } event.processed_at = now; @@ -1341,8 +1447,9 @@ export class InMemoryDatabaseClient implements IDatabaseClient { task: StoredTask | undefined, error: string, now: Date, - ): void { - if (!task?.dead_letter_queue || exec.cancelled) return; + traceContext?: Execution["trace_context"], + ): DeadLetterDelivery | null { + if (!task?.dead_letter_queue || exec.cancelled) return null; const destinationTaskKey = task.dead_letter_task_key || exec.task_key; const duplicate = Array.from(this.executions.values()).some( (destination) => @@ -1350,7 +1457,7 @@ export class InMemoryDatabaseClient implements IDatabaseClient { destination.queue === task.dead_letter_queue && destination.task_key === destinationTaskKey, ); - if (duplicate) return; + if (duplicate) return null; const id = this.generateId(); this.executions.set(id, { id, @@ -1384,8 +1491,15 @@ export class InMemoryDatabaseClient implements IDatabaseClient { dead_letter_error: error, dead_letter_attempts: exec.attempts, dead_letter_failed_at: now, - trace_context: null, + trace_context: traceContext || null, + trace_link_context: null, }); + return { + sourceExecutionId: exec.id, + destinationExecutionId: id, + destinationQueue: task.dead_letter_queue, + destinationTaskKey, + }; } // Helpers diff --git a/packages/pgconductor-js/tests/test-utils.ts b/packages/pgconductor-js/tests/test-utils.ts new file mode 100644 index 0000000..a66b926 --- /dev/null +++ b/packages/pgconductor-js/tests/test-utils.ts @@ -0,0 +1,12 @@ +export async function waitForCondition( + condition: () => boolean | Promise, + timeoutMs = 20_000, +): Promise { + const deadline = Date.now() + timeoutMs; + while (!(await condition())) { + if (Date.now() >= deadline) { + throw new Error(`condition was not met within ${timeoutMs}ms`); + } + await Bun.sleep(25); + } +} diff --git a/packages/pgconductor-js/tests/unit/worker-lifecycle.test.ts b/packages/pgconductor-js/tests/unit/worker-lifecycle.test.ts index 6da1797..d7361a0 100644 --- a/packages/pgconductor-js/tests/unit/worker-lifecycle.test.ts +++ b/packages/pgconductor-js/tests/unit/worker-lifecycle.test.ts @@ -11,14 +11,59 @@ const task = { execute: async () => {}, } as unknown as AnyTask; +test("clears dead-letter targets when a worker lifecycle resets", () => { + const worker = new Worker("default", [task], new MockDatabaseClient(), new DefaultLogger()); + const targets = (worker as any).parentDeadLetterTargets as Map; + targets.set("child", {}); + + (worker as any).resetLifecycle(); + + expect(targets.size).toBe(0); +}); + +test("keeps dead-letter targets until a retried settlement commits", async () => { + let calls = 0; + const db = new MockDatabaseClient({ + returnExecutions: async () => { + expect(targets.size).toBe(1); + calls += 1; + return calls === 1 + ? Promise.reject(new Error("transient settlement failure")) + : { outcomes: [], deliveries: [] }; + }, + }); + const worker = new Worker("default", [task], db, new DefaultLogger(), { flushBatchSize: 1 }); + const targets = (worker as any).parentDeadLetterTargets as Map; + targets.set("child", {}); + (worker as any)._abortController = new AbortController(); + (worker as any).orchestratorId = "orchestrator"; + const result = { + execution_id: "child", + orchestrator_id: "orchestrator", + queue: "default", + task_key: task.name, + status: "completed" as const, + }; + + await (worker as any).flushResults( + (async function* () { + yield result; + })(), + ); + + expect(calls).toBe(2); + expect(targets.size).toBe(0); +}); + test("resets a worker after registration failure and permits retry", async () => { let shouldFail = true; const db = new MockDatabaseClient({ registerWorker: async () => { if (shouldFail) throw new Error("registration failed"); + return []; }, }); - const worker = new Worker("default", [task], db as never, new DefaultLogger()); + const worker = new Worker("default", [task], db, new DefaultLogger()); await expect(worker.start("first-orchestrator")).rejects.toThrow("registration failed"); // A failed startup is not a running worker: stopped must be already settled,