vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Compare commits

base: vinzenz:947d45a854d649c458db32b01ee1f79f7cddf5fc
Branches Tags
vinzenz:main
...
compare: vinzenz:a3a0b34668987d93f9a517d866eb07a8c9f3feec
Branches Tags
vinzenz:main
Author SHA1 Message Date
atlas
a3a0b34668 jobq: completion is pub(crate); the DAG container is not a special case 
`submit` used to complete the container node by hand, right after
inserting it, so it would park in `Finishing` and its children unblock.
That was the last caller of `Scheduler::complete` outside the crate, and
the justification was that the container "never needs claiming or
executing".

It does, though, in the sense that matters: it is a node with no logic of
its own, and the scheduler already knows what to do with one. It declares
no resources, so it is claimable the moment it is inserted; `run_node`'s
`Dag` arm already returns `Ok(())`, exactly as it does for `DeployWindow`,
which is the same shape and was never special-cased. Deleting the inline
completion costs one claim round-trip and removes the only reason the
crate had to expose completion at all.

`complete` is `pub(crate)` now. Completion is reachable only from inside
the future `claim_next` hands back, so a node cannot be finished without
the claim it answers, and cannot be claimed without the future that
finishes it. That was the point of the seam.
 2026-08-02 22:00:34 +02:00
atlas
e646656c92 c0re's queue tests no longer drive the scheduler 
The last two claim-driven tests were both arranging node states to observe
something that never needed a run:

`settled_dag_leaves_the_snapshot_despite_its_skipped_branch` completed all
seven nodes of a rebuild to assert the DAG left the snapshot. That is one
predicate over a list of states. `shown_on_wire` is it, split out of
`dag_view`, and the cases can now be named rather than arranged — including
the empty set, the one input where "any" and "all" disagree. It takes
states rather than projected nodes so the caller skips projecting what it
is about to discard; a `NodeView` costs a `build_logs` lookup.

`failed_node_cancels_downstream_but_afterany_reconcile_runs` asserted three
unrelated things from one arranged failure: the cascade (hive_jobq's, and
already tested there), the wire filter (now `shown_on_wire`), and the
roll-up. `DagView::rollup_state` lives in hive-host-sock, which had no
tests at all — it does now, next to the invariant, covering the ordering
its own doc comment says has silently disagreed with the frontend before.

With nothing left claiming, `Claimed` / `ClaimReady` / `CompleteNode` /
`claim_one` / `settle_rebuild_tail` are deleted. Claim/complete sites in
`job_queue/tests.rs`: 109 -> 0.

jobq narrows to match: `settle` is gone (it was a `claim_one` loop
returning a Vec, and its only callers were tests — it lives in the test
module now), `claim_one` is private, and `complete_growing` is
`pub(crate)`. `claim_next` is the whole run-loop surface.

`complete` stays `pub` for one caller, noted at the definition: `submit`
completes a group root with no logic of its own so it parks in `Finishing`
and its children unblock. That is a statement about the node, not an event
to report, and it wants to be expressible at insert time.
 2026-08-02 22:00:34 +02:00
atlas
ab5744a2bd c0re: history retention is a policy, split from the graph it reads 
`visible_dags` mixed two things: walking the graph to classify containers
live-vs-terminal, and the sort-and-truncate that decides what the
dashboard sees. `retain_history` is the second half, generic over the
handle so it is reachable without a graph at all — a `NodeId` cannot be
fabricated, so a test forced to pass real ones could only get them by
submitting and running DAGs.

Which is exactly what the old test did: `MAX_HISTORY_DAGS + 8` submits,
claim and fail each node, read the ids back out of a snapshot — the
scheduler, the roll-up and the wire projection all standing in the path of
a policy that reads none of them. And it only ever exercised the tiebreak,
because every DAG in that loop settled inside the same wall-clock second,
so `finished_at` tied on all of them. Eviction *by time* — the actual
policy — had no coverage. It does now, along with the live-never-competes
case.

`live_count` and `templates::reconcile_only` were both test-only and lose
their last caller here.
 2026-08-02 22:00:34 +02:00
atlas
5f5898d167 c0re: the cancel tests read what cancel left behind 
`cancel_refuses_running_dag` is deleted outright: c0re's `cancel` is a
delegate and hive_jobq already owns that guarantee in
`cancel_node_refuses_a_group_with_anything_running`. Claiming a node here
to prove it was testing the library through the wrapper.

The other three claimed only to ask "what could still run?", which the
graph answers directly. `cancel_clears_queued_dag` and
`cancel_drops_one_agents_branch_leaving_the_rest` now read pending kinds
(the second per-agent, since the point is that one branch died and its
sibling didn't). `cancelled_dag_still_runs_its_approval_tail` reads the
spared *payload* rather than claiming it: the approval template emits one
tail per outcome and which one survives the cancel is the entire
assertion. Its trailing "unrelated activity doesn't disturb it" half no
longer fails a node in the other DAG — the DAG merely existing is enough
to show roll-up is per-DAG.

`Claimed` loses `dag_id` and `agent`; nothing reads them any more.
 2026-08-02 22:00:34 +02:00
atlas
a039a10e40 c0re: the per-agent template tests read declared shape 
`multi_agent_stop` claimed both heads to show they start together, and
`multi_agent_start` completed both heads to show the stale agent rebuilds
first. Neither needs the scheduler: what makes the subgraphs concurrent is
that each head is a group root with no node-deps holding only its own
agent's lease, and the stale fold is a longer chain declared at submit.
Both are readable the moment submit returns.

`declared_shape_for` slices the shape by the agent a payload names — a
hive-wide DAG interleaves one subgraph per agent and the kinds alone can't
tell two `set_wanted` rows apart. `declared_resources_of_kind` does the
same for the whole family of one kind, replacing the hand-rolled lease
extraction in the restart test.

`boot_sweep_nodes_declare_their_own_resources` only claimed to get at two
node ids; `node_of` gets them without running anything.
 2026-08-02 22:00:34 +02:00
atlas
009e9fafae jobq: Scheduler::new_job is gone 
It handed out exactly the builder JobBuilder::new is pub(crate) to
withhold, which was agreed against more than once. I had left it in
place with a docstring naming it as the hole and folded the fix into an
open question. It was not an open question.

It only survived because two hive-c0re tests minted a builder by hand to
simulate a MetaLock growing its cascade. The grown thing is a template,
so the fix is the same as for the deploy graft and the reconcile
fan-out: call it.

exec.rs's MetaLock arm looped rebuild_nodes inline -- the second and last
construction site declaring nodes inside an executor. It is now
templates::grown_rebuilds, so a test can assert what a sweep declares by
calling the same function production calls.

grown_subgraph_roots_on_emitter_and_rebases_local_deps and
meta_update_grows_cascade_in_dag differed only in RebuildOpts; they are
one test over the declared shape, checking a root chain per agent, a
build each, and a drain each because a boot sweep is graceful.

With those gone, CompleteNode::new_job, CompleteNode::complete_node_growing
and drain_meta_syncs had no callers either. hive-jobq's own two growth
tests use JobBuilder::new() directly -- in-crate, so the wall holds.
grep for new_job across the workspace now returns nothing.
 2026-08-02 22:00:34 +02:00
atlas
53cd010a12 job_queue: the reconcile fan-out declares from templates 
exec.rs's Reconcile arm was the one construction site declaring a
resource inline in an executor rather than in templates.rs. It now calls
templates::fanned_out_mechanical, which is where every other declaration
lives -- construction sites state their own holdings.

That also fixes a test which could not fail. The old one claimed a
Reconcile and then re-declared the fan-out itself, commented "same two
calls the scheduler makes, in the same order" -- a copy of production
inside the test. Had exec.rs stopped declaring the lease, it would have
kept passing. The replacement calls the real function and asserts the
declaration, with no DAG run at all.

The other half of the old test -- that a descendant re-enters its
ancestor's grant instead of taking a second unit of a cap-1 lease -- is
hive-jobq's, tested there by
child_borrows_ancestor_grant_released_when_subtree_done and
nested_borrowers_never_deadlock.
 2026-08-02 22:00:34 +02:00
atlas
eab6bce813 job_queue: four more template tests read the graph 
perm_change and the graceful rebuild chain walked their whole DAG to
collect node kinds in order; both now assert declared_shape. The
graceful one gets a sharper claim out of it -- signal and drain go
between the build and the stop, and nothing else changes -- which is
what distinguishes it from the non-graceful chain.

reparent_bulk needed the node's payload rather than its wiring, so
payload_of() reads it off the graph. The assertion is unchanged: one
node carries every move, because bulk atomicity is why a single node was
chosen.

resubmit_while_running_is_new_dag no longer claims a node to stage the
"while running" part. submit appends a container and inserts the
declared group; it never consults the state of any existing node, so a
running earlier DAG cannot change the outcome. The property is no dedup,
covered by identical_resubmit_is_a_distinct_dag -- this one keeps the
named scenario because a config bump mid-build is what people actually
worry about.
 2026-08-02 22:00:34 +02:00
atlas
ca2c479b17 job_queue: lease release is the crate's, not the host's 
Two more tests drove DAGs to completion to watch an agent lease free up
-- one when a single agent's subgraph settled inside a still-running
multi-agent DAG, the other when a whole power op finished. Releasing a
grant once its owner's subtree is terminal is hive-jobq's, covered by
owner_holds_grant_for_its_whole_subtree,
child_borrows_ancestor_grant_released_when_subtree_done and
leaf_owner_goes_done_directly_and_releases.

Their host-side halves are declarations asserted elsewhere: that each
agent's subgraph is an independent root holding only its own lease is in
multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs, and
that a power op emits no tail node is in
cancelled_power_op_runs_no_compensating_node, which checks the DAG has
no pending nodes left at all.
 2026-08-02 22:00:34 +02:00
atlas
ff70bf029d job_queue: drop the cross-DAG contention tests 
Six tests drove two or three DAGs against each other to watch a build
slot or an agent lease serialize them. In every case the part that is
hive-c0re's -- which nodes declare which resource -- is now a
declared_resources table, and the part that is hive-jobq's -- what a
scheduler does with a contended unit -- is tested in hive-jobq
(build_slot_cap_limits_concurrency_and_release_unblocks,
unrelated_nodes_needing_the_same_resource_are_serialized, the two
sibling_borrowers cases, owner_holds_grant_for_its_whole_subtree, and
the fairness test added in the previous commit).

graceful_signal_and_drain_hold_no_build_slot is absorbed rather than
deleted: its claim is a declaration, so it now sits in the graceful-stop
shape test. Signal and Drain declare the agent lease and no build slot,
which is why a whole-hive graceful stop overlaps every agent's drain at
buildSlots = 1 -- the ceiling is one GRACEFUL_STOP_TIMEOUT in total, not
one per agent.

hive-c0re/src/job_queue/tests.rs: 42 tests to 36, 181 lines lighter.
 2026-08-02 22:00:34 +02:00
atlas
3ebfed1226 jobq: pin the fairness guarantee where it is made 
fifo_fairness_for_the_slot lived in hive-c0re and submitted three
rebuilds, driving one to completion to watch the freed build slot go to
the earlier waiter. The guarantee it was checking is this crate's:
claim_one scans nodes in insertion order and takes the first satisfiable
one. Nothing here tested it -- the property hive-jobq provides was
asserted only downstream, through a host's templates.

a_contended_resource_goes_to_the_oldest_waiter tests it directly.
Mutation-checked: reversing the scan order fails it.

What is hive-c0re's is which nodes contend for the slot at all, and that
is a declaration, so its half is now a declared_resources table with
nothing running. The measured shape corrected an assumption on the way:
MetaSync takes the meta window only, and the agent lease starts at
StopForUpdate -- the first node that touches the container -- not at the
head of the chain. Prebuild deliberately holds no lease, which is what
lets it overlap another DAG on the same agent while the container is
still up.

"Uniform hold across the chain" needs no test of its own: a resource is
held for the acquirer's whole subtree, and the parent nesting is already
asserted in rebuild_chain_is_declared_serial.
 2026-08-02 22:00:34 +02:00
atlas
b9f86e415d job_queue: the reconcile gate is the parent chain, not an edge 
swap_ok_runs_post_swap_before_reconcile drove a rebuild DAG to observe
that Reconcile waits for PostSwap. That ordering is not a dependency
between them: Reconcile deps AfterAny(Prebuild), and PostSwap sits
inside Prebuild's subtree, so Prebuild cannot satisfy the edge while
PostSwap is outstanding.

Renamed to say what it checks, and it asserts the parent chain plus that
edge instead of running anything. Kept as its own test rather than
folded into the chain table because the indirection is the easy thing to
break -- flattening the chain preserves every edge and still loses the
guarantee.

swap_failure_still_runs_reconcile is deleted. Its cascade claims are
hive-jobq's, and the "says so on the wire" half was nothing: snapshot
fills NodeView { state: node.state, .. }, a straight copy of the same
State type, so there is no host-side mapping that could disagree.

failed_reconcile_marks_dag_failed is deleted too: a one-node DAG whose
node fails, asserting the DAG reads Failed, is
failed_child_rolls_parent_up_to_failed restated through a c0re template.
 2026-08-02 22:00:34 +02:00
atlas
e9a84310fe job_queue: test grafted work by declaring it, not by grafting it 
Reproducing the runtime path is not needed to test what the runtime path
declares. What DeployApply grows is deploy_rebuild_nodes' output, and
that is a pure declaration -- so declare it directly and read the shape,
instead of running a deploy far enough to graft it.

That makes the finalize gate visible without any of the walking:
finalize_deploy declares AfterOk on both prebuild and reconcile, so
either root failing skips it, and reconcile hangs off prebuild with
AfterAny so a failed swap still reaches it.

deploy_dag_skips_finalize_but_still_tails_a_failed_graft is gone with
it. Its three declared claims are rows in that table, and its runtime
claims belong to hive-jobq: cascade on failure, roll-up, and first_error
digging past a group root that rolled up Failed while carrying no error
of its own -- which is why the DAG reports the swap's error rather than
nothing.

The grafting mechanism itself is also hive-jobq's and tested there: work
lands under the emitter before it settles, and the emitter parks in
Finishing so a downstream AfterAny gate stays shut while its new
children run.
 2026-08-02 22:00:34 +02:00
atlas
7d1709cfc5 job_queue: one deploy-shape table replaces two DAG walks 
deploy_dag_runs_phases_in_order_and_tails_a_failed_apply and
deploy_dag_skips_apply_but_still_runs_tail_when_verify_fails differed
only in where they injected the failure -- apply in one, verify in the
other -- and each drove the whole DAG to watch the compensation tail run
anyway.

Both follow from a single declared edge. The tail accepts
done|failed|skipped on apply, and skipped is exactly the state apply
lands in when verify failed and it never ran. Asserting that edge covers
both cases without running anything.

The runtime halves are hive-jobq's and tested there: a failed dep
cancels its AfterOk dependents while the AfterAny one still runs, and a
parent rolls up Failed from a failed child -- which is what stops an Ok
tail laundering a failed deploy into a success.

Mutation-checked: turning the tail's after_any(apply) into
after_ok(apply) fails the surviving test on that edge alone.
 2026-08-02 22:00:34 +02:00
atlas
6a43fc2e81 job_queue: three more tests read the graph instead of running it 
non_graceful_rebuild_has_no_signal_or_drain claimed and completed six
nodes to collect their kinds. Mutation-checked rather than trusted: with
the template's graceful flag flipped, it fails with signal and drain in
the list. An absence assertion is the easiest kind to make vacuous, so
it is the one that most needs the check.

reparent_shape claimed a node only to get an id for its resource
assertion. node_of() finds it by kind and asserts there is exactly one,
so the test cannot quietly start being about a different node.

multi_agent_restart asserted "both subgraphs start concurrently" by
claiming and seeing two. That is not one fact but two declared ones plus
a jobq guarantee: both heads are group roots with no deps, so nothing
orders them; and each declares only its own agent lease, so nothing
makes them contend. Whether a scheduler then runs independent,
resource-disjoint roots at once belongs to hive-jobq and is tested
there. This asserts the two declarations.
 2026-08-02 22:00:34 +02:00
atlas
96a0679934 job_queue: drop the erased-recipe test infra 
ErasedRecipe and erase() existed for one test, which put three power-op
cases in a single array. Three recipe closures have distinct types and
cannot share an array element type, so all three were boxed. The array
was the reason, not the specs.

The per-case assertion is now a helper taking the already-submitted DAG
id, so each case submits its own spec at its own concrete type. Nothing
is erased and nothing is boxed; production never needed either.

The final assertion is also stronger than the one it replaces.
claim_ready().is_empty() asks what is runnable at this instant, which a
node that is alive but blocked on a dependency passes -- exactly what a
leftover compensating node would look like. It now asserts the DAG has
no Pending nodes at all. It was also a mutating call inside an
assertion: claim_ready settles the graph.
 2026-08-02 22:00:34 +02:00
atlas
d879d3e67a job_queue: assert declared shape instead of driving the DAG 
Three template tests ran a whole DAG -- claim, complete, repeat -- to
observe an order that is fully determined the moment submit returns.
They now read the graph directly: kinds, parent nesting, and dep edges
with the outcome set each accepts.

rebuild_chain_claims_in_dep_order is renamed, because the old name was
wrong about the mechanism and reading it rather than the graph is how
you stay wrong: only half that chain is dep edges. stop_for_update and
swap declare no deps at all and are ordered by parent nesting -- a
node's sub-nodes run after its own logic. Both axes are asserted now,
since a template can break either independently.

declared_shape spells out each edge's accepted outcomes rather than
bucketing them into ok/any. Bucketing made spawn's three
ResolveApproval tails -- which differ only in accepted outcome -- render
as identical rows, which would have made the assertion a tautology.

Checked by mutation against the code under test rather than the
assertion: dropping .after_ok(signal) from the graceful-stop Drain fails
graceful_stop_shape with a diff naming the one missing edge.
 2026-08-02 22:00:34 +02:00
atlas
a59ad5ce3f job_queue: test error truncation as the pure fn it is 
error_is_truncated submitted a DAG, claimed its head, failed it with a
long string and read the error back out of a snapshot -- four moving
parts to observe one `&str -> String`. The DAG round-trip it depended on
is covered by its own tests either way.

Testing truncate_error directly also reaches the case the round-trip
never could: the cap is a byte length, so a multibyte char straddling it
would panic the slice. That boundary scan is the only non-obvious line
in the function and it had no coverage at all -- the old test used a
repeated ASCII 'x', where byte and char offsets coincide.

Also drops a stale claim from insert_job's doc: it has not recorded a
per-node node_rt since that map was deleted.
 2026-08-02 22:00:34 +02:00
atlas
335ad5e0ee jobq: test the growth invariants where they are enforced 
complete_growing had no test at all -- its only reference was the
internal call from claim_next -- so the rule moved into it in the
previous commit was enforced but unproven.

  - a_completing_node_grows_the_work_it_declared: the declared work
    lands under the emitter, and the emitter parks in Finishing rather
    than going terminal. That ordering is the point of growing as part
    of the completion.
  - a_failed_node_grows_nothing: the rule that moved out of the host.

Both were mutation-checked rather than trusted green: with the
Outcome::Failed guard deleted, a_failed_node_grows_nothing fails on the
appended node while its companion still passes, so the test bites and
the drop is specific to failure rather than blanket.

The departed-parent guard beside it stays untested and says so. Nothing
removes a node from the graph yet (eviction stops retaining a DAG; its
nodes linger) and NodeId cannot be fabricated by construction, so a test
would have to fake the precondition it checks. The comment names the
bounded prune as the point at which it becomes testable.
 2026-08-02 22:00:34 +02:00
atlas
8459cc66bd jobq: complete is the completion, and failure grows nothing 
Two review findings from the previous round, re-checked against the
actual tree rather than against my notes.

`Scheduler::complete` was still a public wrapper whose entire body was
`self.finish(id, outcome)`. Its docstring argued the split was not a
redirect because both completion forms shared `finish` -- but sharing a
private helper is not a reason for two public names. `finish`'s body now
lives in `complete`, and `complete_growing` calls it. Same sharing, one
name, no redirect.

Growth on a failed node is now dropped by `complete_growing` instead of
by the host loop. Failure cancel-cascades to every pending child of the
completing node, and grown work is inserted as its children, so anything
appended here is Skipped by the next statement -- the insert is not
wrong, it is provably pointless. That is a consequence of this crate's
cascade rule, so this crate should be the one enforcing it; a host that
has to remember it can forget it. Behaviour is unchanged: hive-c0re
already dropped growth before calling, and now no longer has to.

`Scheduler::new_job` is left alone but documented for what it is: the
hole in `JobBuilder::new`'s pub(crate) wall, with no non-test caller
since claim_next mints a builder per running node. Closing it is a venue
question rather than a rename, so it stays for now.
 2026-08-02 22:00:34 +02:00
atlas
ab53f6710d job_queue: drop Claim, claim_ready and the completion wrappers 
c0re's run loop now goes through hive_jobq's claim_next seam, so the
host layer no longer needs its own claim/complete vocabulary.

exec::run_node takes (NodeId, &NodeKind) instead of a &Claim snapshot.
The agent already rides the payload, and the DAG id is a derived read
(JobQueue::dag_of) that only three arms want, so it is taken per-arm
rather than eagerly for every node. Two arms (WritePermFile, Reparent)
re-matched the kind behind a bail! that could never fire; the match arm
already destructures the payload, so they take it directly now.

Deleted from the c0re layer:
  - struct Claim
  - JobQueue::claim_ready
  - JobQueue::complete_node / complete_node_growing
  - scheduler::NodeDone / handle_completion

Completion happens inside the future claim_next hands back, so "ran the
node but forgot to complete it" is not expressible on the production
path any more. The node done / node failed logging moved with it -- it
lived in handle_completion but is not dead code.

claim_ready and the completion wrappers were left with no non-test
callers, so the tests carry them as ClaimReady / CompleteNode extension
traits over the crate primitives. JobQueue::new_job stays: run_worker
still mints an empty builder on a failed outcome.
 2026-08-02 22:00:34 +02:00
atlas
be1060e52f refactor(#2949): the mutex holds the scheduler, not a wrapper 
`QueueInner` existed to hold the scheduler *and* a per-node side map. The
map is gone, so it was a struct around one field — and worse, a struct of
a type `hive_jobq` cannot drive: the crate's run-loop seam takes
`&Arc<Mutex<Scheduler<..>>>` specifically.

So `JobQueue` now holds `Arc<Mutex<Sched>>` directly, where `Sched` is
just `Scheduler<NodeKind, Resource>`. Its six methods become free
functions over `&Sched`; all six are `DagView` projections, i.e. the code
the endpoint rework is going to delete anyway, so this does not entrench
them.

This is the precondition for c0re calling `claim_next`, not that switch
itself — `run_worker` still claims through `claim_ready`. Landing it
separately keeps the type change reviewable on its own.
 2026-08-02 22:00:34 +02:00
atlas
d1f1a361f0 feat(#2949): claim_next — the seam that cannot be half-used 
The caller supplies how to run a node and spawns what it gets back; it
never touches claiming or completion. The returned future runs the node
*and completes it*, so "forgot to finish the node" stops being something
a caller can do — completion is inside the thing they spawn.

The `Option` is answered synchronously, before anything is awaited, so the
run loop learns whether there was work without waiting on the node it just
started. That is what lets it choose between claiming again immediately
and backing off; an id alone cannot express that choice.

Locking: taken twice, briefly, and never held across the await — once to
claim, once inside the future to complete. A guard alive across an await
point would make the future non-`Send` and unspawnable, which is also why
the node itself runs unlocked for however long it takes. `Arc` +
`std::sync::Mutex` keep this runtime-agnostic: no tokio in this crate.

`run` receives an owned payload rather than a borrow for the same reason a
`&Job` could not be threaded through the executors: a reference parameter
is live for the whole future, borrowing the graph across the await and
poisoning `Send`.

The output carries the insert result instead of swallowing it. This crate
has no logger by design, so a malformed grown job is reported to the
caller, who can log it. The node completes either way — its own work
already happened.
 2026-08-02 22:00:34 +02:00
atlas
9e91bf7813 refactor(#2949): claim_one is the primitive, settle is it in a loop 
One-at-a-time claiming is what lets a caller choose between claiming again
immediately and backing off — a batch return cannot express that choice,
and the choice is the point: the run loop wants to know there was work
before it decides whether to wait.

`settle()` keeps its exact meaning as `while let Some(id) = claim_one()`.
A node started by an earlier iteration is `Running`, not terminal, so it
cannot satisfy another node's dependency in the same sweep; it only
consumes resources. The crate's ~45 existing `settle()` assertions — which
cover resource borrowing, cap-1 serialisation, roll-up and cancellation —
are what verify that equivalence, so it is checked rather than argued.

`None` means "nothing runnable right now", which is deliberately a
different statement from "nothing pending": a node can be pending and
unrunnable because its resources are held elsewhere.

Cost stated rather than left to be found: each `claim_one` rescans the
pending set, so `settle` is O(n^2) in nodes claimed where the single-pass
version was O(n). The graph is bounded by history retention.
 2026-08-02 22:00:34 +02:00
atlas
77cc7bea6b refactor(#2949): the build-log row carries its node id 
`QueueInner` was `{ sched, node_rt }`, where `node_rt` held exactly one
datum per node: the `build_logs` row id. It existed because a `hive_jobq`
node payload is immutable after insert while the log row is created when
the build starts — so the link could not ride the node.

Invert it: the log row names its node (`build_logs.node_id`, one migration
in the existing `schema_versions` framework). Same single-home property,
in the direction the type system allows.

`QueueInner` is now just the scheduler. That is the point: the queue holds
no per-node side map, so nothing has to be locked alongside the graph.

Deleted as a consequence, each surfaced by dead-code analysis after the
edit above rather than predicted:

- `NodeRuntime`, `node_rt`, `set_build_log_id`, and `build_log_id_of`
  (which linear-scanned the map to match a wire `u64` against opaque
  `NodeId`s). The lookup is an indexed query now.
- `struct Ctx`, entirely. It carried `coord` + `dag_id` + `node_id` into
  the executors so the build-log callback could reach the queue; without
  the callback, `coord`/`dag_id` were never read and `node_id` was already
  on the `Claim` both executors receive.
- `QueueInner::node_running`, which existed only for `set_build_log_id`'s
  "only while running" guard.
- The `Fn(i64)` callbacks on `prebuild_toplevel` / `swap_update` /
  `priv_run_inner`, replaced by a `node_id: Option<u64>` passed down. The
  id travels one way now instead of being registered back.

`meta.rs`'s `nix_logged` passes `None` deliberately: its callers reach it
from outside the queue as well as inside, and nothing reads the link for
them yet.

`id_for_node` takes `MAX(id)` rather than assuming uniqueness — a retried
node opens a second row and the panel wants the current attempt. The test
moved to where the behaviour lives and covers that, plus survival across
completion and non-collision with node-less rows.
 2026-08-02 22:00:34 +02:00
atlas
82ef06f445 refactor(#2949): kill Declare — a running node declares onto its own builder 
A node no longer hands back a recipe for the scheduler to replay later. It
declares straight onto a builder it was given, and that builder is inserted
as part of completing the node.

Deleted: `pub type Declare`, `struct NodeOutput` (+ its hand-written `Debug`),
`JobQueue::append_subgraph`. Nothing added to `Dag` / `DagView`.

jobq gains `Scheduler::new_job()` (the only way to obtain a `JobBuilder`) and
`complete_growing(id, outcome, grown)`, which inserts under `id` and *then*
completes it, so a DAG cannot roll terminal while grown work is still pending.
`complete()` and `complete_growing()` share a private `finish()` rather than
one redirecting through the other. The DAG-gone guard lives beside the graph
now, where it cannot be skipped, instead of being a caller-side lookup.

The growth executors return data (`run_meta_lock -> (Vec<String>, RebuildOpts)`,
`run_reconcile -> Option<NodeKind>`) rather than taking the builder: a `&Job`
parameter is live for the whole function body, and `&RefCell<T>` is never
`Send`, so an async fn taking one cannot be spawned. `run_node` threads the
builder by value and hands it back.

A node can now declare work and then fail, which was previously inexpressible.
`grown` is dropped in that case — failure cancel-cascades downstream, so
inserting it would only add nodes to immediately cancel — and the log line
carries `grown_nodes` so the drop is visible.
 2026-08-02 22:00:34 +02:00
atlas
2454a1ea6a jobq: drop JobBuilder's Default impl so it is really unconstructible outside 
The builder's module doc claimed a builder "cannot be constructed, held
or inserted from outside this crate". Two of those three were false:
`new()` is `pub(crate)`, but a hand-written `impl Default for
JobBuilder` is a trait impl on a `pub` type, so it is public regardless
— `JobBuilder::default()` compiled downstream.

Nothing was unsound (`insert_with` stayed `pub(crate)`, so an
outside-built builder could not reach a graph), but the sentence claimed
more than the visibility enforced, which is the bug this crate's docs
have hit before.

Delete the impl; `new()` constructs directly. The doc now says only what
is enforced, and records why there is no `Default` — so the next person
reaching for one finds the reason instead of adding it back.
 2026-08-02 22:00:34 +02:00
14 changed files with 2054 additions and 1983 deletions
Download patch file Download diff file Expand all files Collapse all files

4
hive-c0re/src/dashboard/build_logs.rs
View file

@ -156,7 +156,7 @@ pub(super) async fn get_build_log_for_node(
State(state): State<AppState>,
AxumPath(node_id): AxumPath<u64>,
) -> Response {
match state.coord.job_queue.build_log_id_of(node_id) {
match state.coord.build_logs.id_for_node(node_id) {
Some(log_id) => get_build_log_full(State(state), AxumPath(log_id)).await,
None => (
StatusCode::NOT_FOUND,
@ -183,7 +183,7 @@ pub(super) async fn get_build_log_raw_for_node(
State(state): State<AppState>,
AxumPath(node_id): AxumPath<u64>,
) -> Response {
match state.coord.job_queue.build_log_id_of(node_id) {
match state.coord.build_logs.id_for_node(node_id) {
Some(log_id) => get_build_log_raw(State(state), AxumPath(log_id)).await,
None => (
StatusCode::NOT_FOUND,

419
hive-c0re/src/job_queue/exec.rs
View file

@ -10,11 +10,9 @@ use std::sync::Arc;
use anyhow::{Context as _, Result};
use super::{Claim, Declare};
use hive_jobq::TerminalState;
use hive_jobq::{NodeId, TerminalState};
use super::model::NodeKind;
use super::resource::Resource;
use crate::coordinator::Coordinator;
use crate::power::{ReconcileAction, reconcile_action};
@ -26,109 +24,104 @@ use crate::power::{ReconcileAction, reconcile_action};
/// N × this timeout.
pub const GRACEFUL_STOP_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(3);
/// Extra signal an executor hands back to the scheduler alongside
/// success.
#[derive(Default)]
pub struct NodeOutput {
/// Whole per-agent *subgraphs* to append into *this same* DAG at
/// runtime — the single in-DAG-growth channel. Each [`Job`] is one
/// independent subgraph, declared but not yet inserted: an executor cannot
/// reach the queue, so it hands the declaration back and the scheduler
/// inserts it via [`super::JobQueue::append_subgraph`] under its own lock,
/// rooted on the emitting node. Used both for the multi-node case
/// (`MetaLock` growing one rebuild subgraph per agent — the startup
/// sweep's stale agents, the meta-update cascade's affected agents) and
/// the single-node case (a `Reconcile` planner emitting its mechanical
/// `Start` / `Stop` as a one-node subgraph). The scheduler applies these
/// *before* the emitting node's completion so the DAG never rolls terminal
/// with the appended work still pending — keeping the lease-window
/// transient held across the sub-step.
pub append_subgraph: Vec<Declare>,
}
impl std::fmt::Debug for NodeOutput {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
// The subgraphs are closures — how many were emitted is the only thing
// there is to say about them before the queue runs them.
f.debug_struct("NodeOutput")
.field("append_subgraph", &self.append_subgraph.len())
.finish()
}
}
/// Build-log sink for one claimed node.
struct Ctx<'a> {
coord: &'a Arc<Coordinator>,
dag_id: u64,
node_id: super::NodeId,
}
impl Ctx<'_> {
fn build_log(&self, log_id: i64) {
if self
.coord
.job_queue
.set_build_log_id(self.dag_id, self.node_id, log_id)
{
self.coord.emit_rebuild_queue_snapshot();
}
}
}
/// Run one claimed node to completion. Called from a task the
/// scheduler spawns per claim; the `Result` (stringified) becomes the
/// node's terminal state.
pub(super) async fn run_node(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
let ctx = Ctx {
coord,
dag_id: claim.dag_id,
node_id: claim.node_id,
};
match &claim.kind {
NodeKind::MetaSync { relock, .. } => run_meta_sync(coord, claim, *relock).await,
NodeKind::Prebuild { .. } => run_prebuild(claim, &ctx).await,
NodeKind::Swap { .. } => run_swap(coord, claim, &ctx).await,
NodeKind::PostSwap { .. } => run_post_swap(coord, claim).await,
NodeKind::Provision { .. } => run_provision(coord, claim).await,
NodeKind::Create { .. } => run_create(claim).await,
///
/// `job` is the node's own growth channel: an executor that decides more work
/// is needed declares it here, and the scheduler inserts it under this node
/// when the node completes. Most executors never touch it. Nothing is inserted
/// while the node runs — the builder is local state, so this stays outside the
/// queue's lock for the whole (often multi-minute) execution.
///
/// ⚠️ Taken **by value and handed back**, not by reference. A `JobBuilder` is
/// `RefCell`-backed: owned it is `Send`, but `&JobBuilder` is not (a shared ref
/// is `Send` only if the referent is `Sync`, and `RefCell` never is). A `&Job`
/// parameter would be live across every `.await` in this fn and make the whole
/// future non-`Send`, which the scheduler's `tokio::spawn` rejects. So the
/// growth executors below return *what to grow* and the declaration happens
/// here, synchronously, between awaits.
///
/// The node is identified by its own id + payload rather than by a `Claim`
/// side-struct: `kind` already carries the agent, and the DAG id is a
/// derived read (`JobQueue::dag_of`) the three arms that need it take
/// themselves. Nothing here needs a claim to exist as a type.
pub(super) async fn run_node(
coord: &Arc<Coordinator>,
job: super::Job,
id: NodeId,
kind: &NodeKind,
) -> (super::Job, Result<()>) {
// The agent this node targets rides the payload — empty for the agentless
// container kinds (`MetaLock`, `Dag`), which never read it.
let agent = kind.agent();
// Every arm is `Result<()>`; the three that grow work declare into `job`
// *synchronously*, after their own awaits have finished. Borrowing `&job`
// inside an `.await` would make this future non-`Send` (see above), so the
// growth executors return what to grow rather than taking the builder.
let result = match kind {
NodeKind::MetaSync { relock, .. } => run_meta_sync(coord, agent, *relock).await,
NodeKind::Prebuild { .. } => run_prebuild(agent, id).await,
NodeKind::Swap { .. } => run_swap(coord, agent, id).await,
NodeKind::PostSwap { .. } => run_post_swap(coord, agent).await,
NodeKind::Provision { .. } => run_provision(coord, agent).await,
NodeKind::Create { .. } => run_create(agent).await,
NodeKind::MetaLock {
sweep,
fanout,
inputs,
} => run_meta_lock(coord, *sweep, fanout.clone(), inputs).await,
NodeKind::Reconcile { .. } => run_reconcile(coord, claim).await,
NodeKind::Start { .. } => run_start(coord, claim).await,
NodeKind::Stop { .. } => run_stop(coord, claim).await,
NodeKind::StopForUpdate { .. } => run_stop_for_update(coord, claim).await,
NodeKind::Signal { .. } => Ok(run_signal(coord, claim)),
NodeKind::Drain { .. } => run_drain(coord, claim).await,
NodeKind::WriteDropin { .. } => run_write_dropin(coord, claim).await,
NodeKind::WritePermFile { .. } => run_write_perm_file(coord, claim).await,
NodeKind::Reparent { .. } => run_reparent(coord, claim).await,
} => run_meta_lock(coord, *sweep, fanout.clone(), inputs)
.await
.map(|(agents, opts)| super::templates::grown_rebuilds(&job, &agents, opts)),
NodeKind::Reconcile { .. } => run_reconcile(coord, agent).await.map(|sub| {
if let Some(kind) = sub {
super::templates::fanned_out_mechanical(&job, kind);
}
}),
NodeKind::Start { .. } => run_start(coord, agent).await,
NodeKind::Stop { .. } => run_stop(coord, agent).await,
NodeKind::StopForUpdate { .. } => run_stop_for_update(coord, agent).await,
NodeKind::Signal { .. } => {
run_signal(coord, agent);
Ok(())
}
NodeKind::Drain { .. } => run_drain(coord, agent).await,
NodeKind::WriteDropin { .. } => run_write_dropin(coord, agent).await,
// The payload rides the node and is destructured here, so the executor
// takes it directly instead of re-matching the kind behind a `bail!`
// that could never fire.
NodeKind::WritePermFile { payload, .. } => run_write_perm_file(coord, agent, payload).await,
NodeKind::Reparent { moves } => run_reparent(coord, moves).await,
NodeKind::MergeVerify { approval_id, .. } => run_merge_verify(coord, *approval_id).await,
NodeKind::DeployApply { approval_id, .. } => {
run_deploy_apply(coord, claim, *approval_id).await
run_deploy_apply(coord, *approval_id).await.map(|()| {
super::templates::deploy_rebuild_nodes(&job, agent, *approval_id);
})
}
NodeKind::FinalizeDeploy { approval_id, .. } => {
run_finalize_deploy(coord, *approval_id).await
}
NodeKind::DeployTail { approval_id, .. } => {
run_deploy_tail(coord, claim, *approval_id).await
run_deploy_tail(coord, coord.job_queue.dag_of(id), agent, *approval_id).await
}
NodeKind::ResolveApproval {
approval_id,
outcome,
} => run_resolve_approval(coord, claim, *approval_id, *outcome).await,
NodeKind::EmitRebuilt { ok, .. } => Ok(run_emit_rebuilt(coord, claim, *ok)),
NodeKind::SetWanted { up, .. } => run_set_wanted(coord, claim, *up),
} => run_resolve_approval(coord, coord.job_queue.dag_of(id), *approval_id, *outcome).await,
NodeKind::EmitRebuilt { ok, .. } => {
run_emit_rebuilt(coord, agent, coord.job_queue.dag_of(id), *ok);
Ok(())
}
NodeKind::SetWanted { up, .. } => run_set_wanted(coord, agent, *up),
// The two nodes that carry no work of their own; completing either
// lets it reach `Finishing` so the nodes under it start.
// - `Dag`: pure grouping container. The DAG's terminal side effect, if
// any, is its own tail node in the graph.
// - `DeployWindow`: pure resource holder — the meta window, agent lease
// and build slot it declares stay held until its subtree settles.
NodeKind::Dag { .. } | NodeKind::DeployWindow { .. } => Ok(NodeOutput::default()),
}
NodeKind::Dag { .. } | NodeKind::DeployWindow { .. } => Ok(()),
};
(job, result)
}
/// Resolve the DAG's approval row the way this node's own `outcome` says.
@ -140,31 +133,30 @@ pub(super) async fn run_node(coord: &Arc<Coordinator>, claim: &Claim) -> Result<
/// since the work already happened and failing the tail would only misreport it.
async fn run_resolve_approval(
coord: &Arc<Coordinator>,
claim: &Claim,
dag_id: Option<u64>,
approval_id: i64,
outcome: TerminalState,
) -> Result<NodeOutput> {
) -> Result<()> {
let reason = (outcome == TerminalState::Failed)
.then(|| coord.job_queue.first_error(claim.dag_id))
.then(|| dag_id.and_then(|dag| coord.job_queue.first_error(dag)))
.flatten();
crate::actions::resolve_approval_dag(coord, approval_id, outcome, reason.as_deref()).await;
Ok(NodeOutput::default())
Ok(())
}
/// Emit this agent's `Rebuilt` manager event. `ok` is not computed — it is which
/// of the tail pair the graph let run. The failure note comes from the DAG's
/// first failing node, since the branch knows *that* it failed but not *why*.
fn run_emit_rebuilt(coord: &Arc<Coordinator>, claim: &Claim, ok: bool) -> NodeOutput {
fn run_emit_rebuilt(coord: &Arc<Coordinator>, agent: &str, dag_id: Option<u64>, ok: bool) {
coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt {
agent: claim.agent.clone(),
agent: agent.to_owned(),
ok,
note: (!ok)
.then(|| coord.job_queue.first_error(claim.dag_id))
.then(|| dag_id.and_then(|dag| coord.job_queue.first_error(dag)))
.flatten(),
sha: None,
tag: None,
});
NodeOutput::default()
}
/// Write the agent's durable power intent — the DAG-node form of the old
@ -175,7 +167,7 @@ fn run_emit_rebuilt(coord: &Arc<Coordinator>, claim: &Claim, ok: bool) -> NodeOu
/// warn-and-continue write, a failed write fails the node (cancel-downstream
/// cancels the `Reconcile`) rather than letting it converge to a stale
/// intent — that atomicity is the point of moving it into the DAG.
fn run_set_wanted(coord: &Arc<Coordinator>, claim: &Claim, up: bool) -> Result<NodeOutput> {
fn run_set_wanted(coord: &Arc<Coordinator>, agent: &str, up: bool) -> Result<()> {
let wanted = if up {
crate::power::Wanted::Up
} else {
@ -183,9 +175,9 @@ fn run_set_wanted(coord: &Arc<Coordinator>, claim: &Claim, up: bool) -> Result<N
};
coord
.power
.set(&claim.agent, wanted)
.with_context(|| format!("set wanted={} for agent {}", wanted.as_str(), claim.agent))?;
Ok(NodeOutput::default())
.set(agent, wanted)
.with_context(|| format!("set wanted={} for agent {agent}", wanted.as_str()))?;
Ok(())
}
/// The rebuild's meta preamble: runtime-dir prep, an idempotent meta
@ -197,12 +189,7 @@ fn run_set_wanted(coord: &Arc<Coordinator>, claim: &Claim, up: bool) -> Result<N
/// Deliberately a separate node from the [`run_prebuild`] it feeds: that
/// build takes minutes and only *reads* the store, so keeping the global
/// window off it is what lets rebuilds of different agents overlap.
async fn run_meta_sync(
coord: &Arc<Coordinator>,
claim: &Claim,
relock: bool,
) -> Result<NodeOutput> {
let name = &claim.agent;
async fn run_meta_sync(coord: &Arc<Coordinator>, name: &str, relock: bool) -> Result<()> {
// Runs while the agent is still up — the runtime dir and MCP listener
// already exist. Use the pure path accessor; no need to re-register the
// listener (event-driven: registered at start/create).
@ -219,7 +206,7 @@ async fn run_meta_sync(
if relock {
crate::meta::lock_update_for_rebuild(name).await?;
}
Ok(NodeOutput::default())
Ok(())
}
/// Out-of-band toplevel build while the container keeps serving: warm
@ -229,18 +216,16 @@ async fn run_meta_sync(
/// container is already down: its only purpose is to shrink the swap's
/// downtime window, so a stopped agent (no uptime to preserve) doesn't
/// pay the double eval — `Swap` builds inline instead.
async fn run_prebuild(claim: &Claim, ctx: &Ctx<'_>) -> Result<NodeOutput> {
let name = &claim.agent;
async fn run_prebuild(name: &str, id: NodeId) -> Result<()> {
// Warm the toplevel build only when the container is up — the whole
// point of prebuild is to shrink the swap's downtime window. A
// stopped agent has no uptime to preserve, so skip the (expensive)
// eval and let the downstream `Swap` build inline.
if crate::lifecycle::is_running(name).await {
let flake_ref = format!("{}#{name}", crate::paths::meta_root().display());
crate::lifecycle::prebuild_toplevel(name, &flake_ref, &|log_id| ctx.build_log(log_id))
.await?;
crate::lifecycle::prebuild_toplevel(name, &flake_ref, Some(id.get())).await?;
}
Ok(NodeOutput::default())
Ok(())
}
/// Profile-swap: re-apply drop-ins (rebuild is the reconcile verb),
@ -248,17 +233,13 @@ async fn run_prebuild(claim: &Claim, ctx: &Ctx<'_>) -> Result<NodeOutput> {
/// (rev marker, `Rebuilt` event, forge/matrix sync, kick, rescan).
/// The recovery-start on failure is NOT here — the DAG's tail
/// `Reconcile` runs after this node terminal ok *or* fail.
async fn run_swap(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> Result<NodeOutput> {
let name = &claim.agent;
async fn run_swap(coord: &Arc<Coordinator>, name: &str, id: NodeId) -> Result<()> {
// Swap runs on an already-existing (stopped) container — runtime dir
// and listener were created earlier. Pure path accessor suffices.
let agent_dir = crate::paths::agent_runtime_dir(name);
let hive = coord.hive_env();
let paths = Coordinator::agent_paths(name, agent_dir);
let result = crate::lifecycle::swap_update(name, &hive, &paths, &|log_id| {
ctx.build_log(log_id);
})
.await;
let result = crate::lifecycle::swap_update(name, &hive, &paths, Some(id.get())).await;
// On success the Ok-only bookkeeping tail (rev marker, forge/matrix
// sync, kick, rescan, snapshot) runs in the sibling `PostSwap` node,
// which deps `AfterOk(Swap)`. On failure `PostSwap` is cancel-cascaded
@ -269,7 +250,7 @@ async fn run_swap(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> Res
if result.is_err() {
coord.rescan_containers_and_emit().await;
}
result.map(|()| NodeOutput::default())
result
}
/// The post-`Swap` bookkeeping tail, split into its own node for dashboard
@ -277,8 +258,7 @@ async fn run_swap(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> Res
/// means the profile swap succeeded. Store/forge/matrix work only — no nix
/// build (build-slot-exempt); the agent lease taken at `Swap` is still held
/// (the whole chain up to `Reconcile` is one agent's subgraph).
async fn run_post_swap(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
let name = &claim.agent;
async fn run_post_swap(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
if let Some(rev) = crate::auto_update::current_flake_rev(&coord.hyperhive_flake)
&& let Err(e) = std::fs::write(crate::paths::applied_rev_marker(name), rev)
{
@ -298,20 +278,19 @@ async fn run_post_swap(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOu
coord.kick_agent(name, "container rebuilt");
coord.rescan_containers_and_emit().await;
crate::dashboard::emit_meta_inputs_snapshot(coord);
Ok(NodeOutput::default())
Ok(())
}
/// First-spawn pre-create provisioning: proposed/applied repos, state
/// subvolume, and the meta `sync_agents` registration. Runs under the
/// deploy window (it declares `Resource::MetaWindow`) so its commit can't
/// land inside another node's staged deploy window.
async fn run_provision(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
let name = &claim.agent;
async fn run_provision(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
let agent_dir = crate::paths::agent_runtime_dir(name);
let hive = coord.hive_env();
let paths = Coordinator::agent_paths(name, agent_dir);
crate::lifecycle::provision_container(name, &hive, &paths).await?;
Ok(NodeOutput::default())
Ok(())
}
/// `nixos-container create` proper — the upstream `Provision` node
@ -320,21 +299,24 @@ async fn run_provision(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOu
/// dir creation and MCP listener registration are deferred to the tail
/// `Reconcile` (`converge_start_preamble` + `register_agent`) so this
/// node stays purely "create", not "create + start".
async fn run_create(claim: &Claim) -> Result<NodeOutput> {
crate::lifecycle::create_only(&claim.agent).await?;
Ok(NodeOutput::default())
async fn run_create(name: &str) -> Result<()> {
crate::lifecycle::create_only(name).await?;
Ok(())
}
/// Meta flake lock bump. Boot-sweep flavour is non-fatal (a failed
/// bump must not cancel the fan-out rebuilds — they proceed against
/// the current lock, exactly like today's sweep); the meta-update
/// flavour propagates errors, and a failed bump fans out nothing.
/// Returns the agents whose rebuild subgraphs the caller should grow into this
/// node, and the options to build them with — rather than declaring them here.
/// The declaration has to happen outside any `.await` (see [`run_node`]).
async fn run_meta_lock(
coord: &Arc<Coordinator>,
sweep: bool,
fanout: Option<Vec<String>>,
inputs: &[String],
) -> Result<NodeOutput> {
) -> Result<(Vec<String>, super::templates::RebuildOpts)> {
if sweep {
if let Err(e) = crate::meta::lock_update_hyperhive().await {
tracing::warn!(error = ?e, "startup sweep: meta lock_update_hyperhive failed");
@ -349,25 +331,13 @@ async fn run_meta_lock(
// drain window rather than being cut off. The per-agent drains overlap,
// so the sweep's cost ceiling is one `GRACEFUL_STOP_TIMEOUT` in total,
// not one per agent.
let append_subgraph = fanout
.unwrap_or_default()
.iter()
.map(|agent| {
let agent = agent.clone();
Box::new(move |b: &super::Job| {
super::templates::rebuild_nodes(
b,
&agent,
super::templates::RebuildOpts {
relock: true,
graceful: true,
},
None,
);
}) as Declare
})
.collect();
return Ok(NodeOutput { append_subgraph });
return Ok((
fanout.unwrap_or_default(),
super::templates::RebuildOpts {
relock: true,
graceful: true,
},
));
}
let _progress = coord.meta_update_guard();
crate::meta::lock_update(inputs).await?;
@ -383,69 +353,46 @@ async fn run_meta_lock(
// cascade children must NOT re-lock, which would revert the bump this
// node just committed (the property the old `fanout_specs` meta-update
// branch encoded).
let append_subgraph = cascade
.iter()
.map(|agent| {
let agent = agent.clone();
Box::new(move |b: &super::Job| {
super::templates::rebuild_nodes(
b,
&agent,
super::templates::RebuildOpts {
relock: false,
graceful: false,
},
None,
);
}) as Declare
})
.collect();
Ok(NodeOutput { append_subgraph })
Ok((
cascade,
super::templates::RebuildOpts {
relock: false,
graceful: false,
},
))
}
/// Idempotent power-converge *planner*: compare `wanted` (durable
/// intent) against observed state and, when they diverge, fan the
/// mechanical `Start` / `Stop` out as a first-class node appended to
/// *this* DAG (a single-node `NodeOutput::append_subgraph` rooted on
/// this node). Does no container work itself — the sub-step becomes
/// visible in the DAG and the lease-window transient (or the sub-step's
/// own node-local guard) rides across it.
async fn run_reconcile(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
let name = &claim.agent;
/// *this* DAG (a single node declared into `job`, rooted on this node).
/// Does no container work itself — the sub-step becomes visible in the
/// DAG and the lease-window transient (or the sub-step's own node-local
/// guard) rides across it.
/// Returns the mechanical node to fan out (`None` on a noop) rather than
/// declaring it — the declaration has to happen outside any `.await`, see
/// [`run_node`]. `NodeKind` carries the agent it targets, so this node's agent
/// is stamped into the fanned-out kind here.
async fn run_reconcile(coord: &Arc<Coordinator>, name: &str) -> Result<Option<NodeKind>> {
let running = crate::lifecycle::is_running(name).await;
let wanted = coord.power.get_or_seed(name, running)?;
// One node targeting this agent, rooted on this reconcile node. `NodeKind`
// carries the agent it targets, so stamp `claim.agent` into the fanned-out
// Start/Stop kind (one in-DAG-growth channel).
let sub = |kind: NodeKind| {
// `Start` / `Stop` declare the lease they run under. This node is their
// parent and holds it, so the declaration is a re-entrant borrow — no
// second unit, no deadlock. It exists so the requirement belongs to the
// node rather than to the fact that a `Reconcile` happens to fan it out.
let lease = Resource::Agent(kind.agent().to_owned());
vec![Box::new(move |b: &super::Job| {
let _ = b.node(kind).needs(lease);
}) as Declare]
};
let append_subgraph = match reconcile_action(wanted, running) {
ReconcileAction::Start => sub(NodeKind::Start {
agent: name.clone(),
Ok(match reconcile_action(wanted, running) {
ReconcileAction::Start => Some(NodeKind::Start {
agent: name.to_owned(),
}),
ReconcileAction::Stop => sub(NodeKind::Stop {
agent: name.clone(),
ReconcileAction::Stop => Some(NodeKind::Stop {
agent: name.to_owned(),
}),
ReconcileAction::Noop => {
tracing::debug!(%name, wanted = wanted.as_str(), running, "reconcile: noop");
Vec::new()
None
}
};
Ok(NodeOutput { append_subgraph })
})
}
/// Mechanical container start — the sub-step a `Reconcile` planner fans
/// out when it observes `wanted = Up` and the container down.
async fn run_start(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
let name = &claim.agent;
async fn run_start(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
// No node-local transient guard: the pill is derived from the running node
// set, and `Start` reports `Starting` via `NodeKind::transient_kind`. This
// used to take one "only when the DAG holds none", which was a second
@ -468,28 +415,26 @@ async fn run_start(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput
coord.register_agent(name)?;
coord.kick_agent(name, "container started");
coord.rescan_containers_and_emit().await;
Ok(NodeOutput::default())
Ok(())
}
/// Mechanical container stop — the sub-step a `Reconcile` planner fans
/// out when it observes `wanted = Offline` and the container up.
async fn run_stop(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
let name = &claim.agent;
async fn run_stop(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
// See `run_start`: no node-local guard — `Stop` reports `Stopping` from its
// own kind now.
crate::lifecycle::kill(name).await?;
coord.unregister_agent(name);
coord.notify_manager(&hive_sh4re::HelperEvent::Killed {
agent: name.clone(),
agent: name.to_owned(),
});
coord.rescan_containers_and_emit().await;
Ok(NodeOutput::default())
Ok(())
}
/// Mechanical stop for the profile swap. Never *changes* `wanted`;
/// noop when already stopped.
async fn run_stop_for_update(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
let name = &claim.agent;
async fn run_stop_for_update(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
if crate::lifecycle::is_running(name).await {
// Seed a missing agent_power row from the PRE-stop observation
// — the DAG's tail `Reconcile` observes only the mechanically
@ -501,7 +446,7 @@ async fn run_stop_for_update(coord: &Arc<Coordinator>, claim: &Claim) -> Result<
crate::lifecycle::kill(name).await?;
coord.rescan_containers_and_emit().await;
}
Ok(NodeOutput::default())
Ok(())
}
/// Set the graceful fence + kick so the harness sees it promptly and
@ -513,20 +458,18 @@ async fn run_stop_for_update(coord: &Arc<Coordinator>, claim: &Claim) -> Result<
/// `GRACEFUL_STOP_TIMEOUT`. Safe because the harness tests the marker at
/// the top of its loop — a paused agent has no turn in flight, so there
/// is nothing to checkpoint.
fn run_signal(coord: &Arc<Coordinator>, claim: &Claim) -> NodeOutput {
if hive_types::Ident::parse(&claim.agent).is_ok_and(|a| Coordinator::is_paused(&a)) {
return NodeOutput::default();
fn run_signal(coord: &Arc<Coordinator>, name: &str) {
if hive_types::Ident::parse(name).is_ok_and(|a| Coordinator::is_paused(&a)) {
return;
}
coord.mark_graceful_stop(&claim.agent);
coord.kick_agent(&claim.agent, "graceful stop requested");
NodeOutput::default()
coord.mark_graceful_stop(name);
coord.kick_agent(name, "graceful stop requested");
}
/// Await the harness clearing the fence (`GracefulStopComplete`) or
/// the timeout — either way the downstream `Reconcile` proceeds with
/// the actual stop.
async fn run_drain(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
let name = &claim.agent;
async fn run_drain(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
let deadline = std::time::Instant::now() + GRACEFUL_STOP_TIMEOUT;
while coord.is_graceful_stop_pending(name) {
if std::time::Instant::now() >= deadline {
@ -536,12 +479,11 @@ async fn run_drain(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
}
coord.clear_graceful_stop(name);
Ok(NodeOutput::default())
Ok(())
}
/// `set_nspawn_flags` + `set_resource_limits` + daemon-reload.
async fn run_write_dropin(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
let name = &claim.agent;
async fn run_write_dropin(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
// write_dropins only needs the path value to build AgentPaths; the
// dir doesn't need to exist at this point (created by ensure_agent_runtime_dir
// on the upstream Prebuild/Start node).
@ -549,19 +491,18 @@ async fn run_write_dropin(coord: &Arc<Coordinator>, claim: &Claim) -> Result<Nod
let hive = coord.hive_env();
let paths = Coordinator::agent_paths(name, agent_dir);
crate::lifecycle::write_dropins(name, &hive, &paths).await?;
Ok(NodeOutput::default())
Ok(())
}
/// Write + commit the perm file(s) (fused under `META_LOCK` so the
/// working tree is never left dirty), then emit the P3RM1SS10NS-tab
/// snapshots so the dashboard reflects the new assignment.
async fn run_write_perm_file(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
async fn run_write_perm_file(
coord: &Arc<Coordinator>,
name: &str,
payload: &super::model::PermPayload,
) -> Result<()> {
use super::model::PermPayload;
let name = &claim.agent;
// The perm file payload rides the node itself (the only consumer).
let NodeKind::WritePermFile { payload, .. } = &claim.kind else {
anyhow::bail!("run_write_perm_file on a non-WritePermFile node");
};
// Runs under the deploy window (it declares `Resource::MetaWindow`): a
// perm commit landing inside another node's staged prepare→finalize
// window would sweep the staged deploy lock into its commit (the
@ -591,7 +532,7 @@ async fn run_write_perm_file(coord: &Arc<Coordinator>, claim: &Claim) -> Result<
}
}
}
Ok(NodeOutput::default())
Ok(())
}
/// Apply the node's `(child, new_parent)` moves as one `META_LOCK`-fused
@ -601,10 +542,10 @@ async fn run_write_perm_file(coord: &Arc<Coordinator>, claim: &Claim) -> Result<
/// the deploy window (it declares `Resource::MetaWindow`), same reasoning as
/// `run_write_perm_file`: a topology commit landing inside another node's
/// staged deploy window would sweep the staged lock into its commit.
async fn run_reparent(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
let NodeKind::Reparent { moves } = &claim.kind else {
anyhow::bail!("run_reparent on a non-Reparent node");
};
async fn run_reparent(
coord: &Arc<Coordinator>,
moves: &[(hive_types::Ident, Option<hive_types::Ident>)],
) -> Result<()> {
let refs: Vec<(&str, Option<&str>)> = moves
.iter()
.map(|(child, parent)| {
@ -618,16 +559,14 @@ async fn run_reparent(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOut
.reparent_bulk_with_notify(&refs)
.await
.map_err(|e| anyhow::anyhow!(e))?;
Ok(NodeOutput::default())
Ok(())
}
/// Deploy phase 1 — drift gate, fetch, eval-verify. Mutates nothing, so a
/// failure here cancel-cascades the rest of the subtree with the forge and the
/// applied repo exactly as they were.
async fn run_merge_verify(coord: &Arc<Coordinator>, approval_id: i64) -> Result<NodeOutput> {
crate::actions::run_deploy_merge_verify(coord, approval_id)
.await
.map(|()| NodeOutput::default())
async fn run_merge_verify(coord: &Arc<Coordinator>, approval_id: i64) -> Result<()> {
crate::actions::run_deploy_merge_verify(coord, approval_id).await
}
/// Deploy phase 2 — the irreversible half: ff-merge, then phase 1 of the
@ -639,27 +578,15 @@ async fn run_merge_verify(coord: &Arc<Coordinator>, approval_id: i64) -> Result<
/// their `MetaSync` declares is re-entered rather than deadlocked against the
/// ancestor already holding it. On failure nothing is appended and the tail
/// compensates, exactly as before.
async fn run_deploy_apply(
coord: &Arc<Coordinator>,
claim: &Claim,
approval_id: i64,
) -> Result<NodeOutput> {
crate::actions::run_deploy_apply(coord, approval_id).await?;
Ok(NodeOutput {
append_subgraph: vec![super::templates::deploy_rebuild_nodes(
claim.kind.agent(),
approval_id,
)],
})
async fn run_deploy_apply(coord: &Arc<Coordinator>, approval_id: i64) -> Result<()> {
crate::actions::run_deploy_apply(coord, approval_id).await
}
/// Deploy phase 3 — close the staged-lock window once the appended rebuild has
/// come up clean: drop the rollback ref, plant the `deployed/<id>` tag, commit
/// the staged lock.
async fn run_finalize_deploy(coord: &Arc<Coordinator>, approval_id: i64) -> Result<NodeOutput> {
crate::actions::run_finalize_deploy(coord, approval_id)
.await
.map(|()| NodeOutput::default())
async fn run_finalize_deploy(coord: &Arc<Coordinator>, approval_id: i64) -> Result<()> {
crate::actions::run_finalize_deploy(coord, approval_id).await
}
/// Deploy compensation + bookkeeping tail. `AfterAny` the apply node, so it
@ -671,12 +598,12 @@ async fn run_finalize_deploy(coord: &Arc<Coordinator>, approval_id: i64) -> Resu
/// the approval row is gone (deny race, purge).
async fn run_deploy_tail(
coord: &Arc<Coordinator>,
claim: &Claim,
dag_id: Option<u64>,
agent: &str,
approval_id: i64,
) -> Result<NodeOutput> {
crate::actions::run_deploy_tail(coord, Some(claim.dag_id), claim.kind.agent(), approval_id)
.await;
Ok(NodeOutput::default())
) -> Result<()> {
crate::actions::run_deploy_tail(coord, dag_id, agent, approval_id).await;
Ok(())
}
/// Compute which agents a `nix flake update <inputs>` on the meta

641
hive-c0re/src/job_queue/mod.rs
View file

@ -36,8 +36,7 @@ pub mod templates;
#[cfg(test)]
mod tests;
use std::collections::HashMap;
use std::sync::Mutex;
use std::sync::{Arc, Mutex};
use chrono::{DateTime, Utc};
use hive_host_sock::jobs::NodeView;
@ -55,16 +54,6 @@ use resource::Resource;
/// borrowed one; only `hive_jobq` can make or insert it.
pub type Job = hive_jobq::JobBuilder<NodeKind, Resource>;
/// A job's shape as a **recipe**: given a builder, declare the nodes.
///
/// What a template returns and what an executor hands back, because neither
/// can build a job itself — `hive_jobq` creates the builder inside its own
/// insertion call and never lets one out. So the transferable thing is the
/// declaring closure, and the queue runs it at the moment it inserts.
///
/// `Send` because an executor's output crosses the scheduler's task boundary.
pub type Declare = Box<dyn FnOnce(&Job) + Send>;
/// A handle to one node a template declared — where its edges, grouping and
/// resources are declared. `Copy`; naming a node as a dependency does not
/// consume the ability to name it again.
@ -99,27 +88,6 @@ pub struct RunningTransient {
pub since: DateTime<Utc>,
}
/// A node claimed for execution — everything the executor needs, snapshotted at
/// claim time.
#[derive(Debug, Clone)]
pub struct Claim {
pub dag_id: u64,
pub node_id: NodeId,
pub kind: NodeKind,
/// The agent this node targets (its own, not a DAG-level field). Empty for
/// the agentless [`NodeKind::MetaLock`] + [`NodeKind::Dag`] container nodes.
pub agent: String,
}
/// Per-node runtime metadata the crate graph doesn't carry. Lifecycle
/// (`started_at` / `finished_at` / `error`) lives on the `hive_jobq::Node`
/// itself now, so only the build-log row link remains host-side (the
/// client fetches the log by node id).
#[derive(Debug, Default, Clone)]
struct NodeRuntime {
build_log_id: Option<i64>,
}
/// An owned read-view of a DAG container's carried metadata ([`NodeKind::Dag`]).
/// Derived on read from the container node — the data has a single home (the
/// node payload); this is not a stored side-table.
@ -129,24 +97,29 @@ struct DagMeta {
created_at: DateTime<Utc>,
}
/// The mutable queue state behind the mutex: the crate scheduler plus the
/// per-node runtime metadata the graph can't carry. A **DAG is a single
/// container node** ([`NodeKind::Dag`], `parent = None`) whose subtree is the
/// DAG's work — so the container's `NodeId` is the DAG id, its rolled-up state
/// is the DAG state, and there are no grouping side-tables: membership + meta
/// are graph queries ([`QueueInner::container`] / [`QueueInner::dag_meta`] +
/// the `hive_jobq::Graph` accessors). One shared crate [`Graph`] holds every DAG.
struct QueueInner {
sched: Scheduler<NodeKind, Resource>,
/// Per-node runtime metadata (the build-log id) — mutable after
/// insert, so it can't ride the immutable node payload.
node_rt: HashMap<NodeId, NodeRuntime>,
}
/// The crate scheduler, specialised to this host's node + resource types.
///
/// A **DAG is a single container node** ([`NodeKind::Dag`], `parent = None`)
/// whose subtree is the DAG's work — so the container's `NodeId` is the DAG id,
/// its rolled-up state is the DAG state, and there are no grouping side-tables:
/// membership + meta are graph queries ([`container`] / [`dag_meta`] + the
/// `hive_jobq::Graph` accessors). One shared crate [`Graph`] holds every DAG.
///
/// There is deliberately **no wrapper struct and no per-node side map**. The
/// last map held the `build_logs` row id; that link now lives on the log row
/// itself (`build_logs.node_id`). With nothing else to guard, the mutex holds
/// the scheduler *directly* — which is what lets `hive_jobq` drive the run loop
/// (it takes `&Arc<Mutex<Scheduler<..>>>`, a type a host-side wrapper could not
/// satisfy).
type Sched = Scheduler<NodeKind, Resource>;
/// The queue. Lives on `Coordinator` (one per hive-c0re process); a single
/// scheduler task ([`scheduler::run_worker`]) drives it.
pub struct JobQueue {
inner: Mutex<QueueInner>,
/// The scheduler, held directly rather than behind a host-side wrapper —
/// `hive_jobq`'s run-loop seam takes `&Arc<Mutex<Scheduler<..>>>`, so this
/// *is* the type the crate drives.
sched: Arc<Mutex<Sched>>,
/// Wakes the scheduler when something new arrives or state changed.
pub(crate) notify: Notify,
}
@ -163,8 +136,19 @@ impl Default for JobQueue {
}
}
/// Insert a declared `job` into the shared graph and record its per-node
/// `node_rt`, returning the inserted ids.
/// A node runner's `Result` as the scheduler's [`Outcome`].
///
/// The failure reason + `finished_at` are stamped onto the graph `Node` by the
/// scheduler (the reason rides `Outcome::Failed`); there is no host-side copy,
/// so nothing needs clearing on success.
fn outcome_of(result: Result<(), String>) -> Outcome {
match result {
Ok(()) => Outcome::Done,
Err(e) => Outcome::Failed(truncate_error(&e)),
}
}
/// Insert a declared `job` into the shared graph, returning the inserted ids.
///
/// A node that declared no parent hangs under `group_parent` — the DAG
/// container for a template, the emitting node for a runtime-appended
@ -178,12 +162,11 @@ impl Default for JobQueue {
/// # Errors
/// Propagates a crate graph-insert error (malformed dep/parent / dep-scope).
fn insert_group(
inner: &mut QueueInner,
inner: &mut Sched,
declare: impl FnOnce(&Job),
group_parent: Option<NodeId>,
) -> anyhow::Result<()> {
inner
.sched
.insert_job(group_parent, |b| {
declare(b);
// c0re names no handles: a DAG is addressed by its container node,
@ -204,16 +187,13 @@ impl JobQueue {
u32::try_from(build_slots.max(1)).unwrap_or(u32::MAX),
);
Self {
inner: Mutex::new(QueueInner {
sched: Scheduler::new(Graph::new(), table),
node_rt: HashMap::new(),
}),
sched: Arc::new(Mutex::new(Scheduler::new(Graph::new(), table))),
notify: Notify::new(),
}
}
fn lock(&self) -> std::sync::MutexGuard<'_, QueueInner> {
self.inner.lock().expect("job_queue mutex poisoned")
fn lock(&self) -> std::sync::MutexGuard<'_, Sched> {
self.sched.lock().expect("job_queue mutex poisoned")
}
/// Submit a DAG: insert a [`NodeKind::Dag`] **container node** carrying the
@ -221,7 +201,13 @@ impl JobQueue {
/// roots re-parented to the container). Returns the container's id as the
/// DAG id — its rolled-up state is the DAG state.
///
/// Takes the spec's recipe by generic, not as a boxed [`Declare`]: a spec
/// The container is an ordinary node: it declares no resources, so the
/// scheduler claims it on the next pass, runs its (empty) logic and parks
/// it in `Finishing`, at which point its children become runnable. Nothing
/// here completes it by hand — a node with no work of its own still goes
/// the way every other node goes.
///
/// Takes the spec's recipe by generic, not as a boxed closure: a spec
/// travels from the template that built it directly into this call, so
/// there is nothing to allocate for.
///
@ -231,7 +217,6 @@ impl JobQueue {
pub fn submit<F: FnOnce(&Job)>(&self, spec: DagSpec<F>) -> anyhow::Result<u64> {
let mut inner = self.lock();
let container = inner
.sched
.append(
NodeKind::Dag {
source: spec.source,
@ -242,100 +227,29 @@ impl JobQueue {
None,
)
.map_err(|e| anyhow::anyhow!("job_queue: container insert failed: {e}"))?;
inner.node_rt.insert(container, NodeRuntime::default());
insert_group(&mut inner, spec.declare, Some(container))?;
// Settle the container's own (no-op) logic immediately so it parks in
// `Finishing` and its children become runnable — it never needs claiming
// or executing, and stays out of `claim_ready`. It rolls up terminal when
// its whole subtree settles (that's the DAG-done signal).
inner.sched.complete(container, Outcome::Done);
drop(inner);
self.notify.notify_one();
Ok(container.get())
}
/// Append a whole *subgraph* into a live DAG at runtime — the single
/// in-DAG-growth primitive. The subgraph is inserted as a [`insert_group`]
/// rooted under `dep_on` (the emitting node): the subgraph's own root becomes
/// a *child* of `dep_on`, its steps children of that root, and the group's
/// agent lease is hoisted onto that root. Ordering root→`dep_on` is the parent
/// gate — the children run once `dep_on` reaches `Finishing`. Because the
/// emitting node stays `Finishing` until this appended subtree is terminal and
/// the DAG's terminal node deps on the top root, roll-up keeps the DAG from
/// settling early with no explicit wiring. A no-op if the DAG is gone.
pub fn append_subgraph(&self, dag_id: u64, declare: Declare, dep_on: NodeId) {
let mut inner = self.lock();
if inner.container(dag_id).is_none() {
return;
}
// Insert the subgraph as a group rooted under the emitting node: the
// subgraph's own root becomes a child of `dep_on`, its steps children of
// that root. No terminal-node wiring — roll-up carries terminality: the
// emitter stays `Finishing` until this appended subtree settles, and the
// container node rolls up terminal only once its whole subtree (incl. this
// appended work) has settled, so the DAG hook waits for free.
if let Err(e) = insert_group(&mut inner, declare, Some(dep_on)) {
tracing::error!(
dag = dag_id,
error = %e,
"job_queue: append_subgraph insert failed"
);
return;
}
drop(inner);
self.notify.notify_one();
}
/// Claim every currently-runnable node, acquiring its resources, and mark it
/// `Running`. Delegates readiness + resource acquisition to the crate's
/// settle loop; builds a [`Claim`] per started node from its payload + its
/// DAG container's metadata. The container node itself is claimed like any
/// other (its executor is an instant no-op that lets its subtree start).
pub fn claim_ready(&self) -> Vec<Claim> {
let mut inner = self.lock();
let inner = &mut *inner;
let started = inner.sched.settle();
let mut claims = Vec::with_capacity(started.len());
for id in started {
let Some(node) = inner.sched.graph().node(id) else {
continue;
};
let kind = node.payload.clone();
let agent = node.payload.agent().to_owned();
let Some(container) = inner.sched.graph().root_of(id) else {
continue;
};
claims.push(Claim {
dag_id: container.get(),
node_id: id,
kind,
agent,
});
// `started_at` is stamped on the graph `Node` by the scheduler's
// transition to `Running` — no host-side copy needed.
}
claims
}
/// Mark a claimed node terminal, recording its outcome + (truncated) error.
/// The crate releases the node's build slot immediately and cascades the
/// `AfterOk` failure cancellation + subtree lease release.
/// The scheduler itself, for `hive_jobq`'s run-loop seam
/// (`Scheduler::claim_next`), which takes exactly this type.
///
/// 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, 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
// copy, so there is nothing to clear here.
let outcome = match result {
Ok(()) => Outcome::Done,
Err(e) => Outcome::Failed(truncate_error(&e)),
};
inner.sched.complete(node_id, outcome);
drop(inner);
self.notify.notify_one();
/// Handing out the `Arc` rather than wrapping each crate call keeps the
/// host from growing a parallel API: the run loop uses `hive_jobq`'s
/// functions directly, and this module stays the thin glue it is being
/// reduced to.
pub(crate) fn sched(&self) -> &Arc<Mutex<Sched>> {
&self.sched
}
/// The DAG container id owning `node`, for log lines and the dashboard.
/// Derived from the graph rather than carried alongside the node — the
/// parent axis already knows it.
#[must_use]
pub fn dag_of(&self, node: NodeId) -> Option<u64> {
self.lock().graph().root_of(node).map(NodeId::get)
}
/// Cancel a DAG that hasn't started yet: every work node is still `Pending`,
@ -360,10 +274,10 @@ impl JobQueue {
/// just that branch. Nothing here knows about DAGs.
pub fn cancel(&self, id: u64) -> bool {
let mut inner = self.lock();
let Some(node) = inner.sched.graph().resolve_id(id) else {
let Some(node) = inner.graph().resolve_id(id) else {
return false;
};
if !inner.sched.cancel_node(node) {
if !inner.cancel_node(node) {
return false;
}
drop(inner);
@ -371,33 +285,6 @@ impl JobQueue {
true
}
/// Link a `build_logs` row to a specific `Running` node.
pub fn set_build_log_id(&self, dag_id: u64, node_id: NodeId, log_id: i64) -> bool {
let mut inner = self.lock();
if inner.sched.graph().root_of(node_id).map(NodeId::get) != Some(dag_id)
|| !inner.node_running(node_id)
{
return false;
}
inner.node_rt.entry(node_id).or_default().build_log_id = Some(log_id);
true
}
/// The `build_logs` row id linked to the wire node id `node_id`, if any —
/// the lookup behind the `GET /api/build-log/<node_id>` query endpoint (the
/// client fetches a node's captured build output on demand rather than
/// receiving it inline). Takes the raw wire `u64` (the endpoint's path
/// param); `node_rt` is keyed by the opaque `NodeId`, so this scans for the
/// matching id — the map is small (live + recently-terminal nodes).
#[must_use]
pub fn build_log_id_of(&self, node_id: u64) -> Option<i64> {
self.lock()
.node_rt
.iter()
.find(|(nid, _)| nid.get() == node_id)
.and_then(|(_, rt)| rt.build_log_id)
}
/// The first failed node's error in `dag_id`, if any has failed yet.
///
/// Unlike the roll-up summary this is readable *mid-flight*, which is the
@ -410,12 +297,8 @@ impl JobQueue {
#[must_use]
pub fn first_error(&self, dag_id: u64) -> Option<String> {
let inner = self.lock();
let container = inner.container(dag_id)?;
inner
.sched
.graph()
.first_error(container)
.map(ToOwned::to_owned)
let container = container(&inner, dag_id)?;
inner.graph().first_error(container).map(ToOwned::to_owned)
}
/// `(agent, label, takes_container_down)` for the live transient-pill set,
@ -447,7 +330,6 @@ impl JobQueue {
pub fn running_transients(&self) -> Vec<RunningTransient> {
let inner = self.lock();
inner
.sched
.graph()
.nodes()
.filter(|n| matches!(n.state, State::Running))
@ -476,204 +358,227 @@ impl JobQueue {
#[must_use]
pub fn snapshot(&self) -> Vec<DagView> {
let inner = self.lock();
let mut ids = inner.visible_dags();
let mut ids = visible_dags(&inner);
ids.sort_unstable_by_key(|c| c.get());
ids.into_iter().filter_map(|c| inner.dag_view(c)).collect()
}
/// Number of live (non-terminal) DAGs — tests + diagnostics.
#[cfg(test)]
#[must_use]
pub fn live_count(&self) -> usize {
let inner = self.lock();
inner
.containers()
.into_iter()
.filter(|&c| inner.sched.graph().is_settled(c) == Some(false))
.count()
ids.into_iter()
.filter_map(|c| dag_view(&inner, c))
.collect()
}
}
impl QueueInner {
/// Whether `id` is a `Running` node.
fn node_running(&self, id: NodeId) -> bool {
self.sched
.graph()
.node(id)
.is_some_and(|n| n.state == State::Running)
}
/// The container node of `dag_id` — the `NodeKind::Dag` root whose id equals
/// `dag_id`. `NodeId` is un-fabricable from a raw `u64`, so this is a search.
fn container(&self, dag_id: u64) -> Option<NodeId> {
self.sched.graph().nodes().find_map(|n| {
(n.parent.is_none()
&& n.id.get() == dag_id
&& matches!(n.payload, NodeKind::Dag { .. }))
/// The container node of `dag_id` — the `NodeKind::Dag` root whose id equals
/// `dag_id`. `NodeId` is un-fabricable from a raw `u64`, so this is a search.
fn container(sched: &Sched, dag_id: u64) -> Option<NodeId> {
sched.graph().nodes().find_map(|n| {
(n.parent.is_none() && n.id.get() == dag_id && matches!(n.payload, NodeKind::Dag { .. }))
.then_some(n.id)
})
}
})
}
/// The container's carried domain metadata as an owned read-view. The data
/// lives solely in the [`NodeKind::Dag`] payload — this is a derived read,
/// not a stored side-table.
fn dag_meta(&self, container: NodeId) -> Option<DagMeta> {
let NodeKind::Dag {
source,
reason,
created_at,
} = &self.sched.graph().node(container)?.payload
else {
return None;
/// The container's carried domain metadata as an owned read-view. The data
/// lives solely in the [`NodeKind::Dag`] payload — this is a derived read,
/// not a stored side-table.
fn dag_meta(sched: &Sched, container: NodeId) -> Option<DagMeta> {
let NodeKind::Dag {
source,
reason,
created_at,
} = &sched.graph().node(container)?.payload
else {
return None;
};
Some(DagMeta {
source: *source,
reason: reason.clone(),
created_at: *created_at,
})
}
/// Project a DAG into its wire [`DagView`]: a near-raw view of the
/// container's work nodes, with `Done` nodes excluded. Lifecycle
/// (`state` / `started_at` / `finished_at` / `error`) is read straight
/// off each `hive_jobq::Node`; the client derives the DAG label, roll-up
/// state, and DAG timestamps from the node set. Non-derivable per-node
/// payload (`approval_id`, meta `inputs`) rides the owning node. Returns
/// `None` when every work node is `Done` or `Skipped` — a fully-settled
/// DAG drops out of the snapshot entirely (a `Failed` one lingers until
/// aged out).
fn dag_view(sched: &Sched, container: NodeId) -> Option<DagView> {
let meta = dag_meta(sched, container)?;
let all: Vec<_> = sched.graph().descendants(container).collect();
// DAG-level timestamps are taken over *all* subtree nodes (including the
// `Done` ones excluded from the wire) — the client can't derive them
// from a `Done`-filtered node set, so the host computes them here.
let mut started: Vec<DateTime<Utc>> = Vec::new();
let mut finished: Vec<DateTime<Utc>> = Vec::new();
for node in &all {
if let Some(s) = node.started_at {
started.push(s);
}
if let Some(f) = node.finished_at {
finished.push(f);
}
}
// Decide which nodes ride the wire *before* projecting any of them: a
// `NodeView` costs a `build_logs` lookup, so building one for a node
// that's about to be dropped would be a query per finished step.
let shown = shown_on_wire(&all.iter().map(|n| n.state).collect::<Vec<_>>())?;
let mut nodes = Vec::new();
for node in shown.into_iter().map(|i| all[i]) {
let id = node.id;
let deps: Vec<u64> = node
.deps
.iter()
.filter_map(|d| match d {
Dep::Node { id, .. } => Some(id.get()),
Dep::Resource { .. } => None,
})
.collect();
// Non-derivable per-node payload rides the node that owns it. Every
// deploy phase carries the approval id, but only the subtree root
// projects it onto the wire — hanging the approval link off all of
// them would render the same card once per phase.
let approval_id = match &node.payload {
NodeKind::DeployWindow { approval_id, .. } => Some(*approval_id),
_ => None,
};
Some(DagMeta {
source: *source,
reason: reason.clone(),
created_at: *created_at,
})
let inputs = match &node.payload {
NodeKind::MetaLock { inputs, .. } => inputs.clone(),
_ => Vec::new(),
};
// Looked up from the log row itself (`build_logs.node_id`), not a
// host-side map. One indexed query per node in the snapshot; the
// node set is bounded by `MAX_HISTORY_DAGS` and the store is a
// local sqlite file, so this is cheaper than the lock contention
// a second shared map would reintroduce.
let build_log_id = crate::build_logs::global().and_then(|h| h.id_for_node(id.get()));
// `node.parent` is the structural jobq parent. Top-level nodes
// have `parent == Some(container)` (direct children of the Dag
// container); those become `parent: None` on the wire since the
// container itself is not part of the work-node payload. Sub-nodes
// carry the id of their containing parent work-node.
let parent = node
.parent
.filter(|&p| p != container)
.map(hive_jobq::NodeId::get);
nodes.push(NodeView {
id: id.get(),
agent: node.payload.agent().to_owned(),
kind: node.payload.as_str().to_owned(),
deps,
state: node.state,
started_at: node.started_at,
finished_at: node.finished_at,
error: node.error.clone(),
approval_id,
inputs,
build_log_id,
parent,
});
}
let is_terminal = sched.graph().is_settled(container) == Some(true);
Some(DagView {
id: container.get(),
source: meta.source,
reason: meta.reason.clone(),
created_at: meta.created_at,
started_at: started.into_iter().min(),
finished_at: is_terminal.then(|| finished.into_iter().max()).flatten(),
nodes,
})
}
/// Project a DAG into its wire [`DagView`]: a near-raw view of the
/// container's work nodes, with `Done` nodes excluded. Lifecycle
/// (`state` / `started_at` / `finished_at` / `error`) is read straight
/// off each `hive_jobq::Node`; the client derives the DAG label, roll-up
/// state, and DAG timestamps from the node set. Non-derivable per-node
/// payload (`approval_id`, meta `inputs`) rides the owning node. Returns
/// `None` when every work node is `Done` or `Skipped` — a fully-settled
/// DAG drops out of the snapshot entirely (a `Failed` one lingers until
/// aged out).
fn dag_view(&self, container: NodeId) -> Option<DagView> {
let meta = self.dag_meta(container)?;
let mut nodes = Vec::new();
// Whether anything in this DAG still has an outcome worth showing.
// Kept separate from `nodes` being non-empty: skipped nodes ride the
// wire so the dashboard can mark the branches that weren't taken, but
// they must not by themselves hold a finished DAG in the snapshot.
let mut any_unsettled = false;
// DAG-level timestamps are taken over *all* subtree nodes (including the
// `Done` ones excluded from the wire) — the client can't derive them
// from a `Done`-filtered node set, so the host computes them here.
let mut started: Vec<DateTime<Utc>> = Vec::new();
let mut finished: Vec<DateTime<Utc>> = Vec::new();
for node in self.sched.graph().descendants(container) {
let id = node.id;
if let Some(s) = node.started_at {
started.push(s);
}
if let Some(f) = node.finished_at {
finished.push(f);
}
// `Done` nodes drop off the wire — a finished step isn't
// interesting. `Skipped` ones stay: which branch a run *didn't*
// take is the readable half of an outcome-branched DAG.
if matches!(node.state, State::Done) {
continue;
}
any_unsettled |= !matches!(node.state, State::Skipped);
let deps: Vec<u64> = node
.deps
.iter()
.filter_map(|d| match d {
Dep::Node { id, .. } => Some(id.get()),
Dep::Resource { .. } => None,
})
.collect();
// Non-derivable per-node payload rides the node that owns it. Every
// deploy phase carries the approval id, but only the subtree root
// projects it onto the wire — hanging the approval link off all of
// them would render the same card once per phase.
let approval_id = match &node.payload {
NodeKind::DeployWindow { approval_id, .. } => Some(*approval_id),
_ => None,
};
let inputs = match &node.payload {
NodeKind::MetaLock { inputs, .. } => inputs.clone(),
_ => Vec::new(),
};
let build_log_id = self.node_rt.get(&id).and_then(|r| r.build_log_id);
// `node.parent` is the structural jobq parent. Top-level nodes
// have `parent == Some(container)` (direct children of the Dag
// container); those become `parent: None` on the wire since the
// container itself is not part of the work-node payload. Sub-nodes
// carry the id of their containing parent work-node.
let parent = node
.parent
.filter(|&p| p != container)
.map(hive_jobq::NodeId::get);
nodes.push(NodeView {
id: id.get(),
agent: node.payload.agent().to_owned(),
kind: node.payload.as_str().to_owned(),
deps,
state: node.state,
started_at: node.started_at,
finished_at: node.finished_at,
error: node.error.clone(),
approval_id,
inputs,
build_log_id,
parent,
});
/// Which of a DAG's work nodes ride the wire, by index into `states` — or
/// `None` when the DAG has nothing left worth showing and drops out of the
/// snapshot entirely.
///
/// Two separate decisions, and conflating them pins every completed deploy in
/// the queue view forever:
/// - **`Done` drops off the wire.** A finished step isn't interesting.
/// `Skipped` stays: which branch a run *didn't* take is the readable half of
/// an outcome-branched DAG.
/// - **`Skipped` alone doesn't hold a DAG in the snapshot.** So "the node list
/// is non-empty" and "there's still something here worth showing" are
/// different questions, and only the second one may drop the DAG.
///
/// Takes states rather than projected nodes so the caller can skip the work of
/// projecting what it's about to discard, and so this is testable without a
/// graph — the states it keys on are ones only a run can produce.
fn shown_on_wire(states: &[State]) -> Option<Vec<usize>> {
let worth_showing = states
.iter()
.any(|s| !matches!(s, State::Done | State::Skipped));
if !worth_showing {
return None;
}
Some(
states
.iter()
.enumerate()
.filter(|(_, s)| !matches!(s, State::Done))
.map(|(i, _)| i)
.collect(),
)
}
/// When a DAG's work node finishes on `finished_at` — the max over its
/// subtree (read off the graph `Node`, as unix seconds), for the history
/// cap ordering.
fn dag_finished_at(sched: &Sched, container: NodeId) -> i64 {
sched
.graph()
.descendants(container)
.filter_map(|n| n.finished_at)
.map(|t| t.timestamp())
.max()
.unwrap_or(0)
}
/// Every DAG container node id in the graph.
fn containers(sched: &Sched) -> Vec<NodeId> {
sched
.graph()
.nodes()
.filter(|n| n.parent.is_none() && matches!(n.payload, NodeKind::Dag { .. }))
.map(|n| n.id)
.collect()
}
/// The **visible** DAG set for the snapshot: every live (non-terminal) DAG,
/// plus the newest [`MAX_HISTORY_DAGS`] terminal ones. Crate nodes for
/// evicted DAGs linger in the graph (bounded-prune is a Stage-C follow-up);
/// this filter is what bounds what the dashboard sees.
fn visible_dags(sched: &Sched) -> Vec<NodeId> {
let mut live: Vec<NodeId> = Vec::new();
let mut terminal: Vec<(NodeId, i64, u64)> = Vec::new();
for c in containers(sched) {
if sched.graph().is_settled(c) == Some(true) {
terminal.push((c, dag_finished_at(sched, c), c.get()));
} else {
live.push(c);
}
if !any_unsettled {
return None;
}
let is_terminal = self.sched.graph().is_settled(container) == Some(true);
Some(DagView {
id: container.get(),
source: meta.source,
reason: meta.reason.clone(),
created_at: meta.created_at,
started_at: started.into_iter().min(),
finished_at: is_terminal.then(|| finished.into_iter().max()).flatten(),
nodes,
})
}
retain_history(live, terminal, MAX_HISTORY_DAGS)
}
/// When a DAG's work node finishes on `finished_at` — the max over its
/// subtree (read off the graph `Node`, as unix seconds), for the history
/// cap ordering.
fn dag_finished_at(&self, container: NodeId) -> i64 {
self.sched
.graph()
.descendants(container)
.filter_map(|n| n.finished_at)
.map(|t| t.timestamp())
.max()
.unwrap_or(0)
}
/// Every DAG container node id in the graph.
fn containers(&self) -> Vec<NodeId> {
self.sched
.graph()
.nodes()
.filter(|n| n.parent.is_none() && matches!(n.payload, NodeKind::Dag { .. }))
.map(|n| n.id)
.collect()
}
/// The **visible** DAG set for the snapshot: every live (non-terminal) DAG,
/// plus the newest [`MAX_HISTORY_DAGS`] terminal ones. Crate nodes for
/// evicted DAGs linger in the graph (bounded-prune is a Stage-C follow-up);
/// this filter is what bounds what the dashboard sees.
fn visible_dags(&self) -> Vec<NodeId> {
let mut live: Vec<NodeId> = Vec::new();
let mut terminal: Vec<(NodeId, i64)> = Vec::new();
for c in self.containers() {
if self.sched.graph().is_settled(c) == Some(true) {
terminal.push((c, self.dag_finished_at(c)));
} else {
live.push(c);
}
}
// Newest first, so truncating to the cap keeps the most recent.
terminal.sort_by(|a, b| b.1.cmp(&a.1).then(b.0.get().cmp(&a.0.get())));
terminal.truncate(MAX_HISTORY_DAGS);
let mut kept = live;
kept.extend(terminal.into_iter().map(|(c, _)| c));
kept
}
/// [`visible_dags`]'s policy, split from the graph it reads: keep every live
/// DAG, plus the newest `cap` terminal ones.
///
/// `terminal` rows are `(handle, finished_at, tiebreak)`. The tiebreak orders
/// DAGs that settled inside the same wall-clock second — which is *most* of
/// them under a burst, and all of them in a test, so it is load-bearing rather
/// than a formality.
///
/// Generic over the handle purely so this is reachable without a graph: a
/// `NodeId` cannot be fabricated, so a test that had to pass real ones could
/// only get them by submitting and running DAGs.
fn retain_history<T>(live: Vec<T>, mut terminal: Vec<(T, i64, u64)>, cap: usize) -> Vec<T> {
// Newest first, so truncating to the cap keeps the most recent.
terminal.sort_by(|a, b| b.1.cmp(&a.1).then(b.2.cmp(&a.2)));
terminal.truncate(cap);
let mut kept = live;
kept.extend(terminal.into_iter().map(|(handle, _, _)| handle));
kept
}
/// Truncate a node error to [`MAX_ERROR_LEN`] on a char boundary, appending `…`.

6
hive-c0re/src/job_queue/model.rs
View file

@ -407,9 +407,9 @@ impl NodeKind {
/// Generic over the recipe rather than boxing it: a spec goes from the template
/// that returns it straight to the `submit` that consumes it, so the closure's
/// concrete type is known the whole way and needs neither an allocation nor a
/// `Send` bound. (The executor's `append_subgraph` is the case that *does* need
/// a boxed [`super::Declare`] — its recipes are collected into a `Vec` and
/// applied later, across a task boundary.)
/// `Send` bound. Nothing boxes a recipe any more — a running node grows its DAG
/// by declaring straight onto the builder it was handed, so there is no recipe
/// to store and replay across a task boundary.
pub struct DagSpec<F> {
pub source: Source,
/// Free-form "why".

138
hive-c0re/src/job_queue/scheduler.rs
View file

@ -16,21 +16,18 @@
//! the DAG settles.
//!
//! In-DAG growth (a `MetaLock` growing rebuild subgraphs, a `Reconcile` fanning
//! its `Start`/`Stop`) flows through `NodeOutput.append_subgraph`, applied
//! before the emitting node completes — see `handle_completion`.
//! its `Start`/`Stop`) is declared onto the builder each node is handed, and
//! inserted as part of completing that node. Completion itself is not this
//! module's job any more: it happens *inside* the future
//! [`hive_jobq::scheduler::Scheduler::claim_next`] hands back, so a node that
//! ran but was never completed is not an expressible state here.
use std::collections::HashMap;
use std::sync::Arc;
use super::Claim;
use super::exec::{self, NodeOutput};
use super::exec;
use crate::coordinator::Coordinator;
struct NodeDone {
claim: Claim,
result: anyhow::Result<NodeOutput>,
}
/// Scheduler loop. Spawned once at hive-c0re startup from `main.rs`.
///
/// Shutdown semantics: subscribes to `coord.shutdown_rx()`. On a true signal
@ -53,7 +50,6 @@ struct NodeDone {
/// reconverging silently.
pub async fn run_worker(coord: Arc<Coordinator>) {
let mut shutdown = coord.shutdown_rx();
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<NodeDone>();
// Last derived pill set we published, keyed by agent (its lease is cap-1,
// so one pill each). Purely the previous value of a *derived* quantity —
// it exists to spot transitions, since the dashboard wants edges
@ -70,24 +66,66 @@ pub async fn run_worker(coord: Arc<Coordinator>) {
return;
}
reconcile_transients(&coord, &mut transients);
let claims = coord.job_queue.claim_ready();
if !claims.is_empty() {
for claim in claims {
tracing::info!(
dag = claim.dag_id,
node = claim.node_id.get(),
kind = claim.kind.as_str(),
agent = %claim.agent,
"job_queue: node running"
);
let coord = Arc::clone(&coord);
let tx = tx.clone();
tokio::spawn(async move {
let result = exec::run_node(&coord, &claim).await;
// Send failure = scheduler gone (shutdown); drop.
let _ = tx.send(NodeDone { claim, result });
});
}
// Claim exactly one node and get back the work that runs it. `Some`
// means something started, so there may be more runnable right now —
// loop again immediately. `None` means nothing is runnable and the
// loop parks below. That decision is the whole reason the crate hands
// back a task rather than an id.
let runner = {
// Two handles, deliberately: `sched` is the scheduler the crate
// locks, `node_coord` is what the node's own future captures. One
// binding can't do both — passing `coord.job_queue.sched()` borrows
// `coord` for the whole call while the `move` closure wants to take
// it.
let sched = Arc::clone(coord.job_queue.sched());
let node_coord = Arc::clone(&coord);
hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, job| {
let coord = node_coord;
async move {
tracing::info!(
dag = coord.job_queue.dag_of(id).unwrap_or_default(),
node = id.get(),
kind = kind.as_str(),
agent = %kind.agent(),
"job_queue: node running"
);
let (grown, result) = exec::run_node(&coord, job, id, &kind).await;
match &result {
Ok(()) => tracing::info!(node = id.get(), "job_queue: node done"),
Err(e) => tracing::warn!(
node = id.get(),
kind = kind.as_str(),
agent = %kind.agent(),
error = %format!("{e:#}"),
grown_nodes = !grown.is_empty(),
"job_queue: node failed"
),
}
// Growth on a failed node is dropped by `complete_growing`,
// not here: failure cancel-cascades inside jobq, so that
// rule is the crate's to enforce and this loop does not get
// to forget it.
(
grown,
super::outcome_of(result.map_err(|e| format!("{e:#}"))),
)
}
})
};
if let Some(runner) = runner {
let done_coord = Arc::clone(&coord);
tokio::spawn(async move {
// Completion happens inside `runner` — it cannot be forgotten
// here, which is why there is no completion channel any more.
let (id, grew) = runner.await;
if let Err(e) = grew {
tracing::warn!(node = id.get(), error = %e, "job_queue: grown job rejected");
}
done_coord.emit_rebuild_queue_snapshot();
// Wake the loop: this node's completion may have unblocked
// dependents. Previously the completion channel did this.
done_coord.job_queue.notify.notify_one();
});
// Newly-started owner nodes now hold their leases — surface the pills.
reconcile_transients(&coord, &mut transients);
coord.emit_rebuild_queue_snapshot();
@ -101,55 +139,11 @@ pub async fn run_worker(coord: Arc<Coordinator>) {
return;
}
}
Some(done) = rx.recv() => {
handle_completion(&coord, done);
}
() = coord.job_queue.notify.notified() => {}
}
}
}
fn handle_completion(coord: &Arc<Coordinator>, done: NodeDone) {
let NodeDone { claim, result } = done;
match result {
Ok(output) => {
tracing::info!(
dag = claim.dag_id,
node = claim.node_id.get(),
"job_queue: node done"
);
// Append any in-DAG subgraphs BEFORE completing this node, so
// completing it doesn't roll the DAG terminal while the appended
// work is still pending. Each subgraph roots on this node
// (`AfterOk`), so it becomes ready the instant this one settles
// `Done` just below — covers both the multi-node case (a `MetaLock`
// growing per-agent rebuild subgraphs) and the single-node case (a
// `Reconcile` planner's `Start` / `Stop`).
for subgraph in output.append_subgraph {
coord
.job_queue
.append_subgraph(claim.dag_id, subgraph, claim.node_id);
}
coord.job_queue.complete_node(claim.node_id, Ok(()));
}
Err(e) => {
let msg = format!("{e:#}");
tracing::warn!(
dag = claim.dag_id,
node = claim.node_id.get(),
kind = claim.kind.as_str(),
agent = %claim.agent,
error = %msg,
"job_queue: node failed"
);
coord.job_queue.complete_node(claim.node_id, Err(msg));
}
}
// The next loop iteration re-reconciles the transient pills against the
// post-completion lease state (a settled subgraph drops its pill).
coord.emit_rebuild_queue_snapshot();
}
/// Publish the transitions between the previously-derived pill set and the
/// current one. `prev` is last loop's derived value, keyed by agent (an agent's
/// lease is cap-1, so at most one pill each).

111
hive-c0re/src/job_queue/templates.rs
View file

@ -28,7 +28,7 @@ use hive_jobq::TerminalState;
use super::model::{DagSpec, NodeKind, PermPayload, Source};
use super::resource::Resource;
use super::{Declare, Handle, Job};
use super::{Handle, Job};
/// The `Rebuilt`-reporting tail pair for a rebuild-shaped DAG: the success node
/// gated on every group-root in `roots`, and the failure node gated on *its*
@ -80,6 +80,39 @@ fn resolve_approval_tails(b: &Job, approval_id: i64, root: Handle<'_>) {
}
}
/// Declare one rebuild subgraph per agent onto the builder a running
/// [`NodeKind::MetaLock`] was handed.
///
/// **Into the emitter's own builder, not as new DAGs.** Growing in-DAG is what
/// roots each subgraph on the `MetaLock`, so the whole sweep (or meta-update
/// cascade) stays one unit of work the operator can watch and cancel, and every
/// rebuild builds against the lock the emitter just bumped.
///
/// Same reason as [`fanned_out_mechanical`] for living here: this was the
/// second construction site declaring nodes inline in an executor.
pub(crate) fn grown_rebuilds(b: &Job, agents: &[String], opts: RebuildOpts) {
for agent in agents {
rebuild_nodes(b, agent, opts, None);
}
}
/// Declare the mechanical node a [`NodeKind::Reconcile`] planner fans out
/// (`Start` / `Stop`) onto the builder it was handed while running.
///
/// `Start` / `Stop` declare the agent lease they run under. Their `Reconcile`
/// parent is holding it already, so the declaration is a **re-entrant borrow**
/// — no second unit, no deadlock. It exists so the requirement belongs to the
/// node rather than to the fact that a `Reconcile` happens to fan it out.
///
/// Lives here rather than inline in `exec.rs` for the same reason every other
/// declaration does: this is the one construction site that was hiding in an
/// executor, which meant the only test of it had to re-declare the same two
/// calls itself and would have kept passing if the executor changed.
pub(crate) fn fanned_out_mechanical(b: &Job, kind: NodeKind) {
let lease = Resource::Agent(kind.agent().to_owned());
let _ = b.node(kind).needs(lease);
}
/// Knobs for [`rebuild_nodes`]. A struct rather than two positional `bool`s so
/// a call site cannot silently swap them.
#[derive(Debug, Clone, Copy)]
@ -235,32 +268,29 @@ pub(crate) fn rebuild_nodes<'a>(
/// finalized. `Reconcile` alone would not do, being `AfterAny` — it reaches
/// `Done` even after a failed `Swap`.
///
/// Appended, not submitted: the roots below become children of the emitting
/// `DeployApply` (see [`super::JobQueue::append_subgraph`]), which puts them
/// inside the `DeployWindow`'s subtree — so the `MetaWindow` this subgraph's
/// `MetaSync` and `FinalizeDeploy` declare is re-entered from the ancestor
/// already holding it rather than deadlocking against it.
pub(crate) fn deploy_rebuild_nodes(agent: &str, approval_id: i64) -> Declare {
let agent = agent.to_owned();
Box::new(move |b: &Job| {
let roots = rebuild_nodes(
b,
&agent,
RebuildOpts {
relock: false,
graceful: false,
},
None,
);
let _finalize = b
.node(NodeKind::FinalizeDeploy {
agent: agent.clone(),
approval_id,
})
.needs(Resource::MetaWindow)
.after_ok(roots.prebuild)
.after_ok(roots.reconcile);
})
/// Declared into a **running** `DeployApply`'s own builder, not submitted: the
/// roots below become children of that node, which puts them inside the
/// `DeployWindow`'s subtree — so the `MetaWindow` this subgraph's `MetaSync`
/// and `FinalizeDeploy` declare is re-entered from the ancestor already holding
/// it rather than deadlocking against it.
pub(crate) fn deploy_rebuild_nodes(b: &Job, agent: &str, approval_id: i64) {
let roots = rebuild_nodes(
b,
agent,
RebuildOpts {
relock: false,
graceful: false,
},
None,
);
let _finalize = b
.node(NodeKind::FinalizeDeploy {
agent: agent.to_owned(),
approval_id,
})
.needs(Resource::MetaWindow)
.after_ok(roots.prebuild)
.after_ok(roots.reconcile);
}
/// One uniform rebuild shape — no `was_running` branch. `StopForUpdate`
@ -375,29 +405,6 @@ pub fn approval_deploy(
}
}
/// A single `Reconcile` node that converges observed power state to the
/// persisted intent — `wanted` is untouched (no `SetWanted`), unlike the
/// operator `start`/`stop` templates. Test-only helper now (used to build
/// single-node lifecycle DAGs that exercise per-agent lease serialization
/// in the queue tests); production paths no longer emit a bare reconcile.
#[cfg(test)]
pub fn reconcile_only(
agent: &str,
source: Source,
reason: String,
) -> DagSpec<impl FnOnce(&Job) + use<>> {
let agent = agent.to_owned();
DagSpec {
source,
reason,
declare: Box::new(move |b: &Job| {
// Name the lease before the agent string is moved into the kind.
let lease = Resource::Agent(agent.clone());
let _reconcile = b.node(NodeKind::Reconcile { agent }).needs(lease);
}),
}
}
/// First-deploy spawn (approval-driven): `Provision` (proposed/applied
/// repos, state subvolume, meta registration) then `Create`
/// (`nixos-container create`), drop-in write, then `Reconcile` starts
@ -480,8 +487,8 @@ pub fn perm_change(
}
/// Meta-input lock bump. The `MetaLock` executor grows one rebuild subgraph
/// per affected agent into *this same* DAG on completion (via
/// `append_subgraph`) — appended *after* the bump lands so their prebuilds
/// per affected agent into *this same* DAG on completion (declared onto the
/// builder it was handed) — appended *after* the bump lands so their prebuilds
/// run against the post-bump lock, and a failed bump appends nothing
/// (replacing the old fan-out-child-DAGs dance).
/// `transient = Rebuilding` because those appended subgraphs are rebuilds:

2078
hive-c0re/src/job_queue/tests.rs
View file

File diff suppressed because it is too large Load diff

49
hive-c0re/src/lifecycle/mod.rs
View file

@ -274,10 +274,10 @@ pub async fn swap_update(
name: &str,
hive: &HiveEnv,
paths: &AgentPaths,
on_build_log_id: &(dyn Fn(i64) + Send + Sync),
node_id: Option<u64>,
) -> Result<()> {
write_dropins(name, hive, paths).await?;
priv_run_inner("update", name, Some(on_build_log_id)).await
priv_run_inner("update", name, node_id).await
}
/// Build the `AgentSpec` list for the meta flake from `nixos-container
@ -582,14 +582,10 @@ pub async fn destroy(name: &str) -> Result<()> {
/// the prebuild happens before stop, and `docs/coordinator.md::Prebuild
/// attr path` for why the explicit nixosConfigurations attr is required.
///
/// `on_build_log_id` fires with the `build_logs` row id as soon as the
/// row opens, so queue-side callers can link their node to the live
/// stream. Pass `&|_| ()` when not needed.
pub async fn prebuild_toplevel(
name: &str,
flake_ref: &str,
on_build_log_id: &(dyn Fn(i64) + Send + Sync),
) -> Result<()> {
/// `node_id` is the queue node this build belongs to, when there is one —
/// it is stored on the `build_logs` row so the dashboard can find the log
/// from the node. Pass `None` for builds that run outside the queue.
pub async fn prebuild_toplevel(name: &str, flake_ref: &str, node_id: Option<u64>) -> Result<()> {
use tokio::io::{AsyncBufReadExt, BufReader};
// Split `<root>#<name>` so we can re-emit with the explicit
// `nixosConfigurations.<name>` segment. The flake_ref shape is
@ -624,15 +620,12 @@ pub async fn prebuild_toplevel(
// into the row; `finish` lands the terminal status before we bail.
let logs = crate::build_logs::global();
let log_id = logs.as_ref().and_then(|h| {
h.start(name, "prebuild", &cmdline)
h.start(name, "prebuild", &cmdline, node_id)
.map_err(|e| {
tracing::warn!(error = ?e, "build_logs: start failed (prebuild log dropped)");
})
.ok()
});
if let Some(id) = log_id {
on_build_log_id(id);
}
let mut child = Command::new("nix")
.args(&args)
@ -784,37 +777,25 @@ async fn priv_run(kind: &str, name: &str) -> Result<()> {
priv_run_inner(kind, name, None).await
}
/// Like `priv_run` but calls `on_log_id(log_id)` immediately after the
/// build-log row is opened — before the actual container op starts.
/// This lets callers surface the row id for live streaming (e.g. the
/// rebuild-queue worker sets `build_log_id` on the queue entry so the
/// dashboard can link to `/api/build-logs/id/{id}/stream`).
/// Like `priv_run` but stamps `node_id` onto the build-log row it opens, so
/// the dashboard can find the log from the queue node (and link to
/// `/api/build-logs/id/{id}/stream`).
///
/// The callback fires only when a build-log row is successfully opened
/// (i.e. the global `BuildLogs` handle is installed AND `h.start()`
/// succeeds). No-op when `on_log_id` is `None` — that's the path for
/// all callers that don't need the id.
async fn priv_run_inner(
kind: &str,
name: &str,
on_log_id: Option<&(dyn Fn(i64) + Send + Sync)>,
) -> Result<()> {
/// This used to be a `Fn(i64)` callback that handed the row id *back* to the
/// queue, which then held it in a side map. The row carries the link itself
/// now, so the id only ever travels one way.
async fn priv_run_inner(kind: &str, name: &str, node_id: Option<u64>) -> Result<()> {
let container = container_name(name);
let cmdline = format!("nixos-container {kind} {container}");
let logs = crate::build_logs::global();
let log_id = logs.as_ref().and_then(|h| {
h.start(name, kind, &cmdline)
h.start(name, kind, &cmdline, node_id)
.map_err(|e| {
tracing::warn!(error = ?e, "build_logs: start failed (priv_run log dropped)");
})
.ok()
});
// Notify the caller as soon as the log row exists so it can surface
// the id for live streaming before the container op even starts.
if let (Some(id), Some(cb)) = (log_id, on_log_id) {
cb(id);
}
// For long-running ops use the streaming protocol so build_logs
// receives lines in real time rather than as a batch at completion.

7
hive-c0re/src/meta.rs
View file

@ -1590,7 +1590,12 @@ async fn nix_logged(dir: &Path, args: &[&str], agent: &str, kind: &str) -> Resul
let cmdline = format!("nix {}", nix_argv(args).join(" "));
let logs = crate::build_logs::global();
let log_id = logs.as_ref().and_then(|h| {
h.start(agent, kind, &cmdline)
// No node id: `nix_logged`'s two callers are meta-flake operations
// reached from outside the queue as well as from inside it, and the
// agent+kind+time listing is how they're surfaced today. Linking them
// to a node would mean threading the id through `meta`'s public API
// for no current reader — worth doing when something wants it.
h.start(agent, kind, &cmdline, None)
.map_err(|e| {
tracing::warn!(error = ?e, %kind, "build_logs: start failed (meta log dropped)");
})

107
hive-c0re/src/stores/build_logs.rs
View file

@ -13,6 +13,8 @@ use serde::Serialize;
use tokio::sync::broadcast;
use utoipa::ToSchema;
use crate::db::Migration;
/// Process-singleton handle, set once at coordinator startup. Lets
/// the `lifecycle` module's `run` / `prebuild_toplevel` access the
/// writer without threading an `Arc<BuildLogs>` through every
@ -65,6 +67,27 @@ CREATE INDEX IF NOT EXISTS idx_build_logs_status_finished
WHERE finished_at IS NOT NULL;
";
/// Ordered schema migrations tracked in `schema_versions` (key `"build_logs"`).
///
/// v1 makes the log row carry its node, replacing the host-side
/// `NodeId -> build_log_id` map the job queue used to hold. The link has a
/// single home again, and the direction is the one the type system allows:
/// a `hive_jobq` node payload is immutable after insert, but the log row is
/// written when the build starts and can name the node it belongs to.
///
/// Legacy rows keep `node_id IS NULL` — they predate the column and no node
/// still exists to link them to, so the dashboard's by-node lookup simply
/// misses them (the by-agent listing, which is how they're reached, is
/// unaffected).
const MIGRATIONS: &[Migration] = &[Migration {
sql: "BEGIN;
ALTER TABLE build_logs ADD COLUMN node_id INTEGER;
CREATE INDEX IF NOT EXISTS idx_build_logs_node
ON build_logs (node_id) WHERE node_id IS NOT NULL;
COMMIT;",
adds_column: Some(("build_logs", "node_id")),
}];
/// Status of a finished build attempt. Stored as the literal string in
/// the `status` column; `NULL` while the attempt is still in progress.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
@ -152,6 +175,7 @@ impl BuildLogs {
let conn = crate::db::open(&path, "build_logs")?;
conn.execute_batch(SCHEMA)
.context("apply build_logs schema")?;
crate::db::apply_versioned_migrations(&conn, "build_logs", MIGRATIONS)?;
let (notify_tx, _) = broadcast::channel(NOTIFY_CAP);
Ok(Self {
conn: Mutex::new(conn),
@ -169,17 +193,49 @@ impl BuildLogs {
/// Open a row for a new build attempt. Returns the assigned id
/// — the caller threads it through `append_stdout` / `append_stderr`
/// while the child runs and into `finish` once it exits.
pub fn start(&self, agent: &str, kind: &str, cmdline: &str) -> Result<i64> {
///
/// `node_id` is the queue node this build belongs to, when there is one.
/// It is `None` for builds that run outside the job queue; those are
/// reachable by agent + time, just not by node.
pub fn start(
&self,
agent: &str,
kind: &str,
cmdline: &str,
node_id: Option<u64>,
) -> Result<i64> {
let now = Utc::now().timestamp();
let node_id = node_id.and_then(|n| i64::try_from(n).ok());
let conn = self.conn.lock().unwrap();
conn.execute(
"INSERT INTO build_logs (agent, kind, cmdline, started_at) VALUES (?1, ?2, ?3, ?4)",
params![agent, kind, cmdline, now],
"INSERT INTO build_logs (agent, kind, cmdline, started_at, node_id) \
VALUES (?1, ?2, ?3, ?4, ?5)",
params![agent, kind, cmdline, now, node_id],
)
.context("insert build_logs row")?;
Ok(conn.last_insert_rowid())
}
/// The most recent build-log row for `node_id`, if any. Replaces the job
/// queue's in-memory `NodeId -> build_log_id` side map: the link lives in
/// the row itself now, so it survives a restart and needs no lock held
/// alongside the scheduler's.
///
/// `MAX(id)` rather than a uniqueness assumption — a node that is retried
/// opens a second row, and the newest is the one the panel should show.
#[must_use]
pub fn id_for_node(&self, node_id: u64) -> Option<i64> {
let node_id = i64::try_from(node_id).ok()?;
let conn = self.conn.lock().unwrap();
conn.query_row(
"SELECT MAX(id) FROM build_logs WHERE node_id = ?1",
params![node_id],
|row| row.get::<_, Option<i64>>(0),
)
.ok()
.flatten()
}
/// Append a single stdout line. Best-effort: errors are logged
/// but never returned to the caller, so a transient sqlite blip
/// never tears down a rebuild's stdout pump.
@ -453,7 +509,7 @@ mod tests {
fn start_appends_finish_flow() {
let (_d, db) = tmpdb();
let id = db
.start("alice", "prebuild", "nix build foo")
.start("alice", "prebuild", "nix build foo", None)
.expect("start");
db.append_stdout(id, "building '/nix/store/abc.drv'");
db.append_stderr(id, "error: line 12");
@ -482,9 +538,9 @@ mod tests {
// assert id-ordering (autoincrement) is the tiebreaker — list
// sorts by started_at DESC but the ORDER BY still produces the
// last-inserted row first when timestamps match.
let id_a1 = db.start("alice", "run", "cmd one").expect("start");
let _id_b = db.start("bob", "run", "cmd two").expect("start");
let id_a2 = db.start("alice", "run", "cmd three").expect("start");
let id_a1 = db.start("alice", "run", "cmd one", None).expect("start");
let _id_b = db.start("bob", "run", "cmd two", None).expect("start");
let id_a2 = db.start("alice", "run", "cmd three", None).expect("start");
db.finish(id_a1, BuildStatus::Ok);
let alice_rows = db.list_recent_for_agent("alice", 10).expect("list");
@ -515,10 +571,12 @@ mod tests {
#[test]
fn vacuum_drops_old_finished_only_per_status() {
let (_d, db) = tmpdb();
let id_fresh_fail = db.start("alice", "run", "fresh fail").expect("start");
let id_old_fail = db.start("alice", "run", "old fail").expect("start");
let id_old_ok = db.start("alice", "run", "old ok").expect("start");
let id_running = db.start("alice", "run", "still running").expect("start");
let id_fresh_fail = db.start("alice", "run", "fresh fail", None).expect("start");
let id_old_fail = db.start("alice", "run", "old fail", None).expect("start");
let id_old_ok = db.start("alice", "run", "old ok", None).expect("start");
let id_running = db
.start("alice", "run", "still running", None)
.expect("start");
db.finish(id_fresh_fail, BuildStatus::Fail);
db.finish(id_old_fail, BuildStatus::Fail);
db.finish(id_old_ok, BuildStatus::Ok);
@ -550,6 +608,31 @@ mod tests {
assert!(db.get_full(id_running).unwrap().is_some());
}
#[test]
fn node_link_survives_completion_and_newest_wins() {
// The queue used to hold this link in an in-memory side map, which
// meant it died with the process and needed the queue lock to read.
// On the row it outlives both the node's completion and a restart.
let (_d, db) = tmpdb();
let first = db.start("alice", "swap", "cmd", Some(7)).expect("start");
db.finish(first, BuildStatus::Fail);
assert_eq!(
db.id_for_node(7),
Some(first),
"link survives the build finishing"
);
// A retried node opens a second row; the panel wants the current
// attempt, not the first one.
let retry = db.start("alice", "swap", "cmd", Some(7)).expect("start");
assert_eq!(db.id_for_node(7), Some(retry), "newest attempt wins");
// Builds that run outside the queue carry no node and are found by
// agent + time instead — they must not collide with node lookups.
db.start("alice", "run", "no node", None).expect("start");
assert_eq!(db.id_for_node(999), None, "unknown node → no row");
}
#[test]
fn append_after_finish_still_appends() {
// Defensive: if a child's stdout pump fires one last line
@ -557,7 +640,7 @@ mod tests {
// append should land on the row (status already set, but the
// log stays consistent with what happened).
let (_d, db) = tmpdb();
let id = db.start("alice", "run", "cmd").expect("start");
let id = db.start("alice", "run", "cmd", None).expect("start");
db.finish(id, BuildStatus::Ok);
db.append_stdout(id, "post-finish trailing line");
let full = db.get_full(id).expect("get").expect("Some");

3
hive-c0re/src/workers/auto_update.rs
View file

@ -391,8 +391,7 @@ fn submit_boot_tree(
n_skipped,
);
let declare: crate::job_queue::Declare =
Box::new(move |b| boot_nodes(b, any_stale, fanout, drifted));
let declare = move |b: &crate::job_queue::Job| boot_nodes(b, any_stale, fanout, drifted);
let spec = DagSpec {
// The sweep's own rebuild subgraphs emit their `Rebuilt` events as they

103
hive-host-sock/src/jobs.rs
View file

@ -206,3 +206,106 @@ impl DagView {
}
}
}
#[cfg(test)]
mod tests {
use chrono::Utc;
use super::{DagView, NodeView, Source, State};
/// A node set carrying nothing but the states — the only input
/// `rollup_state` reads.
fn dag(states: &[State]) -> DagView {
DagView {
id: 1,
source: Source::Manual,
reason: "test".to_owned(),
created_at: Utc::now(),
started_at: None,
finished_at: None,
nodes: states
.iter()
.enumerate()
.map(|(i, &state)| NodeView {
id: i as u64,
agent: "a".to_owned(),
kind: "reconcile".to_owned(),
deps: Vec::new(),
state,
started_at: None,
finished_at: None,
error: None,
approval_id: None,
inputs: Vec::new(),
build_log_id: None,
parent: None,
})
.collect(),
}
}
#[test]
fn a_failure_outranks_everything_and_skipped_counts_for_nothing() {
// The case this replaces used to be arranged in hive-c0re by running a
// rebuild until its Prebuild failed. Only the states ever mattered.
assert_eq!(
dag(&[State::Done, State::Failed, State::Skipped]).rollup_state(),
State::Failed
);
// A failure wins even against a node still going — the DAG's verdict
// is already decided.
assert_eq!(
dag(&[State::Running, State::Failed]).rollup_state(),
State::Failed
);
// Skipped is an expected part of a healthy run: an outcome-branched
// DAG always leaves one branch untaken, so counting it would make
// every successful DAG roll up non-Done.
assert_eq!(
dag(&[State::Done, State::Skipped]).rollup_state(),
State::Done
);
}
#[test]
fn cancelled_outranks_running_and_pending() {
// A cancelled DAG still has its weak-edged tail node to run, so
// Pending-then-Running would flicker back at the operator who just
// cancelled it and read as "the cancel didn't take".
assert_eq!(
dag(&[State::Cancelled, State::Pending]).rollup_state(),
State::Cancelled
);
assert_eq!(
dag(&[State::Cancelled, State::Running]).rollup_state(),
State::Cancelled
);
}
#[test]
fn finishing_still_counts_as_running() {
// The node's own work is done but its sub-nodes are still going, so
// the DAG is in flight. A parent parked in Finishing is the normal
// shape of a subtree mid-run, not an edge case.
assert_eq!(
dag(&[State::Finishing, State::Pending]).rollup_state(),
State::Running
);
assert_eq!(
dag(&[State::Running, State::Pending]).rollup_state(),
State::Running
);
assert_eq!(
dag(&[State::Done, State::Pending]).rollup_state(),
State::Pending
);
}
#[test]
fn an_empty_node_set_reads_done() {
// Every node Done means every node is filtered off the wire, so this
// is what a finished DAG actually looks like to a consumer that has
// one in hand at all.
assert_eq!(dag(&[]).rollup_state(), State::Done);
}
}

27
hive-jobq/src/builder.rs
View file

@ -8,8 +8,10 @@
//! **An insertion API, not a spec factory.** A builder is only ever handed to a
//! closure by the single insertion entry point
//! ([`crate::scheduler::Scheduler::insert_job`]), which inserts the declared
//! nodes and returns the ids the job asked for. It cannot be constructed, held
//! or inserted from outside this crate, and there is no intermediate
//! nodes and returns the ids the job asked for. It cannot be constructed or
//! inserted from outside this crate — `new()` and `insert_with` are both
//! `pub(crate)`, and there is deliberately no `Default` impl, since a trait impl
//! on a `pub` type is public regardless. There is no intermediate
//! node-description type to keep in sync with [`crate::Graph::insert`]'s signature —
//! so a job has no representation that can be passed around instead of being
//! inserted.
@ -242,16 +244,6 @@ pub struct JobBuilder<N, R> {
nodes: RefCell<Vec<Pending<N, R>>>,
}
// Hand-written rather than derived: `#[derive(Default)]` would demand
// `N: Default, R: Default`, which has nothing to do with an empty builder.
impl<N, R> Default for JobBuilder<N, R> {
fn default() -> Self {
Self {
nodes: RefCell::new(Vec::new()),
}
}
}
impl<N, R> JobBuilder<N, R> {
/// A fresh, empty builder.
///
@ -261,8 +253,17 @@ impl<N, R> JobBuilder<N, R> {
/// nodes and returns the ids. Nothing job-shaped is constructible or
/// carryable outside this crate — otherwise it is a spec factory again,
/// just with a builder's name on it.
///
/// Deliberately **not** a `Default` impl. A trait impl on a `pub` type is
/// public no matter how private its inherent constructors are, so
/// `JobBuilder::default()` would hand every downstream crate the builder
/// this fn is `pub(crate)` to withhold. The body is what `#[derive(Default)]`
/// could not be anyway — deriving would demand `N: Default, R: Default`,
/// which has nothing to do with an empty builder.
pub(crate) fn new() -> Self {
Self::default()
Self {
nodes: RefCell::new(Vec::new()),
}
}
/// Whether nothing has been declared yet — for a caller deciding whether an

374
hive-jobq/src/scheduler.rs
View file

@ -1,13 +1,14 @@
//! The settle loop — drives a [`Graph`] to completion over a resource pool the
//! scheduler owns directly.
//!
//! [`Scheduler::settle`] claims every currently-runnable pending node (its
//! [`Scheduler::claim_next`] claims one currently-runnable pending node (its
//! [`Dep::Node`] edges satisfied *and* all its [`Dep::Resource`] units acquired
//! atomically), marks it `Running`, records the units it holds, and returns the
//! newly-started ids for the caller's runner to execute. The runner reports each
//! node's result back with [`Scheduler::complete`]; a running node may grow more
//! work first via [`Scheduler::append`]. Concurrency is emergent from resource
//! capacity — there is no separate active-node cap.
//! atomically), marks it `Running`, records the units it holds, and hands back
//! a future that executes the node **and completes it**, so "forgot to finish
//! the node" is not expressible. One at a time is the primitive on purpose: it
//! lets the caller choose between claiming again and backing off, which a batch
//! return can't express. A running node may grow more work by declaring into
//! the builder it was handed. Concurrency is emergent from resource capacity.
//!
//! Single-threaded by design: the scheduler is the only driver, holds the
//! [`ResourceTable`] as a plain owned field, mutating it through `&mut self` —
@ -29,7 +30,9 @@
//! is a deferred optimization — unsafe under dynamically-appended subnodes.)
use std::collections::HashMap;
use std::future::Future;
use std::hash::Hash;
use std::sync::{Arc, Mutex};
use crate::builder::{BuildError, JobBuilder, NodeGuid};
use crate::resources::ResourceTable;
@ -86,8 +89,8 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
}
/// Append a node under `parent` — e.g. a running node growing more work into
/// its own subtree. Delegates to [`Graph::insert`]; call [`Scheduler::settle`]
/// afterwards to start it once it is runnable.
/// its own subtree. Delegates to [`Graph::insert`]; claim again afterwards
/// to start it once it is runnable.
///
/// # Errors
/// Propagates [`GraphError`] for a dangling dependency or parent id.
@ -114,8 +117,7 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
/// [`Graph::insert_unchecked`]: [`crate::builder::check_job_shape`] has
/// already decided every rejection the graph could raise, so re-validating
/// per node could only report a problem *after* the earlier nodes were
/// inserted. Call [`Scheduler::settle`] afterwards to start whatever became
/// runnable.
/// inserted. Claim again afterwards to start whatever became runnable.
///
/// **Atomic in the job's own shape.** A forward edge, a forward parent, or
/// a request for a handle this job never declared is rejected *before* the
@ -138,27 +140,80 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
})
}
/// Claim every currently-runnable pending node and start it: node-deps
/// Claim **one** currently-runnable pending node and start it: node-deps
/// satisfied and all resource-deps acquired atomically (all-or-nothing).
/// Each claimed node is marked `Running`, its acquired units recorded, and
/// its id returned for the runner to execute. A single pass suffices — a
/// node started here is `Running`, not terminal, so it cannot satisfy another
/// node's dependency in the same pass; it only consumes resources.
/// The node is marked `Running`, its acquired units recorded, and its id
/// returned for the caller to execute. `None` means nothing is runnable
/// right now — which is a different statement from "nothing is pending".
///
/// **Private**: [`Self::claim_next`] is the only way out of this crate.
/// Claiming without the future that completes the node is the sequence the
/// seam exists to make inexpressible, so the primitive stays in here.
#[must_use]
pub fn settle(&mut self) -> Vec<NodeId> {
fn claim_one(&mut self) -> Option<NodeId> {
let pending: Vec<NodeId> = self
.graph
.nodes()
.filter(|n| n.state == State::Pending)
.map(|n| n.id)
.collect();
let mut started = Vec::new();
for id in pending {
if self.node_deps_satisfied(id) && self.try_start(id) {
started.push(id);
}
}
started
pending
.into_iter()
.find(|&id| self.node_deps_satisfied(id) && self.try_start(id))
}
/// Claim one runnable node and return **the work that runs it**, or `None`
/// when nothing is runnable right now.
///
/// This is the seam: the caller supplies how to execute a node and spawns
/// the returned future, but never touches claiming or completion. The
/// future runs the node **and completes it**, so "forgot to finish the
/// node" is not expressible — completion is inside the thing you spawn.
///
/// The `Option` is answered *synchronously*, before anything is awaited, so
/// the caller can decide "claim again immediately" vs "back off" without
/// waiting on the node it just started.
///
/// ## Locking
/// The lock is taken twice, briefly, and **never held across the await**:
/// once here to claim, once inside the future to complete. That is what
/// keeps the returned future `Send` — a guard alive across an await point
/// would poison it — and it is why the node itself runs unlocked, for
/// however many minutes it needs.
///
/// ## Why the payload is cloned
/// `run` gets an owned `N` rather than a borrow: a `&N` parameter is live
/// for the whole future, which both borrows the graph across the await and
/// makes the future non-`Send`.
///
/// The output carries the insert result rather than swallowing it — this
/// crate has no logger, so a malformed grown job is reported to the caller,
/// who is the one that can log it. The node is completed either way: its
/// own work already happened.
pub fn claim_next<F, Fut>(
sched: &Arc<Mutex<Self>>,
run: F,
) -> Option<impl Future<Output = (NodeId, Result<(), BuildError>)> + use<F, Fut, N, R>>
where
N: Clone,
F: FnOnce(NodeId, N, JobBuilder<N, R>) -> Fut,
Fut: Future<Output = (JobBuilder<N, R>, Outcome)>,
{
let (id, payload) = {
let mut guard = sched.lock().expect("jobq scheduler mutex poisoned");
let id = guard.claim_one()?;
let payload = guard.graph.node(id)?.payload.clone();
(id, payload)
};
let sched = Arc::clone(sched);
Some(async move {
let (grown, outcome) = run(id, payload, JobBuilder::new()).await;
let grew = sched
.lock()
.expect("jobq scheduler mutex poisoned")
.complete_growing(id, outcome, grown);
(id, grew)
})
}
/// Try to start node `id`. For each resource it needs, decide per the parent
@ -248,9 +303,13 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
/// (every child `Done`) or [`State::Failed`] (any child `Failed`/`Cancelled`).
/// On failure it is `Failed` at once and its pending sub-nodes are cancelled
/// (gated on a `Finishing` the parent never reached). Terminality then
/// propagates up the parent chain. Call [`Scheduler::settle`] again afterwards
/// to start newly-unblocked work.
pub fn complete(&mut self, id: NodeId, outcome: Outcome) {
/// propagates up the parent chain. Claim again afterwards to start
/// newly-unblocked work.
///
/// `pub(crate)`: completion is reachable only from inside the future
/// [`Self::claim_next`] hands back, so it is not expressible without the
/// claim it answers.
pub(crate) fn complete(&mut self, id: NodeId, outcome: Outcome) {
match outcome {
Outcome::Failed(error) => {
// Record the reason before the terminal transition so it's set
@ -265,6 +324,64 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
self.release_ready();
}
/// [`Scheduler::complete`], plus whatever the node declared into the builder
/// it was handed while running.
///
/// `grown`'s nodes are inserted **under `id`** and *before* the completion,
/// so the node cannot roll terminal with its own appended work still
/// pending — the same ordering the caller previously had to arrange by
/// hand. A job that declares nothing costs nothing: the insert is skipped
/// outright, which is the overwhelmingly common case (most nodes grow no
/// work at all).
///
/// **A failed node grows nothing**, whatever it declared. Failure
/// cancel-cascades to every pending child of `id`, so work inserted here
/// would be `Skipped` by the very next statement — the insert is not wrong,
/// it is provably pointless. This lives here rather than in the caller
/// because it is a consequence of *this crate's* cascade rule; a host that
/// had to remember it could forget it.
///
/// # Errors
/// [`BuildError`] if `grown` is malformed — **and the node is still
/// completed**. Its own work already happened; refusing to complete it
/// would misreport that, and leaving it `Running` forever would wedge the
/// DAG. So the error is returned for the caller to log, not used to abort
/// the completion. This crate has no logger of its own; the caller does.
pub(crate) fn complete_growing(
&mut self,
id: NodeId,
outcome: Outcome,
grown: JobBuilder<N, R>,
) -> Result<(), BuildError> {
// A node that is no longer in the graph grows nothing. The insert below
// is *unchecked* — rooting on a departed parent would plant a dangling
// `parent` edge rather than being rejected. The host used to carry this
// guard itself, as a lookup before a separate append call; it belongs
// here, where the graph is and where it cannot be skipped.
//
// ⚠️ Deliberately untested, and untestable today: nothing removes a node
// from the graph yet (eviction only stops *retaining* a DAG; its nodes
// linger), and `NodeId` cannot be fabricated, so a test would have to
// fake the very condition it checks. This guard is defensive against the
// bounded prune that does not exist yet — when that lands, it needs a
// test, and this comment is the reminder.
let grew = if grown.is_empty()
|| matches!(outcome, Outcome::Failed(_))
|| self.graph.node(id).is_none()
{
Ok(())
} else {
let graph = &mut self.graph;
grown
.insert_with(Some(id), &[], |payload, deps, parent| {
graph.insert_unchecked(payload, deps, parent)
})
.map(|_ids| ())
};
self.complete(id, outcome);
grew
}
/// Whether every direct child of `id` is terminal.
fn all_children_terminal(&self, id: NodeId) -> bool {
self.graph
@ -491,7 +608,7 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
/// done" signal) supplies parent→child ordering; `Dep::Node` edges (which the
/// graph restricts to the same parent group) supply sibling ordering.
/// `Dep::Resource` edges are handled by the atomic acquire in
/// [`Scheduler::settle`], not here.
/// [`Scheduler::try_start`], not here.
fn node_deps_satisfied(&self, id: NodeId) -> bool {
let Some(node) = self.graph.node(id) else {
return false;
@ -533,6 +650,25 @@ mod tests {
name.to_owned()
}
/// Claim every currently-runnable node, as arrangement for the assertions
/// below. Equivalent to calling [`Scheduler::claim_one`] until it yields
/// `None`: a node started by an earlier iteration is `Running`, not
/// terminal, so it cannot satisfy another node's dependency here — it only
/// consumes resources.
///
/// **Was `Scheduler::settle`, a public method.** It was a `claim_one` loop
/// returning a `Vec`, and production never wanted the batch: the run loop
/// takes one node at a time through [`Scheduler::claim_next`] so it can
/// choose between claiming again and backing off, which a batch return
/// can't express. The only callers were tests, so it lives with them.
fn settle<N, R: Clone + Eq + Hash>(s: &mut Scheduler<N, R>) -> Vec<NodeId> {
let mut started = Vec::new();
while let Some(id) = s.claim_one() {
started.push(id);
}
started
}
/// A graph + a resource table with `build-slot` set to `slots`.
fn scheduler_with_slots(slots: u32) -> Scheduler<&'static str, String> {
let mut table = ResourceTable::new();
@ -559,13 +695,97 @@ mod tests {
s.resources.available(&res(name))
}
/// Children of `id`, by payload, in insertion order.
fn children_of(s: &Scheduler<&'static str, String>, id: NodeId) -> Vec<&'static str> {
s.graph()
.nodes()
.filter(|n| n.parent == Some(id))
.map(|n| n.payload)
.collect()
}
/// A node completing `Done` gets the work it declared while running,
/// inserted **under itself** — so the DAG cannot roll terminal with the
/// appended work still pending.
#[test]
fn a_completing_node_grows_the_work_it_declared() {
let mut s = scheduler_with_slots(1);
let n = s.append("emitter", vec![], None).expect("insert");
assert_eq!(settle(&mut s), vec![n]);
let grown = JobBuilder::new();
grown.node("child-a");
grown.node("child-b");
s.complete_growing(n, Outcome::Done, grown)
.expect("well-formed growth");
assert_eq!(children_of(&s, n), vec!["child-a", "child-b"]);
// The emitter parks in `Finishing` rather than going terminal: its own
// appended work is still pending under it. That ordering is the whole
// point of growing *as part of* the completion.
assert_eq!(s.graph().node(n).unwrap().state, State::Finishing);
}
/// A **failed** node grows nothing, whatever it declared.
///
/// The companion to the test above, and the reason this rule lives in the
/// crate rather than in a caller: failure cancel-cascades to every pending
/// child of the completing node, so anything inserted here would be
/// `Skipped` by the very next statement. Enforcing it host-side means every
/// host has to remember it; enforcing it here means none can forget.
#[test]
fn a_failed_node_grows_nothing() {
let mut s = scheduler_with_slots(1);
let n = s.append("emitter", vec![], None).expect("insert");
assert_eq!(settle(&mut s), vec![n]);
let grown = JobBuilder::new();
grown.node("never-runs");
s.complete_growing(n, Outcome::Failed("boom".to_owned()), grown)
.expect("growth is dropped, not rejected");
assert!(
children_of(&s, n).is_empty(),
"a failed node must not append work, got {:?}",
children_of(&s, n)
);
assert_eq!(s.graph().node(n).unwrap().state, State::Failed);
}
/// A contended resource goes to the oldest waiter.
///
/// [`Scheduler::claim_one`] scans [`Graph::nodes`] — insertion order — and
/// takes the first node whose deps are satisfied and whose resources it can
/// acquire. That *is* the fairness guarantee: there is no queue, no
/// priority, just the scan order.
///
/// Load-bearing for any host that submits work over time, because without
/// it a steady arrival rate could starve the earliest waiter indefinitely.
/// It was previously only covered downstream, by a host test driving its own
/// templates — which meant the property this crate provides was asserted
/// everywhere except in this crate.
#[test]
fn a_contended_resource_goes_to_the_oldest_waiter() {
let mut s = scheduler_with_slots(1);
let a = s.append("a", res_dep("build-slot"), None).expect("a");
let b = s.append("b", res_dep("build-slot"), None).expect("b");
let c = s.append("c", res_dep("build-slot"), None).expect("c");
assert_eq!(settle(&mut s), vec![a], "cap 1: only the first can start");
s.complete(a, Outcome::Done);
// b and c are both satisfiable now; b was inserted first.
assert_eq!(settle(&mut s), vec![b], "the freed unit goes to b, not c");
s.complete(b, Outcome::Done);
assert_eq!(settle(&mut s), vec![c]);
}
#[test]
fn leaf_owner_goes_done_directly_and_releases() {
let mut s = scheduler_with_slots(1);
let n = s
.append("build", res_dep("build-slot"), None)
.expect("insert");
assert_eq!(s.settle(), vec![n]);
assert_eq!(settle(&mut s), vec![n]);
assert_eq!(s.graph().node(n).unwrap().state, State::Running);
assert_eq!(avail(&s, "build-slot"), 0);
// No children → completing it goes straight to Done (skips Finishing).
@ -585,7 +805,7 @@ mod tests {
assert!(s.graph().node(ok).unwrap().started_at.is_none());
assert!(s.graph().node(ok).unwrap().finished_at.is_none());
let started = s.settle();
let started = settle(&mut s);
assert!(started.contains(&ok) && started.contains(&bad));
// Running → started_at stamped, finished_at still none.
assert!(s.graph().node(ok).unwrap().started_at.is_some());
@ -621,11 +841,11 @@ mod tests {
let b = s.append("b", res_dep("build-slot"), None).expect("b");
let c = s.append("c", res_dep("build-slot"), None).expect("c");
// cap 2 → a + b start, c blocks on the exhausted slot.
assert_eq!(s.settle(), vec![a, b]);
assert_eq!(settle(&mut s), vec![a, b]);
assert_eq!(s.graph().node(c).unwrap().state, State::Pending);
// a finishes → its slot frees → c can now start.
s.complete(a, Outcome::Done);
assert_eq!(s.settle(), vec![c]);
assert_eq!(settle(&mut s), vec![c]);
assert_eq!(s.graph().node(c).unwrap().state, State::Running);
}
@ -637,16 +857,16 @@ mod tests {
let root = s.append("root", vec![], None).expect("root");
let c1 = s.append("c1", vec![], Some(root)).expect("c1");
let c2 = s.append("c2", vec![], Some(root)).expect("c2");
assert_eq!(s.settle(), vec![root]);
assert_eq!(settle(&mut s), vec![root]);
// Children can't start yet — parent still Running (logic not done).
assert!(s.settle().is_empty(), "children gated on parent logic");
assert!(settle(&mut s).is_empty(), "children gated on parent logic");
s.complete(root, Outcome::Done);
assert_eq!(
s.graph().node(root).unwrap().state,
State::Finishing,
"logic done, children pending → Finishing"
);
let mut started = s.settle();
let mut started = settle(&mut s);
started.sort();
let mut expected = vec![c1, c2];
expected.sort();
@ -670,9 +890,9 @@ mod tests {
let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
let root = s.append("root", vec![], None).expect("root");
let child = s.append("child", vec![], Some(root)).expect("child");
assert_eq!(s.settle(), vec![root]);
assert_eq!(settle(&mut s), vec![root]);
s.complete(root, Outcome::Done);
assert_eq!(s.settle(), vec![child]);
assert_eq!(settle(&mut s), vec![child]);
s.complete(child, Outcome::Failed(String::new()));
assert_eq!(
s.graph().node(root).unwrap().state,
@ -690,10 +910,10 @@ mod tests {
let r = s.append("R", res_dep("build-slot"), None).expect("R");
let c1 = s.append("c1", res_dep("build-slot"), Some(r)).expect("c1");
let c2 = s.append("c2", vec![after_ok(c1)], Some(r)).expect("c2");
assert_eq!(s.settle(), vec![r]);
assert_eq!(settle(&mut s), vec![r]);
s.complete(r, Outcome::Done); // → Finishing (children pending)
assert_eq!(avail(&s, "build-slot"), 0, "held: subtree not terminal");
assert_eq!(s.settle(), vec![c1], "c1 borrows R's slot");
assert_eq!(settle(&mut s), vec![c1], "c1 borrows R's slot");
assert_eq!(avail(&s, "build-slot"), 0, "borrow reuses R's unit");
s.complete(c1, Outcome::Done);
assert_eq!(
@ -701,7 +921,7 @@ mod tests {
0,
"still held: c2 pending in subtree"
);
assert_eq!(s.settle(), vec![c2]);
assert_eq!(settle(&mut s), vec![c2]);
s.complete(c2, Outcome::Done);
assert_eq!(
avail(&s, "build-slot"),
@ -719,14 +939,14 @@ mod tests {
let owner = s
.append("owner", res_dep("agent/foo"), None)
.expect("owner");
assert_eq!(s.settle(), vec![owner]);
assert_eq!(settle(&mut s), vec![owner]);
assert_eq!(avail(&s, "agent/foo"), 0);
let child = s
.append("child", res_dep("agent/foo"), Some(owner))
.expect("child");
s.complete(owner, Outcome::Done); // → Finishing
assert_eq!(avail(&s, "agent/foo"), 0, "held while a borrower pends");
assert_eq!(s.settle(), vec![child]);
assert_eq!(settle(&mut s), vec![child]);
assert_eq!(avail(&s, "agent/foo"), 0, "borrow reuses the one unit");
s.complete(child, Outcome::Done);
assert_eq!(avail(&s, "agent/foo"), 1);
@ -749,13 +969,13 @@ mod tests {
let great = s
.append("great", res_dep("agent/foo"), Some(grand))
.expect("great");
assert_eq!(s.settle(), vec![r]);
assert_eq!(settle(&mut s), vec![r]);
s.complete(r, Outcome::Done);
assert_eq!(s.settle(), vec![child], "child borrows R's grant");
assert_eq!(settle(&mut s), vec![child], "child borrows R's grant");
s.complete(child, Outcome::Done);
assert_eq!(s.settle(), vec![grand], "grand covered, no deadlock");
assert_eq!(settle(&mut s), vec![grand], "grand covered, no deadlock");
s.complete(grand, Outcome::Done);
assert_eq!(s.settle(), vec![great], "great covered too");
assert_eq!(settle(&mut s), vec![great], "great covered too");
assert_eq!(avail(&s, "agent/foo"), 0, "held across the whole nest");
s.complete(great, Outcome::Done);
assert_eq!(s.graph().node(r).unwrap().state, State::Done, "R rolled up");
@ -769,10 +989,14 @@ mod tests {
let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
let a = s.append("a", res_dep("agent/foo"), None).expect("a");
let b = s.append("b", res_dep("agent/foo"), None).expect("b");
assert_eq!(s.settle(), vec![a], "only a acquires; b can't borrow it");
assert_eq!(
settle(&mut s),
vec![a],
"only a acquires; b can't borrow it"
);
assert_eq!(s.graph().node(b).unwrap().state, State::Pending);
s.complete(a, Outcome::Done);
assert_eq!(s.settle(), vec![b]);
assert_eq!(settle(&mut s), vec![b]);
assert_eq!(s.graph().node(b).unwrap().state, State::Running);
}
@ -785,7 +1009,7 @@ mod tests {
let owner = s
.append("owner", res_dep("agent/foo"), None)
.expect("owner");
assert_eq!(s.settle(), vec![owner]);
assert_eq!(settle(&mut s), vec![owner]);
let c1 = s
.append("c1", res_dep("agent/foo"), Some(owner))
.expect("c1");
@ -793,10 +1017,10 @@ mod tests {
.append("c2", res_dep("agent/foo"), Some(owner))
.expect("c2");
s.complete(owner, Outcome::Done); // → Finishing
assert_eq!(s.settle(), vec![c1], "c1 borrows; c2 can't (cap 1)");
assert_eq!(settle(&mut s), vec![c1], "c1 borrows; c2 can't (cap 1)");
assert_eq!(s.graph().node(c2).unwrap().state, State::Pending);
s.complete(c1, Outcome::Done);
assert_eq!(s.settle(), vec![c2], "borrow returned → c2 borrows");
assert_eq!(settle(&mut s), vec![c2], "borrow returned → c2 borrows");
assert_eq!(avail(&s, "agent/foo"), 0, "still just the owner's unit");
}
@ -808,7 +1032,7 @@ mod tests {
let owner = s
.append("owner", res_dep("build-slot"), None)
.expect("owner");
assert_eq!(s.settle(), vec![owner]);
assert_eq!(settle(&mut s), vec![owner]);
assert_eq!(avail(&s, "build-slot"), 1, "owner took one of two");
let c1 = s
.append("c1", res_dep("build-slot"), Some(owner))
@ -817,7 +1041,7 @@ mod tests {
.append("c2", res_dep("build-slot"), Some(owner))
.expect("c2");
s.complete(owner, Outcome::Done); // → Finishing
let mut started = s.settle();
let mut started = settle(&mut s);
started.sort();
let mut expected = vec![c1, c2];
expected.sort();
@ -847,11 +1071,11 @@ mod tests {
None,
)
.expect("weak");
assert_eq!(s.settle(), vec![root]);
assert_eq!(settle(&mut s), vec![root]);
s.complete(root, Outcome::Failed(String::new()));
assert_eq!(s.graph().node(strong1).unwrap().state, State::Skipped);
assert_eq!(s.graph().node(strong2).unwrap().state, State::Skipped);
assert_eq!(s.settle(), vec![weak]);
assert_eq!(settle(&mut s), vec![weak]);
}
/// The direction only a *set* edge can express: a branch that runs solely on
@ -871,7 +1095,7 @@ mod tests {
None,
)
.expect("compensate");
assert_eq!(s.settle(), vec![root]);
assert_eq!(settle(&mut s), vec![root]);
s.complete(root, Outcome::Done);
assert_eq!(
s.graph().node(on_fail).unwrap().state,
@ -879,7 +1103,7 @@ mod tests {
"a Failed-only branch is unsatisfiable once its dep succeeds — and it is \
`Skipped`, not `Cancelled`, so the parent roll-up ignores it"
);
assert!(s.settle().is_empty(), "and nothing is left runnable");
assert!(settle(&mut s).is_empty(), "and nothing is left runnable");
}
/// The mirror: the same branch is exactly what *does* run on failure, while
@ -901,10 +1125,10 @@ mod tests {
None,
)
.expect("on_fail");
assert_eq!(s.settle(), vec![root]);
assert_eq!(settle(&mut s), vec![root]);
s.complete(root, Outcome::Failed("boom".to_owned()));
assert_eq!(s.graph().node(on_ok).unwrap().state, State::Skipped);
assert_eq!(s.settle(), vec![on_fail]);
assert_eq!(settle(&mut s), vec![on_fail]);
}
/// A weak edge accepts a dependency that was *ruled out*, so a tail still
@ -925,11 +1149,11 @@ mod tests {
None,
)
.expect("tail");
assert_eq!(s.settle(), vec![root]);
assert_eq!(settle(&mut s), vec![root]);
s.complete(root, Outcome::Failed("boom".to_owned()));
assert_eq!(s.graph().node(mid).unwrap().state, State::Skipped);
assert_eq!(
s.settle(),
settle(&mut s),
vec![tail],
"the tail runs off a cancelled dependency"
);
@ -966,22 +1190,26 @@ mod tests {
// Everything succeeds: the ok branch runs, the failure branch is ruled out.
let (mut s, a, b, on_ok, on_fail) = build();
assert_eq!(s.settle(), vec![a, b], "both roots start; neither tail can");
assert_eq!(
settle(&mut s),
vec![a, b],
"both roots start; neither tail can"
);
s.complete(a, Outcome::Done);
s.complete(b, Outcome::Done);
assert_eq!(s.settle(), vec![on_ok]);
assert_eq!(settle(&mut s), vec![on_ok]);
s.complete(on_ok, Outcome::Done);
assert_eq!(s.graph().node(on_fail).unwrap().state, State::Skipped);
assert!(s.settle().is_empty());
assert!(settle(&mut s).is_empty());
// One of them fails: the ok branch is ruled out, which is precisely the
// signal the failure branch waits on.
let (mut s, a, b, on_ok, on_fail) = build();
assert_eq!(s.settle(), vec![a, b]);
assert_eq!(settle(&mut s), vec![a, b]);
s.complete(a, Outcome::Failed("boom".to_owned()));
s.complete(b, Outcome::Done);
assert_eq!(s.graph().node(on_ok).unwrap().state, State::Skipped);
assert_eq!(s.settle(), vec![on_fail]);
assert_eq!(settle(&mut s), vec![on_fail]);
}
#[test]
@ -992,7 +1220,7 @@ mod tests {
let root = s.append("root", vec![], None).expect("root");
let child = s.append("child", vec![], Some(root)).expect("child");
let grandchild = s.append("gc", vec![], Some(child)).expect("gc");
assert_eq!(s.settle(), vec![root]);
assert_eq!(settle(&mut s), vec![root]);
s.complete(root, Outcome::Failed(String::new()));
assert_eq!(s.graph().node(child).unwrap().state, State::Skipped);
assert_eq!(s.graph().node(grandchild).unwrap().state, State::Skipped);
@ -1008,7 +1236,7 @@ mod tests {
assert_eq!(s.graph().node(a).unwrap().state, State::Cancelled);
assert_eq!(s.graph().node(b).unwrap().state, State::Skipped);
let c = s.append("c", vec![], None).expect("c");
assert_eq!(s.settle(), vec![c]);
assert_eq!(settle(&mut s), vec![c]);
assert!(!s.cancel_node(c));
assert_eq!(s.graph().node(c).unwrap().state, State::Running);
}
@ -1024,7 +1252,7 @@ mod tests {
let b = s.append("b", vec![after_ok(a)], Some(root)).expect("b");
// The root runs first and parks in `Finishing` while its children are
// outstanding — the state a group root is actually in when cancelled.
assert_eq!(s.settle(), vec![root]);
assert_eq!(settle(&mut s), vec![root]);
s.complete(root, Outcome::Done);
assert_eq!(s.graph().node(root).unwrap().state, State::Finishing);
@ -1045,9 +1273,9 @@ mod tests {
let root = s.append("root", vec![], None).expect("root");
let a = s.append("a", vec![], Some(root)).expect("a");
let b = s.append("b", vec![after_ok(a)], Some(root)).expect("b");
assert_eq!(s.settle(), vec![root]);
assert_eq!(settle(&mut s), vec![root]);
s.complete(root, Outcome::Done);
assert_eq!(s.settle(), vec![a], "a is claimed and running");
assert_eq!(settle(&mut s), vec![a], "a is claimed and running");
assert!(!s.cancel_node(root), "refused while a runs");
assert_eq!(s.graph().node(a).unwrap().state, State::Running);
@ -1076,7 +1304,7 @@ mod tests {
Some(root),
)
.expect("tail");
assert_eq!(s.settle(), vec![root]);
assert_eq!(settle(&mut s), vec![root]);
s.complete(root, Outcome::Done);
assert!(s.cancel_node(root));
@ -1086,7 +1314,7 @@ mod tests {
State::Pending,
"spared, and now runnable since its dep is Cancelled"
);
assert_eq!(s.settle(), vec![tail], "the tail still gets to report");
assert_eq!(settle(&mut s), vec![tail], "the tail still gets to report");
}
#[test]
@ -1094,7 +1322,7 @@ mod tests {
let mut s = scheduler_with_slots(1);
let g = s.append("g", res_dep("agent/foo"), None).expect("g");
let b = s.append("b", res_dep("build-slot"), None).expect("b");
assert_eq!(s.settle().len(), 2);
assert_eq!(settle(&mut s).len(), 2);
let state = s.resource_state();
assert!(state.contains(&(res("agent/foo"), g)));
assert!(state.contains(&(res("build-slot"), b)));