Repository navigation
feat(executions): typed, propagated execution metadata - #72
Merged
Merged
Conversation
psteinroe
force-pushed
the
feat/execution-metadata
branch
2 times, most recently
from
October 2, 2026 23:17
dc6c07f to
529e46c
Compare
psteinroe
force-pushed
the
feat/execution-metadata
branch
from
October 2, 2026 23:24
529e46c to
0f1429f
Compare
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.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
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.invokereaches every task in the workflow without being added to each payload.Propagation
A new
metadata jsonbcolumn on_private_executionsfollows the same SQL paths astrace_context:ctx.invokechildTaskContextpasses its metadata explicitly toctx.invoke,ctx.emitandctx.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 onconductor.invoke,conductor.emitandctx.invokeoverrides, with the same error format as event payloads. The 8 KB limit (octet_length(metadata::text)) is enforced in SQL bypgconductor._private_check_metadata().invoke()andemit_event()call it. Actx.invokeoverride 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 smallvalidateSchemahelper.Verification
tests/integration/execution-metadata.test.tscovers metadata on invoked (single and batch), event-triggered (conductor.emitandctx.emit), child, resumed, retried, dead-letter andctx.schedulecron 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.tscovers the typedctx.metadata, the readonly property, the override callback and the typing of the options objects.just lint,bun run typecheckand the fullbun testpass (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.emitandctx.scheduleonly inherit; they take no metadata override yet.event.execution.metadatato the execution object that middleware receives.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.