coordinator: async partial spawn failure (one daemon Errs after another already spawned) never rolls back the succeeded daemon — orphaned nodes + `running_dataflows`/store divergence (`dora list` stuck "Running")

[phil-opp]

Note: This issue was created by a scheduled, automated Claude code-review check (unattended run). The finding below was verified against the source at commit f3c032f (branch main), but a human should confirm before acting on it.

Summary

On a multi-daemon dataflow, when the daemons report their spawn results asynchronously and one daemon reports success while another reports failure, handle_spawn_result_err marks the dataflow terminally Failed and caches the spawn error — but it does not roll back the daemon(s) that already spawned successfully, and it does not remove the dataflow from running_dataflows. The result is:

1. Orphaned node processes on the succeeded daemon(s) that keep running indefinitely, unmanaged by the coordinator.
2. A durable in-memory vs. persisted-store divergence: the store record says Failed { terminal: true }, but the dataflow stays in running_dataflows, so dora list reports it as Running — forever, until a manual dora stop.

This is the asynchronous sibling of the partial-spawn-failure cases that the codebase already handles correctly (see "Why this is a gap, not by design" below); the async path is simply missing the equivalent rollback + teardown.

Affected code

• binaries/coordinator/src/lib.rs — handle_spawn_result_err (lines ~3046–3077)
• dispatched from the Event::DataflowSpawnResult handler (lines ~2461–2493)

fn handle_spawn_result_err(
    dataflow: &mut RunningDataflow,
    dataflow_id: DataflowId,
    daemon_id: &DaemonId,
    err: eyre::Report,
    store: &dyn CoordinatorStore,
) {
    if !dataflow.spawn_result.is_pending() { /* ignore late ... */ return; }

    // persists Failed { terminal: true } ...
    dataflow.spawn_result.set_result(Err(err));
    // <-- no rollback of already-succeeded daemons,
    //     no removal/teardown of the running_dataflows entry
}

Concrete failure scenario

Dataflow spawned across daemons A and B. Each daemon accepts the spawn request synchronously (run::spawn_dataflow → TriggerSpawnResult(Ok) only means "queued"); the actual node spawn happens asynchronously and is reported later via DaemonEvent::SpawnResult → Event::DataflowSpawnResult.

Ordering 1 — A succeeds first, then B fails:

1. A's Ok arrives → pending_spawn_results drops A; spawn_result stays Pending (B still pending). A's node processes are now running.
2. B's Err arrives (e.g. missing binary/env on machine B) → handle_spawn_result_err runs while spawn_result.is_pending() is still true → persists Failed { terminal: true }, sets spawn_result = Err. A is never told to stop.

Ordering 2 — B fails first, then A succeeds:

1. B's Err arrives → terminal Failed, spawn_result = Err.
2. A's Ok arrives → handle_spawn_result_ok hits the !is_pending() guard and ignores the late success — but A's spawn genuinely succeeded and A's nodes are running. A is never rolled back.

Either way: the succeeded daemon's nodes are orphaned and the dataflow stays in running_dataflows.

Why no recovery path catches it

• Spawn-timeout watchdog (check_spawn_timeouts, line ~3551) filters on df.spawn_result.is_pending(). After handle_spawn_result_err, the result is Cached(Err) — not pending — so the watchdog (which does do fire_and_forget_rollback + full teardown) never touches this dataflow.
• DaemonStatusReport reconcile / orphan-stop (lines ~2511–2654) can stop orphans, but:
  • the archived_dataflows orphan-stop branch (line ~2513) is skipped because this path never archives the dataflow; and
  • the terminal-store orphan-stop branch (status_report_should_stop_orphan, line ~2635) only fires when a daemon actually sends a DaemonStatusReport. The daemon emits DaemonEvent::StatusReport exactly once, immediately before its main event loop (binaries/daemon/src/lib.rs:1787, before while let Some(event) = events.next().await) — i.e. at startup/reconnect, not periodically. The healthy daemon A never disconnects here, so it never re-reports, and this cleanup never runs.
• ControlRequest::List (lines ~1100–1110) unconditionally classifies every entry still in running_dataflows as DataflowStatus::Running, so the divergence is user-visible.

Net effect: leaked/orphaned node processes on the healthy daemon plus a dataflow permanently shown as Running despite a terminal Failed store verdict — cleared only by a manual dora stop (or if daemon A happens to disconnect and reconnect, which re-emits a StatusReport and triggers the terminal orphan-stop).

Why this is a gap, not by design

The two sibling partial-spawn-failure paths both roll back the succeeded daemons and reconcile in-memory state; only the async-report path is missing it:

• run::spawn_dataflow calls rollback_spawned_daemons on a synchronous partial failure (binaries/coordinator/src/run/mod.rs ~88–118).
• check_spawn_timeouts calls fire_and_forget_rollback and performs full teardown (archive, drain stop_reply_senders, synthesize dataflow_results, remove from running_dataflows) on a timed-out spawn (lib.rs ~3568–3742).

handle_spawn_result_err should do the equivalent: roll back the daemons that already succeeded (the daemons.difference(&pending_spawn_results) set) and perform the in-memory teardown so state doesn't diverge — rather than only recording the failure and caching the error.

Suggested direction

In handle_spawn_result_err, when transitioning the dataflow to terminal Failed, additionally:

1. Compute the succeeded-daemon set (dataflow.daemons.difference(&dataflow.pending_spawn_results)) and issue fire_and_forget_rollback (StopDataflow { force: true }) to them, and
2. Perform the same teardown the timeout watchdog does (archive + remove from running_dataflows + synthesize results), so dora list and the store agree.

Because handle_spawn_result_err currently takes &mut RunningDataflow (not the maps / daemon_connections), realizing this likely means moving the rollback/teardown up into the Event::DataflowSpawnResult handler (which has access to running_dataflows, daemon_connections, archived_dataflows, dataflow_results, and the clock), mirroring how check_spawn_timeouts is structured. A regression test analogous to the existing spawn-timeout tests (assert the succeeded daemon receives a StopDataflow and the dataflow leaves running_dataflows) would pin the behavior.