From be27a62fb118f6d7d3c094b1c6c5ebacd8ae5fd7 Mon Sep 17 00:00:00 2001 From: atlas Date: Tue, 28 Jul 2026 00:53:58 +0200 Subject: [PATCH] refactor(#2825): complete_node takes only the node id MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `NodeId` has been globally unique across DAGs since #2801, so the dag id carried no information the node id didn't. The parameter was already underscore-prefixed as unused, but still populated by `claim_ready`, carried through the scheduler's mpsc on every `Claim`, and passed at the call site — three layers of plumbing feeding a dead argument. Removing it surfaced four more dead things it had been keeping alive: `settle_approval_tail`, `settle_rebuild_tail` and `drain_meta_syncs` each took a dag id they only forwarded to `complete_node`, and one `submit` binding was never read. Those are deleted rather than underscore-prefixed — prefixing is what let the original argument survive this long. `Claim.dag_id` stays: it has live consumers in the tracing spans, `append_subgraph`'s container guard, `Ctx` for the build-log link, `first_error`, and the approval-deploy context. --- hive-c0re/src/job_queue/mod.rs | 2 +- hive-c0re/src/job_queue/scheduler.rs | 8 +- hive-c0re/src/job_queue/tests.rs | 168 +++++++++++++-------------- 3 files changed, 83 insertions(+), 95 deletions(-) diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 5e7e2616..1bce674f 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -324,7 +324,7 @@ impl JobQueue { /// Nothing is returned: a DAG's terminal side effects are its own tail nodes /// ([`NodeKind::ResolveApproval`] / [`NodeKind::EmitRebuilt`]), which the /// scheduler claims and runs like any other node. - pub fn complete_node(&self, _dag_id: u64, node_id: NodeId, result: Result<(), String>) { + pub fn complete_node(&self, node_id: NodeId, result: Result<(), String>) { let mut inner = self.lock(); // The failure reason + `finished_at` are stamped onto the graph `Node` // by the scheduler (the reason rides `Outcome::Failed`); no host-side diff --git a/hive-c0re/src/job_queue/scheduler.rs b/hive-c0re/src/job_queue/scheduler.rs index 6191ab6b..ea477e35 100644 --- a/hive-c0re/src/job_queue/scheduler.rs +++ b/hive-c0re/src/job_queue/scheduler.rs @@ -105,9 +105,7 @@ fn handle_completion(coord: &Arc, done: NodeDone) { .job_queue .append_subgraph(claim.dag_id, subgraph, claim.node_id); } - coord - .job_queue - .complete_node(claim.dag_id, claim.node_id, Ok(())); + coord.job_queue.complete_node(claim.node_id, Ok(())); } Err(e) => { let msg = format!("{e:#}"); @@ -119,9 +117,7 @@ fn handle_completion(coord: &Arc, done: NodeDone) { error = %msg, "job_queue: node failed" ); - coord - .job_queue - .complete_node(claim.dag_id, claim.node_id, Err(msg)); + coord.job_queue.complete_node(claim.node_id, Err(msg)); } } // The next loop iteration re-reconciles the transient pills against the diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index d94cdf58..5832a583 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -56,7 +56,7 @@ fn claim_one(q: &JobQueue) -> Claim { /// A template emits one tail per outcome and the graph runs exactly one, so the /// assertion is on *which node was claimed* — that alone says what the approval /// row is about to be resolved as. Nothing computes it. -fn settle_approval_tail(q: &JobQueue, dag_id: u64, approval_id: i64, expect: TerminalState) { +fn settle_approval_tail(q: &JobQueue, approval_id: i64, expect: TerminalState) { let tail = claim_one(q); assert!( matches!( @@ -67,19 +67,19 @@ fn settle_approval_tail(q: &JobQueue, dag_id: u64, approval_id: i64, expect: Ter "expected the {expect:?} ResolveApproval tail for #{approval_id}, got {:?}", tail.kind ); - q.complete_node(dag_id, tail.node_id, Ok(())); + q.complete_node(tail.node_id, Ok(())); } /// The `EmitRebuilt` counterpart of [`settle_approval_tail`] — claim the tail the /// graph let run and assert it's the `ok` one expected. -fn settle_rebuild_tail(q: &JobQueue, dag_id: u64, agent: &str, expect_ok: bool) { +fn settle_rebuild_tail(q: &JobQueue, agent: &str, expect_ok: bool) { let tail = claim_one(q); assert!( matches!(&tail.kind, NodeKind::EmitRebuilt { agent: a, ok } if a == agent && *ok == expect_ok), "expected the ok={expect_ok} EmitRebuilt tail for {agent}, got {:?}", tail.kind ); - q.complete_node(dag_id, tail.node_id, Ok(())); + q.complete_node(tail.node_id, Ok(())); } fn state_of(q: &JobQueue, dag_id: u64) -> State { @@ -228,9 +228,9 @@ fn rebuild_chain_claims_in_dep_order() { q.claim_ready().is_empty(), "chain must serialize: nothing ready while {expected} runs" ); - q.complete_node(id, c.node_id, Ok(())); + q.complete_node(c.node_id, Ok(())); } - settle_rebuild_tail(&q, id, "agent-a", true); + settle_rebuild_tail(&q, "agent-a", true); assert_eq!(state_of(&q, id), State::Done); } @@ -276,7 +276,7 @@ fn graceful_rebuild_chain_drains_before_stopping() { q.claim_ready().is_empty(), "chain must serialize: nothing ready while {expected} runs" ); - q.complete_node(id, c.node_id, Ok(())); + q.complete_node(c.node_id, Ok(())); } assert_eq!(state_of(&q, id), State::Done); } @@ -322,9 +322,9 @@ fn settled_dag_leaves_the_snapshot_despite_its_skipped_branch() { let id = submit(&q, rebuild("agent-a", "r")); for _ in 0..6 { let c = claim_one(&q); - q.complete_node(id, c.node_id, Ok(())); + q.complete_node(c.node_id, Ok(())); } - settle_rebuild_tail(&q, id, "agent-a", true); + settle_rebuild_tail(&q, "agent-a", true); assert!( q.snapshot().iter().all(|d| d.id != id), "a fully settled DAG drops out of the snapshot" @@ -343,7 +343,7 @@ fn build_slot_serializes_nix_heavy_nodes() { let head_a = claim_one(&q); assert_eq!(head_a.dag_id, a); assert_eq!(head_a.kind.as_str(), "meta_sync"); - q.complete_node(a, head_a.node_id, Ok(())); + q.complete_node(head_a.node_id, Ok(())); // a's Prebuild takes the only slot; b's MetaSync is free to run beside it // (different resources), but b's Prebuild is not. let claims = q.claim_ready(); @@ -351,7 +351,7 @@ fn build_slot_serializes_nix_heavy_nodes() { kinds.sort_unstable(); assert_eq!(kinds, vec![(a, "prebuild"), (b, "meta_sync")]); for c in &claims { - q.complete_node(c.dag_id, c.node_id, Ok(())); + q.complete_node(c.node_id, Ok(())); } // Uniform hold: agent-a keeps the build slot across its whole build chain // (Swap re-enters it), so a's StopForUpdate (lease, slot-free) runs but b's @@ -382,7 +382,7 @@ fn two_build_slots_run_two_prebuilds() { for _ in 0..3 { for c in q.claim_ready() { if c.kind.as_str() == "meta_sync" { - q.complete_node(c.dag_id, c.node_id, Ok(())); + q.complete_node(c.node_id, Ok(())); } else { prebuilds.push(c); } @@ -400,7 +400,7 @@ fn fifo_fairness_for_the_slot() { let c = submit(&q, rebuild("agent-c", "r")); let first = claim_one(&q); assert_eq!(first.dag_id, a, "submit order wins the slot"); - q.complete_node(a, first.node_id, Ok(())); + q.complete_node(first.node_id, Ok(())); // Uniform hold: the slot stays with agent-a until its Swap (the last // slot-needer) completes. Drive a's chain; the moment its slot frees, // submit order (b before c) wins it. @@ -413,7 +413,7 @@ fn fifo_fairness_for_the_slot() { } for cl in claims { if cl.dag_id == a { - q.complete_node(a, cl.node_id, Ok(())); + q.complete_node(cl.node_id, Ok(())); } } } @@ -439,20 +439,20 @@ fn lease_serializes_two_lifecycle_dags_for_same_agent() { let first = claim_one(&q); assert_eq!(first.dag_id, restart); assert_eq!(first.kind.as_str(), "stop_for_update"); - q.complete_node(restart, first.node_id, Ok(())); + q.complete_node(first.node_id, Ok(())); // Same DAG keeps the lease through the tail Reconcile (re-entered from the // dep graph — no fresh acquire), since stop's Reconcile can't re-enter it. let second = claim_one(&q); assert_eq!(second.dag_id, restart); assert_eq!(second.kind.as_str(), "reconcile"); - q.complete_node(restart, second.node_id, Ok(())); + q.complete_node(second.node_id, Ok(())); // Restart's work is terminal → its lease releases, so stop's now-unblocked // Reconcile becomes ready (a power op has no tail node, so nothing of // restart's remains claimable). let third = claim_one(&q); assert_eq!(third.dag_id, stop); assert_eq!(third.kind.as_str(), "reconcile"); - q.complete_node(stop, third.node_id, Ok(())); + q.complete_node(third.node_id, Ok(())); assert_eq!(state_of(&q, restart), State::Done); assert_eq!(state_of(&q, stop), State::Done); } @@ -461,7 +461,7 @@ fn lease_serializes_two_lifecycle_dags_for_same_agent() { fn lease_exempt_prebuild_overlaps_other_dag_on_same_agent() { let q = JobQueue::new(2); submit(&q, rebuild("agent-a", "rebuild")); - let stop = submit( + submit( &q, templates::reconcile_only("agent-a", Source::Manual, "stop".to_owned(), None), ); @@ -476,7 +476,7 @@ fn lease_exempt_prebuild_overlaps_other_dag_on_same_agent() { .find(|c| c.kind.as_str() == "meta_sync") .expect("meta_sync claim") .clone(); - q.complete_node(meta_sync.dag_id, meta_sync.node_id, Ok(())); + q.complete_node(meta_sync.node_id, Ok(())); // Prebuild is lease-exempt: the stop's Reconcile keeps the lease // and runs concurrently with the rebuild's out-of-band nix build. let claims = q.claim_ready(); @@ -489,7 +489,7 @@ fn lease_exempt_prebuild_overlaps_other_dag_on_same_agent() { .find(|c| c.kind.as_str() == "prebuild") .expect("prebuild claim") .clone(); - q.complete_node(prebuild.dag_id, prebuild.node_id, Ok(())); + q.complete_node(prebuild.node_id, Ok(())); assert!( q.claim_ready().is_empty(), "StopForUpdate blocked while stop DAG holds the lease" @@ -499,7 +499,7 @@ fn lease_exempt_prebuild_overlaps_other_dag_on_same_agent() { .find(|c| c.kind.as_str() == "reconcile") .expect("reconcile claim") .clone(); - q.complete_node(stop, reconcile.node_id, Ok(())); + q.complete_node(reconcile.node_id, Ok(())); // stop's Reconcile done → its lease frees, so rebuild's StopForUpdate // unblocks. (stop's DAG rolls up terminal; a power op has no tail node, so // nothing of stop's is left in the claim set.) @@ -565,7 +565,7 @@ fn multi_agent_lease_frees_per_subgraph_not_whole_dag() { let mut progressed = false; for c in q.claim_ready() { if c.agent == "agent-a" { - q.complete_node(id, c.node_id, Ok(())); + q.complete_node(c.node_id, Ok(())); progressed = true; } else { b_in_flight = true; // leave agent-b's node running @@ -650,7 +650,7 @@ fn multi_agent_start_one_dag_folds_per_agent_stale_rebuild() { // Complete both heads; the fresh agent then reconciles directly while // the stale agent's subgraph is the rebuild chain (meta_sync first). for c in &heads { - q.complete_node(id, c.node_id, Ok(())); + q.complete_node(c.node_id, Ok(())); } let next = q.claim_ready(); let mut kinds: Vec<(&str, &str)> = next @@ -757,13 +757,13 @@ fn append_subgraph_roots_on_emitter_and_rebases_local_deps() { // Must append BEFORE completing the emitter (the documented contract). q.append_subgraph(id, &subgraph("a"), emitter.node_id); q.append_subgraph(id, &subgraph("b"), emitter.node_id); - q.complete_node(id, emitter.node_id, Ok(())); + q.complete_node(emitter.node_id, Ok(())); // Still ONE DAG; both subgraph roots become ready once the emitter is // Done (rooted on it), each on its own agent lease. Their `MetaSync` heads // take turns on the cap-1 global meta window, so drain those first — what // must be concurrent is the builds. assert_eq!(q.snapshot().len(), 1); - let mut kinds = drain_meta_syncs(&q, id); + let mut kinds = drain_meta_syncs(&q); kinds.sort_unstable(); assert_eq!( kinds, @@ -778,12 +778,12 @@ fn append_subgraph_roots_on_emitter_and_rebases_local_deps() { /// Complete every `MetaSync` head the queue offers (they take turns on the /// cap-1 global meta window) and return whatever else got claimed alongside /// them, as `(agent, kind)` pairs left in flight. -fn drain_meta_syncs(q: &JobQueue, dag: u64) -> Vec<(String, String)> { +fn drain_meta_syncs(q: &JobQueue) -> Vec<(String, String)> { let mut rest = Vec::new(); for _ in 0..3 { for c in q.claim_ready() { if c.kind.as_str() == "meta_sync" { - q.complete_node(dag, c.node_id, Ok(())); + q.complete_node(c.node_id, Ok(())); } else { rest.push((c.agent.clone(), c.kind.as_str().to_owned())); } @@ -832,13 +832,13 @@ fn meta_update_carries_rebuilding_transient_and_grows_cascade_in_dag() { meta_lock.node_id, ); } - q.complete_node(id, meta_lock.node_id, Ok(())); + q.complete_node(meta_lock.node_id, Ok(())); // Still ONE DAG — no child DAGs — and both cascade rebuild subgraphs root // on the MetaLock, each on its own agent lease. The per-agent `MetaSync` // heads serialize on the global meta window (they commit to the meta repo); // the builds behind them do not. assert_eq!(q.snapshot().len(), 1); - let mut kinds = drain_meta_syncs(&q, id); + let mut kinds = drain_meta_syncs(&q); kinds.sort_unstable(); assert_eq!( kinds, @@ -858,15 +858,15 @@ fn failed_node_cancels_downstream_but_afterany_reconcile_runs() { let id = submit(&q, rebuild("agent-a", "r")); let meta_sync = claim_one(&q); assert_eq!(meta_sync.kind.as_str(), "meta_sync"); - q.complete_node(id, meta_sync.node_id, Ok(())); + q.complete_node(meta_sync.node_id, Ok(())); let prebuild = claim_one(&q); assert_eq!(prebuild.kind.as_str(), "prebuild"); - q.complete_node(id, prebuild.node_id, Err("nix build exploded".to_owned())); + q.complete_node(prebuild.node_id, Err("nix build exploded".to_owned())); // StopForUpdate + Swap are cancelled (AfterOk on a failed chain); // the AfterAny Reconcile still runs once Swap is terminal. let reconcile = claim_one(&q); assert_eq!(reconcile.kind.as_str(), "reconcile"); - q.complete_node(id, reconcile.node_id, Ok(())); + q.complete_node(reconcile.node_id, Ok(())); let snap = q.snapshot(); let dag = snap.iter().find(|d| d.id == id).expect("dag"); assert_eq!(dag.rollup_state(), State::Failed, "roll-up failed"); @@ -915,16 +915,16 @@ fn swap_failure_still_runs_reconcile() { // meta_sync + prebuild + stop_for_update for _ in 0..3 { let c = claim_one(&q); - q.complete_node(id, c.node_id, Ok(())); + q.complete_node(c.node_id, Ok(())); } let swap = claim_one(&q); assert_eq!(swap.kind.as_str(), "swap"); - q.complete_node(id, swap.node_id, Err("update failed".to_owned())); + q.complete_node(swap.node_id, Err("update failed".to_owned())); // PostSwap (AfterOk on the failed Swap) is cancel-cascaded; Reconcile is // next-claimable via its AfterAny(PostSwap) edge. let reconcile = claim_one(&q); assert_eq!(reconcile.kind.as_str(), "reconcile"); - q.complete_node(id, reconcile.node_id, Ok(())); + q.complete_node(reconcile.node_id, Ok(())); let all_dags = q.snapshot(); let dag = all_dags.iter().find(|d| d.id == id).expect("dag"); assert_eq!( @@ -958,11 +958,11 @@ fn swap_ok_runs_post_swap_before_reconcile() { // meta_sync + prebuild + stop_for_update for _ in 0..3 { let c = claim_one(&q); - q.complete_node(id, c.node_id, Ok(())); + q.complete_node(c.node_id, Ok(())); } let swap = claim_one(&q); assert_eq!(swap.kind.as_str(), "swap"); - q.complete_node(id, swap.node_id, Ok(())); + q.complete_node(swap.node_id, Ok(())); // PostSwap runs next, and nothing else is claimable while it does — the // tail serializes ahead of Reconcile. let post_swap = claim_one(&q); @@ -971,11 +971,11 @@ fn swap_ok_runs_post_swap_before_reconcile() { q.claim_ready().is_empty(), "Reconcile must wait for PostSwap, not race it" ); - q.complete_node(id, post_swap.node_id, Ok(())); + q.complete_node(post_swap.node_id, Ok(())); let reconcile = claim_one(&q); assert_eq!(reconcile.kind.as_str(), "reconcile"); - q.complete_node(id, reconcile.node_id, Ok(())); - settle_rebuild_tail(&q, id, "agent-a", true); + q.complete_node(reconcile.node_id, Ok(())); + settle_rebuild_tail(&q, "agent-a", true); assert_eq!(state_of(&q, id), State::Done); } @@ -987,7 +987,7 @@ fn failed_reconcile_marks_dag_failed() { templates::reconcile_only("agent-a", Source::Manual, "start".to_owned(), None), ); let c = claim_one(&q); - q.complete_node(id, c.node_id, Err("start failed".to_owned())); + q.complete_node(c.node_id, Err("start failed".to_owned())); assert_eq!(state_of(&q, id), State::Failed); } @@ -1091,12 +1091,12 @@ fn dag_settles_terminal_and_releases_lease_after_work() { // restart = StopForUpdate → Reconcile. let stop = claim_one(&q); assert_eq!(stop.kind.as_str(), "stop_for_update"); - q.complete_node(id, stop.node_id, Ok(())); + q.complete_node(stop.node_id, Ok(())); let rec = claim_one(&q); assert_eq!(rec.kind.as_str(), "reconcile"); // Completing the last work node rolls the container up terminal. A power op // has no tail node, so nothing is left to claim. - q.complete_node(id, rec.node_id, Ok(())); + q.complete_node(rec.node_id, Ok(())); assert!(q.claim_ready().is_empty(), "no tail node to claim"); assert_eq!(state_of(&q, id), State::Done); // Lease released when the work chain settled: a new DAG for the agent claims @@ -1128,13 +1128,13 @@ fn cancelled_dag_still_runs_its_approval_tail() { // The `Cancelled` tail is the only node whose edge accepts a dropped // dependency, so it is the only one `cancel` spares — and claiming it *is* // the assertion that the approval gets resolved as cancelled. - settle_approval_tail(&q, id, 7, TerminalState::Cancelled); + settle_approval_tail(&q, 7, TerminalState::Cancelled); assert_eq!(state_of(&q, id), State::Cancelled); // Unrelated later activity doesn't disturb the settled DAG. let other = submit(&q, rebuild("agent-b", "r")); let c = claim_one(&q); assert_eq!(c.dag_id, other); - q.complete_node(other, c.node_id, Err("boom".to_owned())); + q.complete_node(c.node_id, Err("boom".to_owned())); assert_eq!(state_of(&q, id), State::Cancelled); } @@ -1158,28 +1158,24 @@ fn deploy_dag_runs_phases_in_order_and_tails_a_failed_apply() { matches!(root.kind, NodeKind::DeployWindow { .. }), "root claims first: it holds the meta window for the whole subtree" ); - q.complete_node(id, root.node_id, Ok(())); + q.complete_node(root.node_id, Ok(())); let verify = claim_one(&q); assert!(matches!(verify.kind, NodeKind::MergeVerify { .. })); - q.complete_node(id, verify.node_id, Ok(())); + q.complete_node(verify.node_id, Ok(())); let apply = claim_one(&q); assert!(matches!(apply.kind, NodeKind::DeployApply { .. })); - q.complete_node( - id, - apply.node_id, - Err("nixos-container update blew up".into()), - ); + q.complete_node(apply.node_id, Err("nixos-container update blew up".into())); let tail = claim_one(&q); assert!( matches!(tail.kind, NodeKind::DeployTail { .. }), "AfterAny tail runs on a failed apply — that's the whole point of it" ); - q.complete_node(id, tail.node_id, Ok(())); + q.complete_node(tail.node_id, Ok(())); - settle_approval_tail(&q, id, 7, TerminalState::Failed); + settle_approval_tail(&q, 7, TerminalState::Failed); assert_eq!( state_of(&q, id), State::Failed, @@ -1206,9 +1202,9 @@ fn deploy_apply_grows_rebuild_subgraph_and_finalizes_after_it() { let root = claim_one(&q); assert!(matches!(root.kind, NodeKind::DeployWindow { .. })); - q.complete_node(id, root.node_id, Ok(())); + q.complete_node(root.node_id, Ok(())); let verify = claim_one(&q); - q.complete_node(id, verify.node_id, Ok(())); + q.complete_node(verify.node_id, Ok(())); let apply = claim_one(&q); assert!(matches!(apply.kind, NodeKind::DeployApply { .. })); @@ -1222,7 +1218,7 @@ fn deploy_apply_grows_rebuild_subgraph_and_finalizes_after_it() { apply.node_id, ); assert!(!grown.is_empty(), "subgraph grafted onto the apply node"); - q.complete_node(id, apply.node_id, Ok(())); + q.complete_node(apply.node_id, Ok(())); // The grafted chain runs in rebuild order. `claim_one` asserts exactly one // claimable node at each step, which also proves the `AfterAny` tail stays @@ -1238,7 +1234,7 @@ fn deploy_apply_grows_rebuild_subgraph_and_finalizes_after_it() { ] { let c = claim_one(&q); assert_eq!(c.kind.as_str(), expected, "grafted phase order"); - q.complete_node(id, c.node_id, Ok(())); + q.complete_node(c.node_id, Ok(())); } let finalize = claim_one(&q); @@ -1246,13 +1242,13 @@ fn deploy_apply_grows_rebuild_subgraph_and_finalizes_after_it() { matches!(finalize.kind, NodeKind::FinalizeDeploy { .. }), "the deploy tag is planted only after the rebuild came up clean" ); - q.complete_node(id, finalize.node_id, Ok(())); + q.complete_node(finalize.node_id, Ok(())); let tail = claim_one(&q); assert!(matches!(tail.kind, NodeKind::DeployTail { .. })); - q.complete_node(id, tail.node_id, Ok(())); + q.complete_node(tail.node_id, Ok(())); - settle_approval_tail(&q, id, 11, TerminalState::Done); + settle_approval_tail(&q, 11, TerminalState::Done); assert_eq!(state_of(&q, id), State::Done); } @@ -1270,40 +1266,40 @@ fn deploy_dag_skips_finalize_but_still_tails_a_failed_graft() { ); let root = claim_one(&q); - q.complete_node(id, root.node_id, Ok(())); + q.complete_node(root.node_id, Ok(())); let verify = claim_one(&q); - q.complete_node(id, verify.node_id, Ok(())); + q.complete_node(verify.node_id, Ok(())); let apply = claim_one(&q); q.append_subgraph( id, &templates::deploy_rebuild_nodes("agent-a"), apply.node_id, ); - q.complete_node(id, apply.node_id, Ok(())); + q.complete_node(apply.node_id, Ok(())); for expected in ["meta_sync", "prebuild", "stop_for_update"] { let c = claim_one(&q); assert_eq!(c.kind.as_str(), expected); - q.complete_node(id, c.node_id, Ok(())); + q.complete_node(c.node_id, Ok(())); } let swap = claim_one(&q); assert_eq!(swap.kind.as_str(), "swap"); - q.complete_node(id, swap.node_id, Err("profile swap failed".into())); + q.complete_node(swap.node_id, Err("profile swap failed".into())); // `Reconcile` hangs off `Prebuild` with `AfterAny`, so a failed swap still // reaches it — bringing the container back up is exactly what it's for. let reconcile = claim_one(&q); assert_eq!(reconcile.kind.as_str(), "reconcile"); - q.complete_node(id, reconcile.node_id, Ok(())); + q.complete_node(reconcile.node_id, Ok(())); let tail = claim_one(&q); assert!( matches!(tail.kind, NodeKind::DeployTail { .. }), "finalize is cancel-cascaded, so the tail is the next claimable node" ); - q.complete_node(id, tail.node_id, Ok(())); + q.complete_node(tail.node_id, Ok(())); - settle_approval_tail(&q, id, 13, TerminalState::Failed); + settle_approval_tail(&q, 13, TerminalState::Failed); assert_eq!(state_of(&q, id), State::Failed); assert_eq!( q.first_error(id).as_deref(), @@ -1326,22 +1322,18 @@ fn deploy_dag_skips_apply_but_still_runs_tail_when_verify_fails() { ); let root = claim_one(&q); - q.complete_node(id, root.node_id, Ok(())); + q.complete_node(root.node_id, Ok(())); let verify = claim_one(&q); - q.complete_node( - id, - verify.node_id, - Err("PR head drifted since review".into()), - ); + q.complete_node(verify.node_id, Err("PR head drifted since review".into())); let tail = claim_one(&q); assert!( matches!(tail.kind, NodeKind::DeployTail { .. }), "apply is cancel-cascaded, so the tail is the next claimable node" ); - q.complete_node(id, tail.node_id, Ok(())); + q.complete_node(tail.node_id, Ok(())); - settle_approval_tail(&q, id, 9, TerminalState::Failed); + settle_approval_tail(&q, 9, TerminalState::Failed); assert_eq!(state_of(&q, id), State::Failed); } @@ -1353,7 +1345,7 @@ fn set_build_log_id_links_running_node() { let id = submit(&q, rebuild("agent-a", "r")); let c = claim_one(&q); assert!(q.set_build_log_id(id, c.node_id, 42)); - q.complete_node(id, c.node_id, Ok(())); + q.complete_node(c.node_id, Ok(())); assert!( !q.set_build_log_id(id, c.node_id, 99), "node no longer running → refused" @@ -1392,7 +1384,7 @@ fn history_evicts_oldest_terminals_past_flat_cap() { // drops off the wire entirely, but a `Failed` one is retained (+ // history-capped) so the operator can still triage it. Completing the // node rolls the container up terminal. - q.complete_node(id, c.node_id, Err("boom".to_owned())); + q.complete_node(c.node_id, Err("boom".to_owned())); ids.push(id); } let kept: std::collections::HashSet = q.snapshot().iter().map(|d| d.id).collect(); @@ -1414,7 +1406,7 @@ fn error_is_truncated() { let q = JobQueue::new(1); let id = submit(&q, rebuild("agent-a", "r")); let c = claim_one(&q); - q.complete_node(id, c.node_id, Err("x".repeat(5000))); + q.complete_node(c.node_id, Err("x".repeat(5000))); let snap = q.snapshot(); let err = snap.iter().find(|d| d.id == id).expect("dag").nodes[0] .error @@ -1433,7 +1425,7 @@ fn graceful_stop_shape_signal_drain_reconcile() { for expected in ["set_wanted", "signal", "drain", "reconcile"] { let c = claim_one(&q); assert_eq!(c.kind.as_str(), expected); - q.complete_node(id, c.node_id, Ok(())); + q.complete_node(c.node_id, Ok(())); } assert_eq!(state_of(&q, id), State::Done); } @@ -1451,7 +1443,7 @@ fn graceful_signal_and_drain_hold_no_build_slot() { let kinds: Vec<&str> = heads.iter().map(|c| c.kind.as_str()).collect(); assert_eq!(kinds, vec!["meta_sync", "set_wanted", "set_wanted"]); for c in &heads { - q.complete_node(c.dag_id, c.node_id, Ok(())); + q.complete_node(c.node_id, Ok(())); } // Now the rebuild's Prebuild holds the single slot — and both graceful // stops still proceed to their Signal beside it. @@ -1479,9 +1471,9 @@ fn spawn_shape_provision_create_dropin_reconcile() { let c = claim_one(&q); assert_eq!(c.kind.as_str(), expected); assert_eq!(c.approval_id, Some(7)); - q.complete_node(id, c.node_id, Ok(())); + q.complete_node(c.node_id, Ok(())); } - settle_approval_tail(&q, id, 7, TerminalState::Done); + settle_approval_tail(&q, 7, TerminalState::Done); assert_eq!(state_of(&q, id), State::Done); } @@ -1511,9 +1503,9 @@ fn perm_change_shape_prefixes_rebuild_chain() { ] { let c = claim_one(&q); assert_eq!(c.kind.as_str(), expected); - q.complete_node(id, c.node_id, Ok(())); + q.complete_node(c.node_id, Ok(())); } - settle_rebuild_tail(&q, id, "agent-a", true); + settle_rebuild_tail(&q, "agent-a", true); assert_eq!(state_of(&q, id), State::Done); } @@ -1541,7 +1533,7 @@ fn reparent_shape_is_a_lone_agentless_meta_window_node() { ); assert!(!c.kind.needs_lease()); assert!(!c.kind.needs_build_slot()); - q.complete_node(id, c.node_id, Ok(())); + q.complete_node(c.node_id, Ok(())); assert_eq!(state_of(&q, id), State::Done); } @@ -1562,6 +1554,6 @@ fn reparent_bulk_shape_carries_every_move_on_one_node() { panic!("expected a Reparent node, got {:?}", c.kind); }; assert_eq!(got, &moves); - q.complete_node(id, c.node_id, Ok(())); + q.complete_node(c.node_id, Ok(())); assert_eq!(state_of(&q, id), State::Done); }