Skip to content

feat(tasks): add ctx.start to start an independent execution - #78

Open
psteinroe wants to merge 1 commit into
mainfrom
feat/ctx-start
Open

psteinroe wants to merge 1 commit into
mainfrom
feat/ctx-start

Conversation

@psteinroe

Copy link
Copy Markdown
Owner

What

Adds ctx.start(key, task, payload, options). It starts an independent execution from inside a task and returns its ID right away.

const executionId = await ctx.start("handoff", { name: "support-flow" }, payload, {
  dedupe_key?: string,      // default: `${ctx execution id}:${key}`
  metadata?: Metadata | ((m) => Metadata), // inherited by default, same override shape as ctx.invoke
  run_at?: Date,
  priority?: number,
  group?: string,
});

Why

Until now a task could start independent work only through ctx.emit or conductor.invoke inside a ctx.step. The second option loses metadata inheritance and needs a hand-made dedupe key. ctx.invoke always suspends until the child settles. Fabrial's ctx.handoff and workflow.asTool() need to start work now and maybe await it somewhere else.

How

  • One statement (buildStartExecution) locks the starting execution (locked_by must still match), inserts the new execution and saves the step { id }. Retries and resumes read the memoized step. If the worker no longer holds the starter, ctx.start hangs up with suspended, the same way waitForEvent does.
  • Dedupe: the default dedupe_key is derived from the starter's execution ID and the step key. If a crash happens after the insert but before the step is saved, the retry still lands on the same execution. A retried statement after a lost response does too.
  • Holding a key: if an execution already holds the key, it is returned as is. The upsert is on conflict (task_key, dedupe_key, queue) do update set dedupe_key = excluded.dedupe_key, so it never supersedes or updates that execution. That differs from pgconductor.invoke(), which supersedes a running holder and would create a second execution when a retry follows a crash. A custom dedupe_key therefore works as "start or reconnect".
  • No lineage: the started execution has no parent_execution_id and no waiting edge. Settlement's orphaned_children CTE fails a completed execution that has a parent but no waiter ("Parent timed out before child completed"), so setting that column would break started executions. Cancelling the starter doesn't cancel the started execution, because cancellation follows waiting_on_execution_id only.
  • ctx.invoke and ctx.start now share a small childMetadata() helper for resolving inherited or overridden metadata.

Await or read the started execution by ID with conductor.waitForResult or conductor.getExecution (#70).

Verification

New tests/integration/start-support.test.ts:

  • the starter completes while the started execution is still blocked; the started execution then finishes and is awaitable, keeps priority and group, and has no parent
  • a resume plus a retry, where the attempt crashed after deleting the start step (simulating a crash between the insert and the step save), return the same ID, and only one execution exists. Swapping the default key for a random one makes this test fail.
  • a lost response for the start statement is retried without a second execution
  • cancelling the starter (while it sleeps) leaves the started execution running to completion
  • metadata is inherited, and an override applies only to that execution
  • a dedupe_key held by a running conductor.invoke execution returns that ID and neither supersedes it nor runs it twice

Type test: ctx.start returns string and checks metadata types.

just lint, just format, bun run typecheck and the full bun test (405 pass) all pass.

Follow-ups

  • ctx.waitForAny with an { execution: id } branch: feat/wait-for-any has not merged yet. Once it lands, add that branch on top, reusing the settlement path that wakes ctx.invoke parents.
  • docs/content/api/task-context.md documents ctx.executionId, but TaskContext doesn't expose it. Out of scope here.

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant