Skip to content

feat(executions): typed, propagated execution metadata - #72

Merged
psteinroe merged 1 commit into
mainfrom
feat/execution-metadata
Oct 2, 2026
Merged

psteinroe merged 1 commit into
mainfrom
feat/execution-metadata

Conversation

@psteinroe

Copy link
Copy Markdown
Owner

Executions now carry user metadata: a small typed JSON object that is set when work enters the system and is inherited by everything created from it. A request, tenant or reply reference set on conductor.emit / conductor.invoke reaches every task in the workflow without being added to each payload.

const conductor = Conductor.create({ sql, tasks, events, context: {}, metadata: z.object({ tenant: z.string() }) });
await conductor.emit("slack.mentioned", payload, { metadata: { tenant: "acme" } });
await conductor.invoke(task, payload, { metadata });
await ctx.invoke("reply", task, payload, { metadata: (m) => ({ ...m, replyTo: other }) }); // this child only
ctx.metadata; // Readonly<Metadata> | undefined

Propagation

A new metadata jsonb column on _private_executions follows the same SQL paths as trace_context:

  • event → triggered executions
  • parent → ctx.invoke child
  • terminal failure → dead-letter delivery
  • cron occurrence → next occurrence

TaskContext passes its metadata explicitly to ctx.invoke, ctx.emit and ctx.schedule. Retries and resumes reuse the same row, so they keep it. Cron schedules declared as task triggers have none.

Validation and size limit

If Conductor.create({ metadata }) is given a Standard Schema, metadata is validated on conductor.invoke, conductor.emit and ctx.invoke overrides, with the same error format as event payloads. The 8 KB limit (octet_length(metadata::text)) is enforced in SQL by pgconductor._private_check_metadata(). invoke() and emit_event() call it. A ctx.invoke override is checked with the same function before the handler hangs up, so an oversized child fails only that attempt and never fails the batched settlement statement. Inherited copies are not re-checked. Event payload validation now shares a small validateSchema helper.

Verification

  • tests/integration/execution-metadata.test.ts covers metadata on invoked (single and batch), event-triggered (conductor.emit and ctx.emit), child, resumed, retried, dead-letter and ctx.schedule cron executions (including the next occurrence). It also checks that an override applies only to its child, that schema validation works, that >8 KB is rejected on invoke and emit, and that an oversized override fails the parent while a bystander in the same flush completes.
  • tests/unit/execution-metadata-types.test.ts covers the typed ctx.metadata, the readonly property, the override callback and the typing of the options objects.
  • just lint, bun run typecheck and the full bun test pass (344 tests).

Docs: new guide at crafting-tasks/execution-metadata.md, plus updated Conductor and Task Context API pages and AGENTS.md.

Follow-ups

  • ctx.emit and ctx.schedule only inherit; they take no metadata override yet.
  • Batch task events do not expose metadata on event.execution.
  • feat/middleware should add metadata to the execution object that middleware receives.
  • Pre-existing typing quirk, not changed here: conductor.invoke(task, [...items]) also matches the single-payload overload when the task's payload type is {}, so per-item options are not type-checked for payload-less tasks.

@psteinroe psteinroe added the ready label Oct 2, 2026
@psteinroe
psteinroe force-pushed the feat/execution-metadata branch 2 times, most recently from dc6c07f to 529e46c Compare October 2, 2026 23:17
@psteinroe
psteinroe force-pushed the feat/execution-metadata branch from 529e46c to 0f1429f Compare October 2, 2026 23:24
@psteinroe
psteinroe merged commit cd3ee3f into main Oct 2, 2026
9 checks passed
@psteinroe
psteinroe deleted the feat/execution-metadata branch October 2, 2026 23:29
psteinroe added a commit that referenced this pull request Oct 13, 2026
Webhook providers redeliver, so ingress may emit the same delivery
twice. `conductor.emit` and `ctx.emit` now take a `dedupe_key` option,
named like the one on `conductor.invoke`. Repeating a key for the same
event name stores nothing and returns the original event ID. `emit()`
always returns the event ID; the `dedupe_key` is the caller's own key
and is never returned.

```ts
const eventId = await conductor.emit("slack.mentioned", payload, { dedupe_key: `slack:${deliveryId}` });
await ctx.emit("x", payload, { dedupe_key });
```

**SQL**

`pgconductor.emit_event` gains a trailing `p_dedupe_key text default
null`, so existing plain-SQL callers are unaffected. The two mechanisms
stay separate:
- the client-generated `p_id` uuid from #63 makes a retry after a lost
response idempotent;
- `p_dedupe_key` is the caller's key. The dispatch execution stores
`'<event name>:<dedupe_key>'` as its `dedupe_key`, and the existing
unique `(task_key, dedupe_key, queue)` constraint enforces it.

The insert now uses an untargeted `on conflict do nothing`, which covers
both. It then returns the stored row's ID. This is
insert-or-return-existing, not invoke's dedupe semantics: the original
payload is kept. Concurrent emits with the same key serialize on the
unique index. The loser waits for the winner to commit, then returns the
winner's ID.

**Retention window**

The internal `pgconductor.event-dispatch` task changes from
`removeOnComplete: true` to `{ days: 1 }`, in both the task definition
and the `_private_tasks` seed row. Completed dispatch rows now stay for
a day, and the daily maintenance run removes them. Every worker,
including the internal `pgconductor.internal` one, registers a
maintenance cron, so keys are remembered for 1–2 days. The tests check
that maintenance is scheduled on the internal queue.

The scope (per event name) and the window are documented in
`triggers.md` ("Deduplicating Emits"), `api/conductor.md`, a new
`ctx.emit` section in `api/task-context.md`, and next to invoke in
`deduplication.md`.

**Verification**

New integration tests in `event-pipeline.test.ts`:
- The same key twice returns the first event ID and stores one dispatch.
A plain-SQL `p_dedupe_key` caller gets the same ID. The same key under
another event name emits a second event.
- Concurrent emits from two transactions with the same key: the second
blocks, then returns the first ID, and there is one dispatch.
- A repeated `conductor.emit` key after dispatch triggers the listener
once and resumes the waiter once. The waiter's second wait times out
instead of seeing the repeat.
- `ctx.emit` with a key across a failed and retried attempt stores one
event.
- The key is still remembered 23h after dispatch. After maintenance
cleanup at 25h, the same key emits a new event.

Three existing tests asserted that the dispatch row is deleted on
completion. They now assert it is completed. `just lint`, `bun run
typecheck` and the full `bun test` pass.

**Follow-ups**

- #72 (execution metadata) also adds an options object to `emit`.
Whichever lands second merges `{ dedupe_key, metadata }` into one
options type.
- A permanently failed dispatch keeps its key until retention removes
it, and the dispatch task has no `removeOnFail`. Out of scope here.
- Very long keys (above roughly 2.5 KB with the event name) would exceed
the btree entry size of the `dedupe_key` index. Not bounded here.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant