Reconcile df.start() with the durable engine via committed intent

[tjgreen42]

Problem

df.start() writes its df rows in the caller's transaction but starts the workflow engine out-of-band on a separate connection. Those two writes can diverge:

• Ghost — a rolled-back df.start() leaves the engine running a workflow with no df record behind it.
• Stuck — a committed df instance whose engine start was lost (worker down, dropped connection) never runs.

Root cause: the df write and the engine start happen in two different transactions with nothing reconciling them.

Approach

Treat the committed df row as the source of truth (intent) and let the background worker converge the engine to match, through duroxide's Rust API (Client::start_orchestration / get_orchestration_status).

• df.start() persists a start_input payload on df.instances and issues a transactional NOTIFY; it no longer starts the engine itself.
• The worker LISTENs for instant starts and periodically sweeps for committed-but-unstarted instances the engine does not yet know about, starting each exactly once via an atomic claim.

The df write and the engine start stay in separate transactions and are reconciled asynchronously: a rolled-back df.start() starts nothing (no ghost), and a committed instance is always eventually started (no stuck).

Exactly-once, crash recovery, and trust boundary

• Exactly-once: the instant and sweep paths both claim the row with an atomic UPDATE ... RETURNING; only the winner starts the engine.
• Crash recovery: the claim commits before the engine start, so a worker crash in between would strand the row in running. The sweep therefore also recovers stale claims — a running row the engine reports NotFound for whose claim is older than the grace is re-claimed and re-started.
• Security: start_input is caller-writable and the orchestration selects the graph/role from the input's instance id. The worker forces the engine input's instance id to the claimed row id, and execute() uses the durable runtime's ctx.instance_id() instead of the payload's — so a crafted start_input cannot redirect the superuser worker to run another user's instance.
• Resilience: the start listener is re-established if it drops (so reconcile_interval=0 still keeps the instant path); the sweep is batch-bounded and yields to shutdown; a startup catch-up sweep recovers starts whose NOTIFY was missed during downtime.

Changes

• src/dsl.rs — df.start() intent path (persist start_input, transactional pg_notify); legacy inline start retained for pre-0.2.4 schemas (version-gated).
• src/worker.rs — duroxide Client, LISTEN + reconcile sweep with stale-claim recovery, exactly-once claim, listener rebuild, bounded batches.
• src/orchestrations/execute_function_graph.rs — trust ctx.instance_id() over the caller-supplied payload.
• Schema (0.2.4) — df.instances.start_input JSONB (fresh install + upgrade script), grant/revoke + backfill for existing df roles. GUCs pg_durable.reconcile_interval (15s) / reconcile_grace (300s).
• Docs/tests — docs/spec-reconcile-start.md, USER_GUIDE section, upgrade-testing notes; tests/e2e/sql/25_reconcile_start.sql (ghost / stuck / stale-claim, RED on main), tests/e2e/sql/26_reconcile_security.sql (cross-tenant redirect blocked), upgrade grant-backfill coverage.

Backward compatibility

New .so runs against un-upgraded 0.2.2/0.2.3 schemas: df.start() falls back to the inline start until the column exists, and the worker re-probes so an in-place ALTER EXTENSION UPDATE enables reconciliation without a restart.

Scope

Start-convergence only. The engine-orphan cleanup direction (engine has it, df does not) is out of scope; df.cancel()/df.signal() stay on the existing path, with a targeted re-cancel if an instance is cancelled between the claim and the start.

Testing

build · clippy (nightly-equivalent) · fmt · unit 194 · E2E 38 · upgrade 37 — all green locally and in CI.

[tjgreen42]

Superseding this approach. After review discussion we're pivoting away from deferring the start (persist-intent + worker-driven start) to a simpler model: keep the eager start in df.start() but propagate the enqueue error (fail-fast), and add a periodic background reaper that drives duroxide's retention/deletion API. The rolled-back-start 'ghost' is provably inert (the orchestration fails at graph-load before any node runs), so it only needs cleanup, not prevention — and that cleanup doubles as the duroxide-side retention pg_durable currently lacks. Replacement PRs to follow.