feat(telemetry): add durable propagation and metrics

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.

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" },

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$;
 

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;

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",

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:",

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<TDef>;
 		}
 
-		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<boolean> {