Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
221c3f35a1 | ||
|
|
6b800600e2 | ||
|
|
4590b04cac |
37 changed files with 966 additions and 1659 deletions
|
|
@ -145,9 +145,10 @@ commit the agent's last successful build pinned.
|
|||
|
||||
swarm-controller commits `internal/meta`: one `flake.nix` and `flake.lock` for
|
||||
every agent any hive places (the pins:
|
||||
[`docs/swarm/builder.md`](../swarm/builder.md#meta-flake-lock)). A hive builds
|
||||
every agent from it. While the repo has no commit that pins an agent's current
|
||||
config, that agent's build fails and names the agent. The controller queues
|
||||
[`docs/swarm/builder.md`](../swarm/builder.md#meta-flake-lock)). The swarm
|
||||
builder builds every agent from it, and a hive deploys that build. While the
|
||||
repo has no commit that pins an agent's current config, that agent's deploy
|
||||
fails and names the agent. The controller queues
|
||||
its update of the repo before the deploy request for a merged config commit
|
||||
and for a new agent, and sends the deploy only when that update succeeded and
|
||||
the swarm builder built the agent from the updated repo (the
|
||||
|
|
@ -159,28 +160,24 @@ the swarm builder built the agent from the updated repo (the
|
|||
a custom agent from the config commit at the head of `agent-configs/<n>`
|
||||
`main`, an archetype agent from the archetype it has now. An agent switched
|
||||
between custom and archetype, or to another archetype, fails that test.
|
||||
A hive without the forge credential file that nix's git reads, a repo with no
|
||||
commit, a head that lacks the agent and a head behind the config each fail
|
||||
the build with an error that names the agent. `MetaSync` records the commit,
|
||||
A repo with no commit, a head that lacks the agent and a head behind the
|
||||
config each fail the deploy with an error that names the agent. `MetaSync` records the commit,
|
||||
with the hyperhive and config commits the head pins, in `hive-c0re`'s own state,
|
||||
`/var/lib/hyperhive/meta-rev/<n>`, which the agent can't write (the container mounts its runtime
|
||||
directory read-write); `Prebuild` and `Swap` read that
|
||||
record, so both build one flake. A first deploy resolves the commit in its `Create`
|
||||
record, so both expect the closure of one commit. A first deploy resolves the commit in its `Create`
|
||||
node, since it has no `MetaSync`.
|
||||
- **The build.** `hive-c0re` prebuilds
|
||||
`git+<forge>/internal/meta.git?rev=<commit>#nixosConfigurations.<n>…` and
|
||||
names the commit to `hive-priv` (`meta_rev` on `UpdateContainer` and
|
||||
`CreateContainer`). `hive-priv` accepts only a 40-digit lowercase hex
|
||||
commit and builds the flake from its own forge URL
|
||||
(`HIVE_PRIV_FORGE_URL`); the URL never comes from a request. On the wire the
|
||||
- **The closure.** The hive builds nothing. `Prebuild`, `Swap` and `Create` read
|
||||
the agent's record in the swarm's `agent-closures` bucket (the controller
|
||||
writes it from the builder's reply) and have `hive-priv` download the closure
|
||||
it names from the swarm cache (`RealiseClosure`). `Swap` and `Create` then
|
||||
give `hive-priv` the store path (`toplevel` on `UpdateContainer` and
|
||||
`CreateContainer`), which accepts only the path its last `RealiseClosure` for
|
||||
that agent returned. On the wire the
|
||||
`hive_priv_sock::PrivRequest` variants carry a required field:
|
||||
`{"op":"update_container","name":"atlas","stream":true,"meta_rev":"<40 hex>"}`.
|
||||
- **Fetching.** nix fetches the forge's inputs with `git`. Both `hive-c0re`
|
||||
and `hive-priv` carry a `$HOME/.gitconfig` that answers for the forge host
|
||||
through `git-credential-hive-forge` (the hive's forge read token, published to
|
||||
`/run/hyperhive/forge-hive-token`) and rewrites the
|
||||
forge's `https` URL, which `internal/meta`'s lock names its inputs by, to
|
||||
`http`.
|
||||
`{"op":"update_container","name":"atlas","stream":true,"toplevel":"/nix/store/<hash>-nixos-system-atlas"}`.
|
||||
The three cases when the swarm's closure isn't the one this hive expects
|
||||
are in [`docs/swarm/builder.md`](../swarm/builder.md#rebuilds-download-the-closure).
|
||||
- **Hyperhive.** The commit pins one hyperhive rev for the swarm, so an agent
|
||||
runs the hyperhive of the controller's deploy, not its hive's.
|
||||
- **Markers.** After a successful build `hive-c0re` writes the hyperhive
|
||||
|
|
|
|||
|
|
@ -69,14 +69,11 @@ forge container:
|
|||
- seeding `proposed` from `agent-configs/<agent>` on first deploy
|
||||
- cloning and pulling `/knowledge`
|
||||
|
||||
The builds read as the hive's account too. nix fetches the config inputs and
|
||||
`internal/meta` with `git`, and `git` gets its credential from
|
||||
`git-credential-hive-forge`, which reads `/run/hyperhive/forge-hive-token`.
|
||||
hive-c0re writes that file (account on the first line, token on the second)
|
||||
from the hive's store token at each boot and each `internal/meta` read, when its
|
||||
content differs. A hive whose store holds no token has no such file, fetches
|
||||
nothing private through nix and fails its builds: they have no other flake than
|
||||
`internal/meta`. This also holds on a hive with no forge container.
|
||||
The hive builds nothing, so no nix process reads the forge as the hive's
|
||||
account. hive-c0re hands the token to each `git` command it runs
|
||||
(`http.extraHeader`) and keeps no copy on disk. A hive whose store holds no token
|
||||
reads nothing private from the forge. This also holds on a hive with no forge
|
||||
container.
|
||||
|
||||
Roll a change to any of this out store host first (its granter writes the
|
||||
controller's grant on `swarm/hives/+/forge/token`), then the controller host,
|
||||
|
|
|
|||
|
|
@ -301,12 +301,6 @@ nix build "git+<forge>/internal/meta.git?rev=<rev>#argus.config.system.build.top
|
|||
nix build "git+<forge>/internal/meta.git?rev=<rev>#nixosConfigurations.argus.config.system.build.toplevel"
|
||||
```
|
||||
|
||||
`lifecycle::prebuild_toplevel` hit this once by constructing the attr
|
||||
path as `{flake_ref}.config…` — which produced `#argus.config…`
|
||||
instead of `#nixosConfigurations.argus.config…`. The fix:
|
||||
`split_once('#')` to separate flake path from name, then template
|
||||
`{path}#nixosConfigurations.{name}.config.system.build.toplevel`.
|
||||
|
||||
### Containerized nix-daemon needs `sandbox-fallback = true`
|
||||
|
||||
Agent containers bind-mount the host's nix-daemon socket. nspawn
|
||||
|
|
|
|||
|
|
@ -42,11 +42,11 @@ because there is no malformed spec to reject.
|
|||
|
||||
Nix-heavy — hold one of the `buildSlots` permits for the node's duration:
|
||||
|
||||
| Node | Wraps |
|
||||
| ---------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
|
||||
| `Prebuild` | `lifecycle::prebuild_toplevel` — build the toplevel out-of-band while the container keeps serving (its meta preamble is the upstream `MetaSync` node). Skipped when the container is already down; `Swap` builds inline instead |
|
||||
| `Swap` | drop-in rewrite + `nixos-container update` profile-swap (requires the container stopped); the post-swap bookkeeping tail lives in the sibling `RebuildBookkeeping` node |
|
||||
| `Create` | first-spawn `nixos-container create` proper; assumes the upstream `Provision` node already registered the agent; downloads the closure the swarm cached for the `internal/meta` commit it resolves, else builds it locally, and marks the agent's pins |
|
||||
| Node | Wraps |
|
||||
| ---------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
|
||||
| `Prebuild` | download the swarm's closure for the agent into the store while the container keeps serving (its meta preamble is the upstream `MetaSync` node). With no closure to deploy it fails here, before the stop. Skipped when the container is already down; `Swap` downloads it instead |
|
||||
| `Swap` | drop-in rewrite + profile-swap onto the swarm's closure (requires the container stopped); the post-swap bookkeeping tail lives in the sibling `RebuildBookkeeping` node |
|
||||
| `Create` | first-spawn `nixos-container create` proper; assumes the upstream `Provision` node already registered the agent; downloads the closure the swarm built for the agent and creates the container from it (with none to deploy it fails and creates nothing), then marks the agent's pins |
|
||||
|
||||
Cheap — no build slot:
|
||||
|
||||
|
|
@ -54,7 +54,7 @@ Cheap — no build slot:
|
|||
|
||||
| Node | Behavior |
|
||||
| -------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
|
||||
| `MetaSync` | the rebuild's preamble — rebuild-dir prep, the agent-set refresh (`agent_set::refresh`), and the `internal/meta` commit `Prebuild` and `Swap` build from, resolved once and recorded; fails when the repo doesn't pin the agent's current config. Declares no resources |
|
||||
| `MetaSync` | the rebuild's preamble — rebuild-dir prep, the agent-set refresh (`agent_set::refresh`), and the `internal/meta` commit `Prebuild` and `Swap` expect the swarm's closure to be built from, resolved once and recorded; fails when the repo doesn't pin the agent's current config. Declares no resources |
|
||||
| `Provision` | first-spawn pre-create provisioning — proposed/applied repos, state subvolume, agent-set refresh (`agent_set::refresh`); runs ahead of `Create`. Store-only, no container yet |
|
||||
| `Reconcile` | idempotent power converge: read `wanted` (below) + observed state; start if `Up` & down (cold-start fallback included), stop if `Offline` & up, else noop |
|
||||
| `Start` | mechanical container start — runtime dir + drop-ins, `start_with_fallback`, MCP listener registration, the manager kick. Fanned out by a `Reconcile` that observed `wanted = Up` and the container down |
|
||||
|
|
@ -72,7 +72,6 @@ Cheap — no build slot:
|
|||
| `SetWanted` | write the durable power intent (`wanted = Up`/`Offline`) as the head node of a power-op DAG. Takes the agent lease even though it's a store write, so the intent write and the tail `Reconcile` are atomic per-agent — two racing power ops can't clobber each other's intent before either reconciles |
|
||||
| `EmitRebuilt` | tail of a rebuild — emit the agent's `Rebuilt` manager event (ok/fail per outcome, nothing on cancel). One node per agent _and_ per outcome |
|
||||
| `WriteDropin` | `set_nspawn_flags` + `set_resource_limits` + daemon-reload |
|
||||
| `ForgeSweep` | one-shot boot-time publish of the hive's forge read token for nix (`forge::ensure_all`) as a first-class node, so it shows as real work on the dashboard instead of running invisibly in a bare `tokio::spawn`. Agentless |
|
||||
| `KnowledgePull` | `/knowledge` pull (`knowledge::pull`): at boot (commits that landed while `hive-c0re` was down), on the swarm knowledge-changed event, and hourly as a fallback. Holds `Resource::KnowledgeTree` (capacity 1), so two pulls never overlap on the working tree; each trigger queues its own pass, which waits for the resource if one is already live. Agentless |
|
||||
| `WantedPull` | pull of the agent set the swarm controller declares for this hive (`wanted::pull`), converging the agents it names. Runs at boot, waiting up to a minute in total for the queue connection, and again after every attach of the wanted-state watch (which retries with a backoff when the attach fails or the stream ends); the deploy event (`swarm_status`) and the watch are the fast path, this repairs a missed one. A first deploy already queued or running for an agent isn't queued again. Agentless |
|
||||
|
||||
|
|
@ -186,7 +185,7 @@ resources are free. Resources:
|
|||
clobber intent before either reconciles.) **Lease-exempt**: `MetaSync`,
|
||||
`Prebuild`, `Provision` — they
|
||||
touch the store, not the running container, which is exactly why a
|
||||
stop can land while another DAG's prebuild is still building. Also exempt, for a different
|
||||
stop can land while another DAG's prebuild is still downloading. Also exempt, for a different
|
||||
reason, are the rebuild subtree's own members (`StopForUpdate`, `Swap`,
|
||||
`Signal`, `Drain`, `RebuildBookkeeping`): they genuinely do touch the
|
||||
container, but their `AgentWindow` brace holds the lease above them — see
|
||||
|
|
@ -344,17 +343,17 @@ one go, rather than the double-bounce a live `update` would trigger.
|
|||
Sequence for a rebuild DAG (each step is its own queue node):
|
||||
|
||||
1. `MetaSync` — rebuild-dir prep, the agent-set refresh, and the choice of the
|
||||
`internal/meta` commit to build from. Short, and it declares no resources.
|
||||
2. `Prebuild` — build the new `system.build.toplevel` **before** stopping.
|
||||
The container keeps serving the previous generation while eval + fetch +
|
||||
build happen out-of-band. `nixos-container update` then finds the result
|
||||
cached and skips straight to the profile-swap. Build failures surface
|
||||
here, before `Swap` touches the running container. Skipped when the
|
||||
container is already down — a stopped agent has no uptime to preserve,
|
||||
so `Swap` builds inline instead of paying the double eval.
|
||||
`internal/meta` commit the swarm's closure should come from. Short,
|
||||
and it declares no resources.
|
||||
2. `Prebuild` — download the swarm's closure for the agent **before** stopping.
|
||||
The container keeps serving the previous generation while the download
|
||||
happens out-of-band. A missing or unpullable closure fails the DAG here,
|
||||
before `Swap` touches the running container, and the container keeps
|
||||
running. Skipped when the container is already down — a stopped agent has
|
||||
no uptime to preserve, so `Swap` downloads it.
|
||||
3. `StopForUpdate` — bring the container down (noop when already stopped).
|
||||
4. `Swap` — `nixos-container update` profile-swap from `internal/meta`
|
||||
(near-instant after the prebuild).
|
||||
4. `Swap` — profile-swap onto the swarm's closure (near-instant after the
|
||||
prebuild).
|
||||
5. `Reconcile` — boot into the new generation when `wanted = Up`; the
|
||||
in-container activation script transitions old → new. Holds no build
|
||||
slot, so the next DAG's `Prebuild` overlaps the container boot.
|
||||
|
|
@ -374,18 +373,10 @@ fallback.
|
|||
|
||||
### Spawn path (new container)
|
||||
|
||||
For a first-time `create`, `nixos-container create` is atomic: if the build fails,
|
||||
it leaves no container record to clean up. A separate prebuild would just duplicate
|
||||
the eval, so it's skipped. Sequence: `create` from `internal/meta` → write nspawn
|
||||
flags → `systemctl daemon-reload` → `start`.
|
||||
|
||||
### Prebuild attr path
|
||||
|
||||
`nix build` doesn't autoresolve `<flake-root>#<name>` against `nixosConfigurations` the
|
||||
way `nixos-container` does internally. nix requires the explicit attr path
|
||||
`<flake-root>#nixosConfigurations.<name>.config.system.build.toplevel`;
|
||||
using the bare `<flake-root>#<name>` ref would make nix look in `packages`, `legacyPackages`,
|
||||
or the flake root directly — none of which exist in the rendered meta flake.
|
||||
For a first-time `create`, `nixos-container create` is atomic: it leaves no container
|
||||
record to clean up if it fails. The `Create` node downloads the swarm's closure first,
|
||||
so a spawn with no closure to deploy fails before it creates the container. Sequence:
|
||||
download the closure → `create` from it → write nspawn flags → `systemctl daemon-reload` → `start`.
|
||||
|
||||
---
|
||||
|
||||
|
|
@ -396,11 +387,10 @@ limits, build parallelism, and first-spawn latency.
|
|||
|
||||
### Build slots
|
||||
|
||||
`buildSlots` (default `1`) sets how many nix-heavy job-queue nodes
|
||||
`buildSlots` (default `1`) sets how many closure-fetching job-queue nodes
|
||||
(prebuilds, profile swaps, first-spawn creates, meta lock bumps) run
|
||||
concurrently. The default serializes all heavy nix work like the pre-DAG
|
||||
rebuild queue did; raise it on hosts with the cores/RAM to build multiple
|
||||
agent toplevels at once. Per-agent correctness is independent of the count —
|
||||
concurrently. The default serializes all of them; raise it on hosts with
|
||||
the bandwidth and disk to take more agents at once. Per-agent correctness is independent of the count —
|
||||
each agent's container-affecting ops serialize on its lifecycle lease
|
||||
regardless.
|
||||
|
||||
|
|
|
|||
|
|
@ -200,12 +200,9 @@ minutes. The builder takes the same one-build slot as PR checks, and
|
|||
refuses a request that waited 30 seconds for it. It runs `nix build` of
|
||||
`nixosConfigurations.<agent>.config.system.build.toplevel` from
|
||||
`git+https://<forge>/internal/meta.git?rev=<head>`, with no input overridden:
|
||||
the committed `flake.lock` decides every input. Like the hive's build, it uses
|
||||
that attribute, so a green build here means the flake evaluates and builds for
|
||||
the builder's platform. A hive [downloads the closure instead of building it](#what-hives-do-with-the-cache)
|
||||
when the record is for the commit it builds from; otherwise it builds locally,
|
||||
from whatever head it reads at deploy time, which can be newer than the one the
|
||||
swarm built.
|
||||
the committed `flake.lock` decides every input. A green build here means the
|
||||
flake evaluates and builds for the builder's platform. A hive [downloads the closure](#rebuilds-download-the-closure)
|
||||
this build produced and never builds one itself.
|
||||
|
||||
The node ends `done` or `failed`. Every failure skips the deploy and shows on the
|
||||
Jobs page as a failed `build_meta` node, and the timer builds it again:
|
||||
|
|
@ -478,11 +475,9 @@ means check attic's fingerprint format; it says nothing about an attacker. A
|
|||
|
||||
### Rebuilds download the closure
|
||||
|
||||
A rebuild's `Prebuild` node, which runs while the agent is still up, tries the
|
||||
cache before it builds. It reads the agent's record in `agent-closures` and
|
||||
goes on only when the record is for the `internal/meta` commit `MetaSync`
|
||||
recorded for this rebuild and says the builder pushed the closure. It then
|
||||
asks `hive-priv` to realise the record's store path (`RealiseClosure`). `hive-priv` runs
|
||||
A hive builds nothing. A rebuild's `Prebuild` node, which runs while the agent is
|
||||
still up, reads the agent's record in `agent-closures` and asks `hive-priv` to
|
||||
realise the record's store path (`RealiseClosure`). `hive-priv` runs
|
||||
`nix-store --realise` with `--max-jobs 0`, so nix can't build the path, and with
|
||||
the host's configured cache address, and the public key and netrc from the
|
||||
root-owned `/run/hyperhive-priv/attic/`, as per-call settings. The settings replace the host's `substituters` and
|
||||
|
|
@ -490,32 +485,32 @@ root-owned `/run/hyperhive-priv/attic/`, as per-call settings. The settings repl
|
|||
source, and nix stores the path only when the cache's key signed it. The host's
|
||||
nix configuration stays as it was.
|
||||
|
||||
The deploy log says which way it went:
|
||||
`Swap` then does the same lookup and download, a no-op once the path is in the
|
||||
store, and gives `hive-priv` the store path to apply. A stopped agent skips
|
||||
`Prebuild`, so `Swap` downloads it. A first deploy's `Create` node does the
|
||||
same and creates the container from that path.
|
||||
|
||||
- `prebuild: substituted <path>`: the closure is in the store, and the local build found it there and had nothing to build.
|
||||
- `prebuild: the downloaded closure <path> is not the one this hive evaluates, <path>; built the latter locally`, at warn level: the builder and the hive evaluate different closures for the agent, which is worth a look. The download was no use and the rebuild built locally, with the agent still up.
|
||||
- `prebuild: built locally: <reason>`: the reason is the first check that failed,
|
||||
such as `no-record`, `rev-mismatch`, `not-cached`, or what `hive-priv` answered
|
||||
(`not available: …`, for a cache that's down or has no credentials). The rebuild builds as before.
|
||||
- The same line at error level, with `no signature by the swarm cache's key` in
|
||||
the reason: nix refused the path because no trusted key signed it. The rebuild builds locally
|
||||
and `hive-priv` logs the refusal as well.
|
||||
A rebuild or create has three outcomes, by what the lookup and the download find:
|
||||
|
||||
After a download, `Prebuild` still runs the local `nix build` of the agent's toplevel,
|
||||
while the agent is up. It's an evaluation and no build when the hive evaluates the path the
|
||||
swarm built, and a full build, as without a cache, when it doesn't, so `Swap` never meets a
|
||||
build it didn't expect. `Swap` still evaluates and builds in `hive-priv`, and finds the path in
|
||||
the store. A stopped agent skips `Prebuild`, so it still builds in `Swap`.
|
||||
| Found | What happens |
|
||||
| ----------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
|
||||
| a record that the builder pushed, for the commit this hive chose | The hive downloads the closure and deploys it. The hive logs `prebuild: substituted <path>` (`create: substituted <path>` for a create) at info level. |
|
||||
| a record that the builder pushed, for another commit (the hive and the builder disagree) | The hive downloads the builder's closure and deploys it, with a warning that names the agent, both commits and the path. The hive evaluates nothing, and records the pins of the commit the builder built the closure from, so the agent's markers say what runs. |
|
||||
| no record, or one the hive can't use (not pushed, queue or cache unreachable, download or signature failed) | The rebuild or create fails with an error that names the agent and the reason, logged at error level, and nothing changes. A running container keeps running unchanged, because `Prebuild` fails before `StopForUpdate`. A create makes no container. |
|
||||
|
||||
The hive never falls back to a local build. The third row has one owner in the
|
||||
code, `without_closure` in `hive-c0re/src/job_queue/exec.rs`, so a change to what
|
||||
an unreachable cache costs is one function.
|
||||
|
||||
Mixed versions:
|
||||
|
||||
| hive | controller or record | what happens |
|
||||
| ---- | -------------------- | ---------------------------------------------------------------------------------- |
|
||||
| old | any | builds as before |
|
||||
| new | old, or no record | builds locally and logs why |
|
||||
| new | cache down | the download fails, the hive builds locally and logs why |
|
||||
| new | signature mismatch | nix refuses the path, the hive builds locally and logs an error |
|
||||
| new | old `hive-priv` | the verb is unknown to it, so the call fails and the hive builds locally, as above |
|
||||
| hive | controller or record | what happens |
|
||||
| ---- | -------------------- | ------------------------------------------------------------------------------------------ |
|
||||
| old | any | builds locally, as before |
|
||||
| new | old, or no record | refuses the create; a running agent keeps running, and the rebuild logs an error naming it |
|
||||
| new | cache down | the download fails; same as no record, with the download's error in the log |
|
||||
| new | signature mismatch | nix refuses the path; same as no record, and `hive-priv` logs the refusal as well |
|
||||
| new | old `hive-priv` | it refuses the request, which has no `meta_rev`, so the swap or create fails |
|
||||
|
||||
## Credentials
|
||||
|
||||
|
|
|
|||
|
|
@ -156,7 +156,7 @@ pieces fit, and the reasoning behind them, is in [`credentials.md`](credentials.
|
|||
| the cache's public key and pull token, for hive-priv | `swarm-controller` copies them to `swarm/hives/<hive>/attic/{public-key,pull-token}` (see the cache-facts row above) | `hive-priv-cache-facts` (root) reads those two under the hive's own store identity every ten minutes and writes `/run/hyperhive-priv/attic/public-key` and `netrc` (tmpfs, root, each `0600`, directory `0700`); the cache's address comes from `services.hyperhive.swarm.attic.domain` and `cache`, not from the store. A store that holds no facts, or that doesn't answer, leaves the files as they stand, and a hive that never got them has `hive-priv` answer `not available`. |
|
||||
| the forge's copy of its OIDC secret | `swarm-bao-forge-oidc.service` reads it out of the swarm secret store under the forge publisher's leaf, wherever authelia runs | `/run/hive-forge-oidc/<id>.secret` on the host (tmpfs), bound into the forge container, handed to `forgejo-sso-source` through `LoadCredential` |
|
||||
| the homeserver's copy of its OIDC secret | `swarm-bao-matrix-oidc.service` reads it out of the swarm secret store under matrix-ctl's leaf, wherever authelia runs | `/run/hive-matrix-oidc/<id>.secret` on the host (tmpfs), bound into the homeserver container, handed to tuwunel through `LoadCredential` |
|
||||
| this hive's forge read token | `swarm-controller`, into the secret store at `swarm/hives/<hive>/forge/token`, for the `hive-<hive>` account; a pass at start and every five minutes re-mints a missing or stale token | `hive-c0re` reads it under the hive's own store identity, and copies it (account on the first line, token on the second) to `/run/hyperhive/forge-hive-token` (tmpfs, `0600` `hive-core`) for nix's `git`, which `hive-priv` also runs. Each `internal/meta` read and each boot rewrites the copy when it differs; where they differ after a rotation, a build that starts before the next read fails to authenticate, and a retry reads the new token |
|
||||
| this hive's forge read token | `swarm-controller`, into the secret store at `swarm/hives/<hive>/forge/token`, for the `hive-<hive>` account; a pass at start and every five minutes re-mints a missing or stale token | `hive-c0re` reads it under the hive's own store identity for each forge read, and hands it to the `git` commands it runs. It keeps no copy on disk |
|
||||
| squid's users file | `hive-c0re` (`workers::squid_users`), every 30 seconds, from the proxy credentials of its agents and the self-check | `services.hyperhive.deploy.hive-controller.squid.usersFile` (`/var/lib/hive-squid/users`, `hive-core:squid`, `0640`): `<user>:<hash>` lines, SHA-512 crypt hashes only, renamed into place when it changed. `unrestricted` beside it lists the users whose agent holds the swarm's `unrestricted_internet` flag, names only; it lists every agent the hive never read as off `systemd-tmpfiles` creates it empty so squid's helper never sees a missing file. A hive whose agents have no credential yet (an older controller) has an empty file: every login fails and the self-check reports `ok=false`; an agent without one sends none, so it goes direct |
|
||||
| the squid self-check's password | `swarm-controller` mints `swarm/hives/<hive>/squid-selfcheck`; `squid-selfcheck-credential` (root, every 10 minutes) reads it as the hive | `/run/credstore/hive-squid.selfcheck` (tmpfs, root, `0400`), imported by `squid-selfcheck.service` as a systemd credential and handed to `curl` on its stdin. Both self-check users (`selfcheck.squid` and the always-unrestricted `selfcheck-unrestricted.squid`) use it, so whoever holds it has unrestricted egress through this squid Missing, or different from the users file: the self-check logs `credential=absent` or a 407 and fails, and nothing else breaks |
|
||||
|
||||
|
|
|
|||
|
|
@ -252,34 +252,34 @@ for every operation that genuinely requires root.
|
|||
**Narrow interface** — `PrivRequest` variants map 1:1 to specific
|
||||
known operations; there is no arbitrary command pass-through:
|
||||
|
||||
| Operation | What it runs |
|
||||
| ------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
|
||||
| `StartContainer` / `StopContainer` | `nixos-container start/stop <name>` |
|
||||
| `KillContainer` | `machinectl kill <machine> --signal=SIGKILL` (`nixos-container` has no kill verb) |
|
||||
| `CreateContainer` / `UpdateContainer` | `nixos-container create/update <name> --flake <ref>` |
|
||||
| `RealiseClosure` | `nix-store --realise <store_path> --max-jobs 0` against the swarm cache, with the host's configured cache address, and the public key and netrc from root-owned `/run/hyperhive-priv/attic/`, as per-call settings that replace the host's substituters and trusted keys and require a signature. The path must be `/nix/store/<32 chars>-<name>` and the agent name valid. It downloads and never builds |
|
||||
| `DestroyContainer` | `nixos-container destroy <name>` |
|
||||
| `ListContainers` | `nixos-container list` |
|
||||
| `ReloadGatewayNginx` | `systemctl reload/start/reset-failed nginx` (host unit; the unit name is hard-coded, not a parameter) |
|
||||
| `WriteNspawnFlags` | write `/etc/nixos-containers/<container>.conf` (bind-mount list, network isolation vars, and `--ephemeral` after the root's salvage). hive-priv first copies a stopped container's root to the agent's `work/`, then renames it aside (it deletes nothing), following no link the container could have planted |
|
||||
| `EnsureAgentSocketDir` | `mkdirat` `/run/hive-agent/<name>` `0751 root:root` under an `O_NOFOLLOW` parent fd; leaves an existing directory alone, refuses anything else |
|
||||
| `WriteResourceLimits` | write `CPUQuota=`/`MemoryMax=`/`CPUWeight=`/`IOWeight=` systemd drop-in for agent container, with an `ExecStartPre=` that runs `hive-priv check-rootfs <container>` (no-follow check of the root paths the start script touches; see persistence.md), and a GC root on the build it names |
|
||||
| `RemoveServiceDropin` | remove `container@<name>.service.d/` drop-in on destroy, and the GC root of the hive-priv build it named |
|
||||
| `DaemonReload` | `systemctl daemon-reload` |
|
||||
| `ControlInfraContainer` | `systemctl <action> nginx.service` for the gateway — the `InfraContainer` enum is the allowlist, and serde rejects unknown names at the wire boundary (`hive-c0re`, the forge and the matrix have no variant, so no request can name them) |
|
||||
| `SyncAgentTmpfiles` | unlink `/etc/tmpfiles.d/hyperhive-agents.conf` and return `Ok` |
|
||||
| `SetAgentPaused` | create / remove the `<state>/<name>/harness/paused` marker that parks an agent's turn loop |
|
||||
| `EnsureAgentSubvolume` | `btrfs subvolume create <state>/<name>` for a new agent — no-op when the path exists or the filesystem isn't btrfs |
|
||||
| `UpgradeAgentSubvolume` | migrate an existing plain state dir into a subvolume: create, `cp -a --reflink=auto`, atomic swap. Operator opt-in, and the caller stops the agent first |
|
||||
| `DeleteAgentSubvolume` | `btrfs subvolume delete <state>/<name>` — purge path only, never a plain destroy |
|
||||
| `EnsureBtrfsQuota` | `btrfs quota enable <AGENT_STATE_ROOT>`. Operator opt-in — enabling forces a full rescan |
|
||||
| `ReadSubvolumeUsage` | `btrfs qgroup show -f --raw <state>/<name>` |
|
||||
| `SetSubvolumeQuota` | `btrfs qgroup limit <bytes\|none> <state>/<name>` |
|
||||
| `MeasureAgentDisk` | `du -s -x -b -c -P` over `<state>/<name>/state`, `<state>/<name>/work` and `/var/lib/nixos-containers/h-<name>`, killed after 10 min. hive-priv derives the paths from the name; the request carries none |
|
||||
| `SnapshotAgentSubvolume` | `btrfs subvolume snapshot -r <agent_root> <snapshot_path>` — the frozen point-in-time copy a migration streams from |
|
||||
| `DeleteAgentSnapshot` | `btrfs subvolume delete <snapshot_path>` |
|
||||
| `SendAgentSnapshotToFile` | `btrfs send [-p <parent>] <snapshot>` into a bare filename under `MIGRATE_STAGING_ROOT`; refuses to overwrite an existing export |
|
||||
| `SendAgentSnapshotToFd` | `btrfs send [-p <parent>] <snapshot>` into a **file descriptor passed with the request** (`SCM_RIGHTS`). hive-c0re connects to the peer and hands over the connected socket, so hive-priv never learns an address or protocol. hive-priv requires exactly one descriptor here, and refuses one that arrives alongside any other operation |
|
||||
| Operation | What it runs |
|
||||
| ------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
|
||||
| `StartContainer` / `StopContainer` | `nixos-container start/stop <name>` |
|
||||
| `KillContainer` | `machinectl kill <machine> --signal=SIGKILL` (`nixos-container` has no kill verb) |
|
||||
| `CreateContainer` / `UpdateContainer` | `nixos-container create <name> --system-path <toplevel>`; for an update, `nix-env -p <profile> --set <toplevel>` and a reload of the unit. `<toplevel>` must be the store path the last successful `RealiseClosure` returned for that agent (in memory, per agent), so hive-c0re can't name a path never realised for the agent; hive-priv builds nothing. The binding doesn't mean hive-c0re can't choose the closure: it can still ask hive-priv to download any swarm-signed path for any agent and then apply it. The swarm key and the cache bound that to a cross-assignment risk (one agent gets another's closure), not an arbitrary-path one |
|
||||
| `RealiseClosure` | `nix-store --realise <store_path> --max-jobs 0` against the swarm cache, with the host's configured cache address, and the public key and netrc from root-owned `/run/hyperhive-priv/attic/`, as per-call settings that replace the host's substituters and trusted keys and require a signature. The path must be `/nix/store/<32 chars>-<name>` and the agent name valid. It downloads and never builds |
|
||||
| `DestroyContainer` | `nixos-container destroy <name>` |
|
||||
| `ListContainers` | `nixos-container list` |
|
||||
| `ReloadGatewayNginx` | `systemctl reload/start/reset-failed nginx` (host unit; the unit name is hard-coded, not a parameter) |
|
||||
| `WriteNspawnFlags` | write `/etc/nixos-containers/<container>.conf` (bind-mount list, network isolation vars, and `--ephemeral` after the root's salvage). hive-priv first copies a stopped container's root to the agent's `work/`, then renames it aside (it deletes nothing), following no link the container could have planted |
|
||||
| `EnsureAgentSocketDir` | `mkdirat` `/run/hive-agent/<name>` `0751 root:root` under an `O_NOFOLLOW` parent fd; leaves an existing directory alone, refuses anything else |
|
||||
| `WriteResourceLimits` | write `CPUQuota=`/`MemoryMax=`/`CPUWeight=`/`IOWeight=` systemd drop-in for agent container, with an `ExecStartPre=` that runs `hive-priv check-rootfs <container>` (no-follow check of the root paths the start script touches; see persistence.md), and a GC root on the build it names |
|
||||
| `RemoveServiceDropin` | remove `container@<name>.service.d/` drop-in on destroy, and the GC root of the hive-priv build it named |
|
||||
| `DaemonReload` | `systemctl daemon-reload` |
|
||||
| `ControlInfraContainer` | `systemctl <action> nginx.service` for the gateway — the `InfraContainer` enum is the allowlist, and serde rejects unknown names at the wire boundary (`hive-c0re`, the forge and the matrix have no variant, so no request can name them) |
|
||||
| `SyncAgentTmpfiles` | unlink `/etc/tmpfiles.d/hyperhive-agents.conf` and return `Ok` |
|
||||
| `SetAgentPaused` | create / remove the `<state>/<name>/harness/paused` marker that parks an agent's turn loop |
|
||||
| `EnsureAgentSubvolume` | `btrfs subvolume create <state>/<name>` for a new agent — no-op when the path exists or the filesystem isn't btrfs |
|
||||
| `UpgradeAgentSubvolume` | migrate an existing plain state dir into a subvolume: create, `cp -a --reflink=auto`, atomic swap. Operator opt-in, and the caller stops the agent first |
|
||||
| `DeleteAgentSubvolume` | `btrfs subvolume delete <state>/<name>` — purge path only, never a plain destroy |
|
||||
| `EnsureBtrfsQuota` | `btrfs quota enable <AGENT_STATE_ROOT>`. Operator opt-in — enabling forces a full rescan |
|
||||
| `ReadSubvolumeUsage` | `btrfs qgroup show -f --raw <state>/<name>` |
|
||||
| `SetSubvolumeQuota` | `btrfs qgroup limit <bytes\|none> <state>/<name>` |
|
||||
| `MeasureAgentDisk` | `du -s -x -b -c -P` over `<state>/<name>/state`, `<state>/<name>/work` and `/var/lib/nixos-containers/h-<name>`, killed after 10 min. hive-priv derives the paths from the name; the request carries none |
|
||||
| `SnapshotAgentSubvolume` | `btrfs subvolume snapshot -r <agent_root> <snapshot_path>` — the frozen point-in-time copy a migration streams from |
|
||||
| `DeleteAgentSnapshot` | `btrfs subvolume delete <snapshot_path>` |
|
||||
| `SendAgentSnapshotToFile` | `btrfs send [-p <parent>] <snapshot>` into a bare filename under `MIGRATE_STAGING_ROOT`; refuses to overwrite an existing export |
|
||||
| `SendAgentSnapshotToFd` | `btrfs send [-p <parent>] <snapshot>` into a **file descriptor passed with the request** (`SCM_RIGHTS`). hive-c0re connects to the peer and hands over the connected socket, so hive-priv never learns an address or protocol. hive-priv requires exactly one descriptor here, and refuses one that arrives alongside any other operation |
|
||||
|
||||
**Closures from the swarm cache** — `RealiseClosure` is how a hive's rebuild
|
||||
takes the closure the swarm built instead of building it. Root-owned sources
|
||||
|
|
|
|||
|
|
@ -1,62 +1,19 @@
|
|||
//! The closure the swarm built for an agent at a given `internal/meta` commit,
|
||||
//! as a rebuild's `Prebuild` downloads it instead of building.
|
||||
//! The closure the swarm built for an agent, as a deploy reads its record
|
||||
//! before downloading it.
|
||||
|
||||
use anyhow::{Result, anyhow};
|
||||
use swarm_wire::agent_closures::AgentClosure;
|
||||
|
||||
use super::probe::{self, Unpullable};
|
||||
use super::probe;
|
||||
|
||||
/// The store path of the closure `agent`'s record names, when the record is a
|
||||
/// build of `rev` that the builder pushed to the cache.
|
||||
/// `agent`'s record in `agent-closures`; `None` when the builder has recorded
|
||||
/// no build of it.
|
||||
///
|
||||
/// # Errors
|
||||
/// No record, one for another commit, or one the builder did not push; or the
|
||||
/// queue cannot be reached. The message is what a deploy log says.
|
||||
pub async fn path_for(agent: &str, rev: &str) -> Result<String> {
|
||||
/// The queue cannot be reached, or the record cannot be read.
|
||||
pub async fn record(agent: &str) -> Result<Option<AgentClosure>> {
|
||||
let client = crate::swarm_queue::client()
|
||||
.await
|
||||
.ok_or_else(|| anyhow!("this hive has no connection to the swarm queue"))?;
|
||||
let record = probe::read_record(&client, agent).await?;
|
||||
path_in(record.as_ref(), rev)
|
||||
}
|
||||
|
||||
fn path_in(record: Option<&AgentClosure>, rev: &str) -> Result<String> {
|
||||
probe::current_record(record, Some(rev))
|
||||
.map(|record| record.store_path.clone())
|
||||
.map_err(|Unpullable { reason, detail }| anyhow!("{}: {detail}", reason.label()))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
const REV: &str = "a1b2c3";
|
||||
const PATH: &str = "/nix/store/0123456789abcdfghijklmnpqrsvwxyz-nixos-system-atlas";
|
||||
|
||||
fn record(rev: &str, cached: bool) -> AgentClosure {
|
||||
AgentClosure {
|
||||
meta_rev: rev.to_owned(),
|
||||
store_path: PATH.to_owned(),
|
||||
system: "x86_64-linux".to_owned(),
|
||||
built_unix: 1,
|
||||
cached,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_pushed_build_of_the_rev_names_its_path() {
|
||||
assert_eq!(path_in(Some(&record(REV, true)), REV).expect("path"), PATH);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn anything_else_names_why_not() {
|
||||
for (record, why) in [
|
||||
(None, "no-record"),
|
||||
(Some(record("other", true)), "rev-mismatch"),
|
||||
(Some(record(REV, false)), "not-cached"),
|
||||
] {
|
||||
let err = path_in(record.as_ref(), REV).expect_err(why);
|
||||
assert!(format!("{err}").starts_with(why), "{err}");
|
||||
}
|
||||
}
|
||||
probe::read_record(&client, agent).await
|
||||
}
|
||||
|
|
|
|||
|
|
@ -29,9 +29,8 @@ const LAST_STOPPED_RUNNING_KEY: &str = "last_stopped_running";
|
|||
|
||||
pub struct Coordinator {
|
||||
pub broker: Arc<Broker>,
|
||||
/// Full build-log capture. `lifecycle::run` /
|
||||
/// `lifecycle::prebuild_toplevel` `start()` a row per attempt,
|
||||
/// pipe every stdout/stderr line into it, and `finish()` it on
|
||||
/// Full build-log capture. `lifecycle::run` `start()`s a row per attempt,
|
||||
/// pipes every stdout/stderr line into it, and `finish()`es it on
|
||||
/// child exit. Dashboard reads it via `list_recent_for_agent` /
|
||||
/// `get_full` for the per-card chip + side-panel viewer. See
|
||||
/// `build_logs.rs` for retention.
|
||||
|
|
@ -228,7 +227,7 @@ pub struct ServeConfig {
|
|||
#[serde(flatten)]
|
||||
pub env: HiveEnv,
|
||||
pub model_prices: crate::hive_stats::PriceTable,
|
||||
/// Number of concurrent nix-heavy job-queue nodes (prebuild /
|
||||
/// Number of concurrent closure-fetching job-queue nodes (prebuild /
|
||||
/// profile-swap / create / meta lock). hive-c0re-local like
|
||||
/// `model_prices` — never injected into containers. Set via
|
||||
/// `services.hyperhive.deploy.hive-controller.buildSlots`.
|
||||
|
|
@ -374,9 +373,8 @@ impl Coordinator {
|
|||
"build_logs: closed rows left in-flight by a previous run"
|
||||
);
|
||||
}
|
||||
// Install the process-wide handle so `lifecycle::run` /
|
||||
// `lifecycle::prebuild_toplevel` can write without us having
|
||||
// to thread an `Arc<BuildLogs>` through every public entry
|
||||
// Install the process-wide handle so `lifecycle::run` can write
|
||||
// without us having to thread an `Arc<BuildLogs>` through every public entry
|
||||
// point in the lifecycle surface.
|
||||
crate::build_logs::install(build_logs.clone());
|
||||
let power = Arc::new(crate::power::PowerStore::open(db_path).context("open agent_power")?);
|
||||
|
|
|
|||
|
|
@ -20,20 +20,13 @@ const CONFIG_BRANCH: &str = "main";
|
|||
/// [`MetaSource`] over the forge API.
|
||||
pub(crate) struct ForgeSource {
|
||||
api: Forgejo,
|
||||
/// Whether the credential file nix's `git` reads holds this hive's token.
|
||||
nix_credential: bool,
|
||||
}
|
||||
|
||||
/// A source reading as this hive, or `None` when it holds no forge token. It
|
||||
/// also refreshes the credential file nix's `git` fetches with, which the
|
||||
/// build after this read needs to be current.
|
||||
/// A source reading as this hive, or `None` when it holds no forge token.
|
||||
pub(crate) async fn internal_meta_source() -> Option<ForgeSource> {
|
||||
let token: ReaderToken = super::reader_token().await?;
|
||||
match token.api() {
|
||||
Ok(api) => Some(ForgeSource {
|
||||
api,
|
||||
nix_credential: token.publish_for_nix(),
|
||||
}),
|
||||
Ok(api) => Some(ForgeSource { api }),
|
||||
Err(e) => {
|
||||
tracing::warn!(error = ?e, "internal/meta: cannot build a forge client");
|
||||
None
|
||||
|
|
@ -72,10 +65,6 @@ impl ForgeSource {
|
|||
}
|
||||
|
||||
impl MetaSource for ForgeSource {
|
||||
fn nix_can_fetch(&self) -> bool {
|
||||
self.nix_credential
|
||||
}
|
||||
|
||||
async fn meta_head(&self) -> anyhow::Result<Option<String>> {
|
||||
use anyhow::Context as _;
|
||||
|
||||
|
|
@ -113,25 +102,3 @@ impl MetaSource for ForgeSource {
|
|||
self.branch_head(CONFIG_ORG, agent, CONFIG_BRANCH).await
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use forgejo_api::Auth;
|
||||
|
||||
fn source(nix_credential: bool) -> ForgeSource {
|
||||
let url = url::Url::parse("http://forge.test").expect("url");
|
||||
ForgeSource {
|
||||
api: Forgejo::new(Auth::None, url).expect("client"),
|
||||
nix_credential,
|
||||
}
|
||||
}
|
||||
|
||||
/// The control (`true`) shows the answer follows the field rather than
|
||||
/// being a constant.
|
||||
#[test]
|
||||
fn nix_fetches_only_where_its_credential_was_published() {
|
||||
assert!(source(true).nix_can_fetch());
|
||||
assert!(!source(false).nix_can_fetch());
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
//! The hive's forge access: reads of the forge go through [`reader_token`] on
|
||||
//! every hive, and [`ensure_all`] publishes that token for nix's `git`. The
|
||||
//! every hive. The
|
||||
//! swarm-wide orgs, teams and repos, and each agent's read grant on
|
||||
//! `internal/docs`, are swarm-controller's. Full design:
|
||||
//! `docs/integrations/forge.md`.
|
||||
|
|
@ -10,7 +10,7 @@ mod reconcile;
|
|||
mod repos;
|
||||
|
||||
pub(crate) use internal_meta::internal_meta_source;
|
||||
pub use reader::{publish_nix_credential, reader_token};
|
||||
pub use reader::reader_token;
|
||||
pub use reconcile::{fetch_forge_main, reconcile_config_apply, reconcile_config_status};
|
||||
pub use repos::{clone_config_into_proposed, fetch_config_main_into_applied};
|
||||
|
||||
|
|
@ -145,16 +145,6 @@ where
|
|||
.map_err(|_| anyhow::anyhow!("{op}: no answer from the forge within {API_TIMEOUT:?}"))
|
||||
}
|
||||
|
||||
/// Publish this hive's forge read token for nix's `git` at startup, so a build
|
||||
/// that starts before any `internal/meta` read still has a credential. Every
|
||||
/// hive runs it: the token is the hive's own, in the store, whether or not a
|
||||
/// forge container is local.
|
||||
pub async fn ensure_all() {
|
||||
if publish_nix_credential().await {
|
||||
tracing::debug!("forge: nix's forge credential is current");
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{API_TIMEOUT, bounded, git_url_with_base};
|
||||
|
|
|
|||
|
|
@ -1,14 +1,6 @@
|
|||
//! The credential this hive reads the forge with: its own read-only account,
|
||||
//! whose token `swarm-controller` mints into the store at
|
||||
//! `swarm/hives/<hive>/forge/token`.
|
||||
//!
|
||||
//! nix's `git` cannot reach the store, so [`ReaderToken::publish_for_nix`]
|
||||
//! copies the token to a runtime file that `git-credential-hive-forge`
|
||||
//! (`nix/host-modules/lib/hive-forge-git.nix`) reads.
|
||||
|
||||
use std::path::Path;
|
||||
|
||||
use anyhow::Result;
|
||||
|
||||
use super::basic_auth_header;
|
||||
|
||||
|
|
@ -39,30 +31,6 @@ impl ReaderToken {
|
|||
pub(crate) fn api(&self) -> anyhow::Result<forgejo_api::Forgejo> {
|
||||
super::api(&self.value)
|
||||
}
|
||||
|
||||
/// Copy this token to the file nix's `git` authenticates from
|
||||
/// ([`crate::paths::FORGE_HIVE_TOKEN`]). Whether the file now holds it; a
|
||||
/// failed write is logged and leaves the file as it was.
|
||||
pub(crate) fn publish_for_nix(&self) -> bool {
|
||||
let path = Path::new(crate::paths::FORGE_HIVE_TOKEN);
|
||||
match self.write_credential(path) {
|
||||
Ok(()) => true,
|
||||
Err(e) => {
|
||||
tracing::warn!(path = %path.display(), error = ?e, "forge: could not write the nix forge credential");
|
||||
false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Login on the first line, token on the second. Written only when the
|
||||
/// content differs, so a build's resolve is not a disk write.
|
||||
fn write_credential(&self, path: &Path) -> Result<()> {
|
||||
let content = format!("{}\n{}\n", self.login, self.value);
|
||||
if std::fs::read(path).is_ok_and(|have| have == content.as_bytes()) {
|
||||
return Ok(());
|
||||
}
|
||||
crate::secret_file::write_secret_0600(path, content.as_bytes())
|
||||
}
|
||||
}
|
||||
|
||||
/// The token to read the forge with, or `None` when the store holds none for
|
||||
|
|
@ -77,16 +45,6 @@ pub async fn reader_token() -> Option<ReaderToken> {
|
|||
token
|
||||
}
|
||||
|
||||
/// Publish this hive's read token for nix; see [`ReaderToken::publish_for_nix`].
|
||||
/// Whether the file holds a token.
|
||||
pub async fn publish_nix_credential() -> bool {
|
||||
let Some(token) = reader_token().await else {
|
||||
tracing::debug!("forge: no hive read token in the store; nix has no forge credential");
|
||||
return false;
|
||||
};
|
||||
token.publish_for_nix()
|
||||
}
|
||||
|
||||
/// This hive's token from the store, under the hive's own identity.
|
||||
///
|
||||
/// `None`, never an error, for every way it can come up empty — no hive name,
|
||||
|
|
@ -131,8 +89,6 @@ async fn stored_reader_token() -> Option<ReaderToken> {
|
|||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::os::unix::fs::PermissionsExt as _;
|
||||
|
||||
fn stored() -> ReaderToken {
|
||||
ReaderToken {
|
||||
login: "hive-alpha".to_owned(),
|
||||
|
|
@ -140,73 +96,6 @@ mod tests {
|
|||
}
|
||||
}
|
||||
|
||||
fn mode(path: &Path) -> u32 {
|
||||
std::fs::metadata(path).expect("stat").permissions().mode() & 0o777
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_nix_credential_is_the_hives_login_then_its_token_at_0600() {
|
||||
let dir = tempfile::tempdir().expect("tempdir");
|
||||
let path = dir.path().join("forge-hive-token");
|
||||
stored().write_credential(&path).expect("written");
|
||||
assert_eq!(
|
||||
std::fs::read_to_string(&path).expect("read"),
|
||||
"hive-alpha\nfrom-the-store\n"
|
||||
);
|
||||
assert_eq!(mode(&path), 0o600);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_rotated_token_replaces_the_nix_credential() {
|
||||
let dir = tempfile::tempdir().expect("tempdir");
|
||||
let path = dir.path().join("forge-hive-token");
|
||||
stored().write_credential(&path).expect("first");
|
||||
let rotated = ReaderToken {
|
||||
login: "hive-alpha".to_owned(),
|
||||
value: "rotated".to_owned(),
|
||||
};
|
||||
rotated.write_credential(&path).expect("second");
|
||||
assert_eq!(
|
||||
std::fs::read_to_string(&path).expect("read"),
|
||||
"hive-alpha\nrotated\n"
|
||||
);
|
||||
assert_eq!(mode(&path), 0o600);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_unchanged_credential_is_not_rewritten() {
|
||||
let dir = tempfile::tempdir().expect("tempdir");
|
||||
let path = dir.path().join("forge-hive-token");
|
||||
stored().write_credential(&path).expect("first");
|
||||
// A rewrite goes through a sibling temp file and a rename, which
|
||||
// gives the file a new inode.
|
||||
let before = std::fs::metadata(&path).expect("stat");
|
||||
stored().write_credential(&path).expect("second");
|
||||
let after = std::fs::metadata(&path).expect("stat");
|
||||
assert_eq!(
|
||||
std::os::unix::fs::MetadataExt::ino(&before),
|
||||
std::os::unix::fs::MetadataExt::ino(&after)
|
||||
);
|
||||
}
|
||||
|
||||
/// `hive-forge-git.nix`'s credential helper names this path as a literal.
|
||||
#[test]
|
||||
fn the_nix_credential_lives_on_the_runtime_tmpfs_where_the_helper_reads_it() {
|
||||
assert_eq!(
|
||||
crate::paths::FORGE_HIVE_TOKEN,
|
||||
"/run/hyperhive/forge-hive-token"
|
||||
);
|
||||
assert!(crate::paths::FORGE_HIVE_TOKEN.starts_with(crate::paths::RUNTIME_ROOT));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_path_that_cannot_be_written_is_an_error_not_a_panic() {
|
||||
let dir = tempfile::tempdir().expect("tempdir");
|
||||
let file = dir.path().join("a-file");
|
||||
std::fs::write(&file, "").expect("file");
|
||||
assert!(stored().write_credential(&file.join("sub")).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_header_names_the_tokens_own_account() {
|
||||
use base64::Engine as _;
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ use std::path::PathBuf;
|
|||
use std::sync::{Arc, Mutex, OnceLock};
|
||||
|
||||
use anyhow::{Context as _, Result};
|
||||
use swarm_wire::agent_closures::AgentClosure;
|
||||
|
||||
use hive_jobq::NodeId;
|
||||
|
||||
|
|
@ -34,7 +35,7 @@ pub const GRACEFUL_STOP_TIMEOUT: std::time::Duration = std::time::Duration::from
|
|||
const PAUSE_ACK_TIMEOUT: std::time::Duration = GRACEFUL_STOP_TIMEOUT;
|
||||
|
||||
/// The container-state calls the executors below make on the host: systemd,
|
||||
/// `hive-priv`, the nix build, the `internal/meta` record. Production uses
|
||||
/// `hive-priv`, the swarm's closure record, the `internal/meta` record. Production uses
|
||||
/// [`RealHost`], which forwards to `lifecycle` / `meta_rev` / `Coordinator`
|
||||
/// unchanged; tests substitute a recording fake.
|
||||
pub(super) trait Host: Sync {
|
||||
|
|
@ -44,28 +45,21 @@ pub(super) trait Host: Sync {
|
|||
async fn kill(&self, name: &str) -> Result<()>;
|
||||
/// The `internal/meta` commit `MetaSync` recorded for the agent.
|
||||
fn recorded_rev(&self, name: &str) -> Result<String>;
|
||||
/// The store path of the closure the swarm built for the agent at the
|
||||
/// `internal/meta` commit `rev` and pushed to its cache. The error says why
|
||||
/// there is none.
|
||||
async fn cached_closure(&self, name: &str, rev: &str) -> Result<String>;
|
||||
/// The record the swarm's builder wrote for the agent; `None` when it wrote
|
||||
/// none.
|
||||
async fn closure_record(&self, name: &str) -> Result<Option<AgentClosure>>;
|
||||
/// The pins `internal/meta` at the commit `rev` holds for the agent.
|
||||
async fn pins_at(&self, name: &str, rev: &str) -> Result<MetaBuild>;
|
||||
/// Download `store_path` from the swarm's cache into the store, never
|
||||
/// building it.
|
||||
async fn realise_closure(&self, name: &str, store_path: &str) -> Result<()>;
|
||||
/// Warm the agent's toplevel build from the `internal/meta` commit `rev`.
|
||||
/// The toplevel it printed, when it printed one.
|
||||
async fn prebuild_toplevel(
|
||||
&self,
|
||||
name: &str,
|
||||
rev: &str,
|
||||
node_id: u64,
|
||||
) -> Result<Option<String>>;
|
||||
/// Re-apply drop-ins, then `nixos-container update` at `rev`.
|
||||
/// Re-apply drop-ins, then point the container at the closure `toplevel`.
|
||||
async fn swap_update(
|
||||
&self,
|
||||
name: &str,
|
||||
hive: &HiveEnv,
|
||||
paths: &AgentPaths,
|
||||
rev: &str,
|
||||
toplevel: &str,
|
||||
node_id: u64,
|
||||
) -> Result<()>;
|
||||
/// Write or remove the agent's pause marker.
|
||||
|
|
@ -80,7 +74,8 @@ pub(super) trait Host: Sync {
|
|||
async fn refresh_agent_set(&self) -> Result<()>;
|
||||
/// Pin the `internal/meta` commit the agent builds from.
|
||||
async fn resolve_meta(&self, name: &str) -> Result<MetaBuild>;
|
||||
/// Record `build` as what `Prebuild` and `Swap` build from.
|
||||
/// Record `build` as the commit `Prebuild` and `Swap` expect the swarm's
|
||||
/// closure to be built from.
|
||||
fn record_meta(&self, name: &str, build: &MetaBuild) -> Result<()>;
|
||||
/// The build `MetaSync` recorded, if any.
|
||||
fn recorded_build(&self, name: &str) -> Option<MetaBuild>;
|
||||
|
|
@ -88,8 +83,8 @@ pub(super) trait Host: Sync {
|
|||
fn mark_deployed(&self, name: &str, build: &MetaBuild) -> Result<()>;
|
||||
/// First-spawn provisioning: repos, state subvolume, agent-set refresh.
|
||||
async fn provision(&self, name: &str, paths: &AgentPaths) -> Result<()>;
|
||||
/// `nixos-container create` at `rev`.
|
||||
async fn create_only(&self, name: &str, rev: &str) -> Result<()>;
|
||||
/// `nixos-container create` from the closure `toplevel`.
|
||||
async fn create_only(&self, name: &str, toplevel: &str) -> Result<()>;
|
||||
/// Write the nspawn and resource-limit drop-ins, then reload systemd.
|
||||
async fn write_dropins(&self, name: &str, hive: &HiveEnv, paths: &AgentPaths) -> Result<()>;
|
||||
/// Start preamble, then start with the cold-start fallback.
|
||||
|
|
@ -104,8 +99,6 @@ pub(super) trait Host: Sync {
|
|||
async fn delete_state_subvolume(&self, name: &str) -> Result<()>;
|
||||
/// The persistent directories a purge removes.
|
||||
fn purge_dirs(&self, name: &str) -> Vec<PathBuf>;
|
||||
/// Publish the hive's forge credential for nix.
|
||||
async fn forge_ensure_all(&self);
|
||||
/// One `/knowledge` pull.
|
||||
async fn knowledge_pull(&self, coord: &Arc<Coordinator>) -> Result<()>;
|
||||
/// One swarm wanted-state pull.
|
||||
|
|
@ -128,33 +121,27 @@ impl Host for RealHost {
|
|||
Ok(crate::meta_rev::recorded_or_err(name)?.rev)
|
||||
}
|
||||
|
||||
async fn cached_closure(&self, name: &str, rev: &str) -> Result<String> {
|
||||
crate::attic::closure::path_for(name, rev).await
|
||||
async fn closure_record(&self, name: &str) -> Result<Option<AgentClosure>> {
|
||||
crate::attic::closure::record(name).await
|
||||
}
|
||||
|
||||
async fn pins_at(&self, name: &str, rev: &str) -> Result<MetaBuild> {
|
||||
crate::meta_rev::resolve_at(name, rev).await
|
||||
}
|
||||
|
||||
async fn realise_closure(&self, name: &str, store_path: &str) -> Result<()> {
|
||||
crate::priv_client::realise_closure(name, store_path).await
|
||||
}
|
||||
|
||||
async fn prebuild_toplevel(
|
||||
&self,
|
||||
name: &str,
|
||||
rev: &str,
|
||||
node_id: u64,
|
||||
) -> Result<Option<String>> {
|
||||
let flake_ref = crate::meta_rev::flake_ref(name, rev);
|
||||
crate::lifecycle::prebuild_toplevel(name, &flake_ref, Some(node_id)).await
|
||||
}
|
||||
|
||||
async fn swap_update(
|
||||
&self,
|
||||
name: &str,
|
||||
hive: &HiveEnv,
|
||||
paths: &AgentPaths,
|
||||
rev: &str,
|
||||
toplevel: &str,
|
||||
node_id: u64,
|
||||
) -> Result<()> {
|
||||
crate::lifecycle::swap_update(name, hive, paths, rev, Some(node_id)).await
|
||||
crate::lifecycle::swap_update(name, hive, paths, toplevel, Some(node_id)).await
|
||||
}
|
||||
|
||||
async fn set_paused(&self, agent: &hive_types::Ident, paused: bool) -> Result<()> {
|
||||
|
|
@ -198,8 +185,8 @@ impl Host for RealHost {
|
|||
crate::lifecycle::provision_container(name, paths).await
|
||||
}
|
||||
|
||||
async fn create_only(&self, name: &str, rev: &str) -> Result<()> {
|
||||
crate::lifecycle::create_only(name, rev).await
|
||||
async fn create_only(&self, name: &str, toplevel: &str) -> Result<()> {
|
||||
crate::lifecycle::create_only(name, toplevel).await
|
||||
}
|
||||
|
||||
async fn write_dropins(&self, name: &str, hive: &HiveEnv, paths: &AgentPaths) -> Result<()> {
|
||||
|
|
@ -239,10 +226,6 @@ impl Host for RealHost {
|
|||
.collect()
|
||||
}
|
||||
|
||||
async fn forge_ensure_all(&self) {
|
||||
crate::forge::ensure_all().await;
|
||||
}
|
||||
|
||||
async fn knowledge_pull(&self, coord: &Arc<Coordinator>) -> Result<()> {
|
||||
crate::workers::knowledge::pull(coord).await
|
||||
}
|
||||
|
|
@ -300,11 +283,11 @@ async fn dispatch(
|
|||
// growth executors return what to grow rather than taking the builder.
|
||||
let result = match kind {
|
||||
NodeKind::MetaSync { .. } => run_meta_sync(host, agent).await,
|
||||
NodeKind::Prebuild { .. } => run_prebuild(host, agent, id).await,
|
||||
NodeKind::Prebuild { .. } => run_prebuild(host, agent).await,
|
||||
NodeKind::Swap { .. } => run_swap(host, coord, agent, id).await,
|
||||
NodeKind::RebuildBookkeeping { .. } => run_rebuild_bookkeeping(host, coord, agent).await,
|
||||
NodeKind::Provision { .. } => run_provision(host, agent).await,
|
||||
NodeKind::Create { .. } => run_create(host, agent, id).await,
|
||||
NodeKind::Create { .. } => run_create(host, agent).await,
|
||||
NodeKind::DestroyContainer { .. } => run_destroy_container(host, coord, agent).await,
|
||||
NodeKind::PurgeState { .. } => {
|
||||
run_purge_state(host, agent).await;
|
||||
|
|
@ -343,23 +326,12 @@ async fn dispatch(
|
|||
// `Finishing` so the nodes under it start. What they declare stays held
|
||||
// until their whole subtree settles.
|
||||
NodeKind::AgentWindow { .. } => Ok(()),
|
||||
NodeKind::ForgeSweep => run_forge_sweep(host).await,
|
||||
NodeKind::KnowledgePull => run_knowledge_pull(host, coord, KNOWLEDGE_PULL_DEADLINE).await,
|
||||
NodeKind::WantedPull => run_wanted_pull(host, coord).await,
|
||||
};
|
||||
(builder, result)
|
||||
}
|
||||
|
||||
/// Boot-time publish of the hive's forge credential for nix, as a DAG node —
|
||||
/// see [`NodeKind::ForgeSweep`]. `forge::ensure_all` logs its own outcome and
|
||||
/// is best-effort (a build re-publishes the credential before it fetches), so
|
||||
/// this wrapper has nothing left to report; it exists to make the step a
|
||||
/// visible unit of work.
|
||||
async fn run_forge_sweep(host: &impl Host) -> Result<()> {
|
||||
host.forge_ensure_all().await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// `/knowledge` pull as a DAG node — see [`NodeKind::KnowledgePull`]. Every
|
||||
/// pass reports here, whoever submitted it: as the node's own error on the
|
||||
/// dashboard, and into the debounced warning.
|
||||
|
|
@ -533,10 +505,11 @@ fn run_set_wanted(coord: &Arc<Coordinator>, agent: &str, up: bool) -> Result<()>
|
|||
}
|
||||
|
||||
/// The rebuild's preamble: runtime-dir prep, the agent-set refresh, and the
|
||||
/// `internal/meta` commit `Prebuild` and `Swap` build from.
|
||||
/// `internal/meta` commit `Prebuild` and `Swap` expect the swarm's closure to
|
||||
/// be built from.
|
||||
///
|
||||
/// Fails when `internal/meta` does not pin the agent's current config yet:
|
||||
/// there is no other flake to build it from.
|
||||
/// there is no commit to deploy.
|
||||
async fn run_meta_sync(host: &impl Host, name: &str) -> 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
|
||||
|
|
@ -550,98 +523,129 @@ async fn run_meta_sync(host: &impl Host, name: &str) -> Result<()> {
|
|||
}
|
||||
host.prepare_rebuild_dirs(name, &paths).await?;
|
||||
host.refresh_agent_set().await?;
|
||||
// Resolved once here so `Prebuild` and `Swap` build the same flake.
|
||||
// Resolved once here so `Prebuild` and `Swap` expect the same commit.
|
||||
let build = host.resolve_meta(name).await?;
|
||||
host.record_meta(name, &build)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Out-of-band toplevel build while the container keeps serving: warm
|
||||
/// `system.build.toplevel` so the later `Swap` hits cache and skips
|
||||
/// straight to the profile-swap, from the `internal/meta` commit the upstream
|
||||
/// `MetaSync` node recorded. The warm build is skipped when the
|
||||
/// 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.
|
||||
///
|
||||
/// The swarm's cache is tried first: when the swarm built this agent at the
|
||||
/// recorded commit and pushed the closure, it is downloaded. The local build
|
||||
/// runs either way. After a download it is an evaluation and a no-op build when
|
||||
/// this hive evaluates the path the swarm built, and builds it, with the agent
|
||||
/// still up, when it doesn't; `Swap` never meets a build it didn't expect.
|
||||
async fn run_prebuild(host: &impl Host, 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.
|
||||
/// Download the swarm's closure for the agent while the container keeps
|
||||
/// serving, so a rebuild with no closure fails here, before `StopForUpdate`
|
||||
/// takes a healthy container down. `Swap` downloads the same closure again
|
||||
/// (a no-op once it is in the store) and applies it. Skipped when the container
|
||||
/// is already down: there is no uptime to protect, and `Swap` fails the same way.
|
||||
async fn run_prebuild(host: &impl Host, name: &str) -> Result<()> {
|
||||
if host.is_running(name).await {
|
||||
let rev = host.recorded_rev(name)?;
|
||||
let downloaded = match substitute(host, name, &rev).await {
|
||||
Ok(path) => Some(path),
|
||||
Err(fallback) => {
|
||||
let reason = &fallback.reason;
|
||||
if fallback.signature_refused {
|
||||
tracing::error!(%name, %rev, "prebuild: built locally: {reason}");
|
||||
} else {
|
||||
tracing::warn!(%name, %rev, "prebuild: built locally: {reason}");
|
||||
}
|
||||
None
|
||||
}
|
||||
};
|
||||
let evaluated = host.prebuild_toplevel(name, &rev, id.get()).await?;
|
||||
if let Some(path) = downloaded {
|
||||
match drift(&path, evaluated.as_deref()) {
|
||||
None => tracing::info!(%name, %rev, "prebuild: substituted {path}"),
|
||||
Some(note) => tracing::warn!(%name, %rev, "prebuild: {note}"),
|
||||
}
|
||||
}
|
||||
let closure = closure_for(host, name, &rev, "prebuild").await?;
|
||||
tracing::info!(%name, %rev, "prebuild: substituted {}", closure.path);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// What to log when the closure downloaded for a rebuild isn't the one this
|
||||
/// hive's evaluation produced, which means the builder and the hive disagree
|
||||
/// about the agent's flake. `None` when they agree.
|
||||
fn drift(downloaded: &str, evaluated: Option<&str>) -> Option<String> {
|
||||
match evaluated {
|
||||
Some(path) if path == downloaded => None,
|
||||
Some(path) => Some(format!(
|
||||
"the downloaded closure {downloaded} is not the one this hive evaluates, {path}; built the latter locally"
|
||||
)),
|
||||
None => Some(format!(
|
||||
"downloaded {downloaded}, but the local build printed no path to compare it with"
|
||||
)),
|
||||
}
|
||||
/// Why the swarm's closure for an agent cannot be deployed.
|
||||
enum NoClosure {
|
||||
/// The builder has recorded no build of the agent.
|
||||
NoRecord,
|
||||
/// A build is recorded, but its closure cannot be had: the record could not
|
||||
/// be read, the builder did not push the closure, or the download failed.
|
||||
Unpullable(String),
|
||||
}
|
||||
|
||||
/// Why a rebuild or create builds instead of downloading the swarm's closure.
|
||||
struct LocalBuild {
|
||||
reason: String,
|
||||
/// The cache served a path no trusted key signed.
|
||||
signature_refused: bool,
|
||||
/// The closure the swarm built for an agent, in the store.
|
||||
struct Closure {
|
||||
path: String,
|
||||
/// The `internal/meta` commit the builder built it from.
|
||||
meta_rev: String,
|
||||
}
|
||||
|
||||
/// Download the closure the swarm built for `name` at `rev`; its store path.
|
||||
async fn substitute(host: &impl Host, name: &str, rev: &str) -> Result<String, LocalBuild> {
|
||||
let path = host
|
||||
.cached_closure(name, rev)
|
||||
/// The closure the swarm built for `name`, downloaded into the store. The
|
||||
/// deploy applies the swarm's closure whatever commit it was built from: when
|
||||
/// that is not the `rev` this hive chose, it is logged, and the caller records
|
||||
/// the pins of the commit actually deployed ([`deployed_pins`]).
|
||||
///
|
||||
/// # Errors
|
||||
/// [`without_closure`]'s refusal, after logging it.
|
||||
async fn closure_for(host: &impl Host, name: &str, rev: &str, step: &str) -> Result<Closure> {
|
||||
pull_closure(host, name, rev, step)
|
||||
.await
|
||||
.map_err(|e| LocalBuild {
|
||||
reason: format!("{e:#}"),
|
||||
signature_refused: false,
|
||||
})?;
|
||||
host.realise_closure(name, &path).await.map_err(|e| {
|
||||
let reason = format!("{e:#}");
|
||||
LocalBuild {
|
||||
signature_refused: reason.contains(hive_priv_sock::SIGNATURE_REFUSED),
|
||||
reason,
|
||||
}
|
||||
})?;
|
||||
Ok(path)
|
||||
.map_err(|why| without_closure(name, rev, step, &why))
|
||||
}
|
||||
|
||||
/// Profile-swap: re-apply drop-ins (rebuild is the reconcile verb),
|
||||
/// `nixos-container update`, then the post-rebuild bookkeeping tail
|
||||
async fn pull_closure(
|
||||
host: &impl Host,
|
||||
name: &str,
|
||||
rev: &str,
|
||||
step: &str,
|
||||
) -> Result<Closure, NoClosure> {
|
||||
let record = host
|
||||
.closure_record(name)
|
||||
.await
|
||||
.map_err(|e| NoClosure::Unpullable(format!("{e:#}")))?
|
||||
.ok_or(NoClosure::NoRecord)?;
|
||||
if !record.cached {
|
||||
return Err(NoClosure::Unpullable(
|
||||
"the builder did not push the closure".to_owned(),
|
||||
));
|
||||
}
|
||||
host.realise_closure(name, &record.store_path)
|
||||
.await
|
||||
.map_err(|e| NoClosure::Unpullable(format!("{e:#}")))?;
|
||||
if record.meta_rev != rev {
|
||||
tracing::warn!(
|
||||
%name, %rev, built_from = %record.meta_rev,
|
||||
"{step}: the swarm built {} from internal/meta {}, not the {rev} this hive chose; deploying the swarm's",
|
||||
record.store_path, record.meta_rev
|
||||
);
|
||||
}
|
||||
Ok(Closure {
|
||||
path: record.store_path,
|
||||
meta_rev: record.meta_rev,
|
||||
})
|
||||
}
|
||||
|
||||
/// The pins to record for a deploy of `closure` when this hive chose the commit
|
||||
/// `rev`: those of the commit the closure was built from, so the markers say
|
||||
/// what runs. `None` when they are the same commit.
|
||||
///
|
||||
/// # Errors
|
||||
/// The pins of the closure's commit could not be read. The caller fails before
|
||||
/// applying anything.
|
||||
async fn deployed_pins(
|
||||
host: &impl Host,
|
||||
name: &str,
|
||||
closure: &Closure,
|
||||
rev: &str,
|
||||
) -> Result<Option<MetaBuild>> {
|
||||
if closure.meta_rev == rev {
|
||||
return Ok(None);
|
||||
}
|
||||
let pins = host
|
||||
.pins_at(name, &closure.meta_rev)
|
||||
.await
|
||||
.with_context(|| format!("{name}: reading the pins of the commit the swarm built from"))?;
|
||||
Ok(Some(pins))
|
||||
}
|
||||
|
||||
/// What a rebuild or create does when there is no closure to deploy: it
|
||||
/// fails, and nothing changes, so a running container keeps running as it is
|
||||
/// and a create makes none. The failure is logged at error level with the
|
||||
/// agent's name.
|
||||
///
|
||||
/// This is the one place that decides what an unpullable closure (the cache
|
||||
/// unreachable, a TLS error, a failed download) costs. A rebuild never builds
|
||||
/// locally on the hive.
|
||||
fn without_closure(name: &str, rev: &str, step: &str, why: &NoClosure) -> anyhow::Error {
|
||||
let reason = match why {
|
||||
NoClosure::NoRecord => "the swarm has recorded no build of this agent".to_owned(),
|
||||
NoClosure::Unpullable(e) => format!("the swarm's closure cannot be pulled: {e}"),
|
||||
};
|
||||
tracing::error!(%name, %rev, "{step}: {reason}; nothing was changed");
|
||||
anyhow::anyhow!("{name}: {reason}")
|
||||
}
|
||||
|
||||
/// Profile-swap: re-apply drop-ins (rebuild is the reconcile verb), point the
|
||||
/// container at the swarm's closure, then the post-rebuild bookkeeping tail
|
||||
/// (rev marker, `Rebuilt` event, forge sync, kick, rescan).
|
||||
/// The recovery-start on failure is NOT here — the DAG's tail
|
||||
/// `Reconcile` runs after this node terminal ok *or* fail.
|
||||
|
|
@ -657,7 +661,19 @@ async fn run_swap(
|
|||
let hive = coord.hive_env();
|
||||
let paths = Coordinator::agent_paths(name, agent_dir)?;
|
||||
let rev = host.recorded_rev(name)?;
|
||||
let result = host.swap_update(name, &hive, &paths, &rev, id.get()).await;
|
||||
let closure = closure_for(host, name, &rev, "swap").await?;
|
||||
let deployed = deployed_pins(host, name, &closure, &rev).await?;
|
||||
tracing::info!(%name, %rev, "swap: applying {}", closure.path);
|
||||
let result = host
|
||||
.swap_update(name, &hive, &paths, &closure.path, id.get())
|
||||
.await;
|
||||
// The bookkeeping node marks what the record holds, so it holds what was
|
||||
// applied.
|
||||
if result.is_ok()
|
||||
&& let Some(deployed) = &deployed
|
||||
{
|
||||
host.record_meta(name, deployed)?;
|
||||
}
|
||||
// On success the Ok-only bookkeeping tail (rev marker, forge
|
||||
// sync, kick, rescan, snapshot) runs in the sibling `RebuildBookkeeping` node,
|
||||
// which deps `AfterOk(Swap)`. On failure `RebuildBookkeeping` is cancel-cascaded
|
||||
|
|
@ -719,30 +735,16 @@ async fn run_provision(host: &impl Host, name: &str) -> Result<()> {
|
|||
/// node stays purely "create", not "create + start". Marks the agent as
|
||||
/// running the pins it was created from.
|
||||
///
|
||||
/// The swarm's cache is tried first, as in [`run_prebuild`]: when it holds the
|
||||
/// closure for the resolved commit it is downloaded, and the local evaluation
|
||||
/// that follows is a no-op build on a path match and builds the evaluated path
|
||||
/// on a mismatch. Without a cached closure the create builds it locally.
|
||||
async fn run_create(host: &impl Host, name: &str, id: NodeId) -> Result<()> {
|
||||
/// The container is created from the swarm's closure for the resolved commit;
|
||||
/// with none to deploy the create fails and makes no container.
|
||||
async fn run_create(host: &impl Host, name: &str) -> Result<()> {
|
||||
let build = host.resolve_meta(name).await?;
|
||||
match substitute(host, name, &build.rev).await {
|
||||
Ok(path) => {
|
||||
let evaluated = host.prebuild_toplevel(name, &build.rev, id.get()).await?;
|
||||
match drift(&path, evaluated.as_deref()) {
|
||||
None => tracing::info!(%name, rev = %build.rev, "create: substituted {path}"),
|
||||
Some(note) => tracing::warn!(%name, rev = %build.rev, "create: {note}"),
|
||||
}
|
||||
}
|
||||
Err(fallback) => {
|
||||
let reason = &fallback.reason;
|
||||
if fallback.signature_refused {
|
||||
tracing::error!(%name, rev = %build.rev, "create: built locally: {reason}");
|
||||
} else {
|
||||
tracing::warn!(%name, rev = %build.rev, "create: built locally: {reason}");
|
||||
}
|
||||
}
|
||||
}
|
||||
host.create_only(name, &build.rev).await?;
|
||||
let closure = closure_for(host, name, &build.rev, "create").await?;
|
||||
let build = deployed_pins(host, name, &closure, &build.rev)
|
||||
.await?
|
||||
.unwrap_or(build);
|
||||
tracing::info!(%name, rev = %build.rev, "create: substituted {}", closure.path);
|
||||
host.create_only(name, &closure.path).await?;
|
||||
host.mark_deployed(name, &build)?;
|
||||
Ok(())
|
||||
}
|
||||
|
|
|
|||
|
|
@ -11,16 +11,17 @@ use hive_jobq::NodeId;
|
|||
use hive_jobq::scheduler::Scheduler;
|
||||
|
||||
use super::{
|
||||
Host, dispatch, drift, run_create, run_destroy_bookkeeping, run_destroy_container, run_drain,
|
||||
run_emit_rebuilt, run_forge_sweep, run_knowledge_pull, run_meta_sync, run_pause_drain,
|
||||
run_pause_signal, run_prebuild, run_provision, run_purge_state, run_rebuild_bookkeeping,
|
||||
run_reconcile, run_set_wanted, run_signal, run_start, run_stop, run_stop_for_update, run_swap,
|
||||
run_wanted_pull, run_write_dropin, substitute,
|
||||
Host, dispatch, run_create, run_destroy_bookkeeping, run_destroy_container, run_drain,
|
||||
run_emit_rebuilt, run_knowledge_pull, run_meta_sync, run_pause_drain, run_pause_signal,
|
||||
run_prebuild, run_provision, run_purge_state, run_rebuild_bookkeeping, run_reconcile,
|
||||
run_set_wanted, run_signal, run_start, run_stop, run_stop_for_update, run_swap,
|
||||
run_wanted_pull, run_write_dropin,
|
||||
};
|
||||
use crate::coordinator::{AgentPaths, Coordinator, HiveEnv};
|
||||
use crate::job_queue::model::NodeKind;
|
||||
use crate::meta_rev::MetaBuild;
|
||||
use crate::power::Wanted;
|
||||
use swarm_wire::agent_closures::AgentClosure;
|
||||
|
||||
const REV: &str = "2222222222222222222222222222222222222222";
|
||||
const CLOSURE: &str = "/nix/store/0123456789abcdfghijklmnpqrsvwxyz-nixos-system-alice";
|
||||
|
|
@ -30,18 +31,12 @@ const CLOSURE: &str = "/nix/store/0123456789abcdfghijklmnpqrsvwxyz-nixos-system-
|
|||
struct FakeHost {
|
||||
running: bool,
|
||||
rev: Option<String>,
|
||||
/// The store path the swarm's record names, when it has a pushed build of
|
||||
/// `REV`; `None` reads as no record.
|
||||
closure: Option<String>,
|
||||
/// A failing `realise` says nix refused the path for want of a signature.
|
||||
refusal: bool,
|
||||
/// The toplevel this hive's evaluation produces; `None` is `CLOSURE`.
|
||||
evaluates: Option<String>,
|
||||
/// What the swarm's builder recorded for the agent; `None` reads as no
|
||||
/// record.
|
||||
built: Option<AgentClosure>,
|
||||
/// Store paths a download put in the store.
|
||||
realised: Mutex<Vec<String>>,
|
||||
/// Store paths a prebuild had to build because the store lacked them.
|
||||
builds: Mutex<Vec<String>>,
|
||||
/// Names of the calls that fail (`kill`, `realise`, `prebuild`, `swap_update`,
|
||||
/// Names of the calls that fail (`kill`, `realise`, `closure_record`, `swap_update`,
|
||||
/// `set_paused`, ...). `recorded_build` fails by finding nothing.
|
||||
failing: Vec<&'static str>,
|
||||
/// The sweeps and pulls never finish, so only a deadline ends them.
|
||||
|
|
@ -56,6 +51,7 @@ impl FakeHost {
|
|||
Self {
|
||||
running: true,
|
||||
rev: Some(REV.to_owned()),
|
||||
built: Some(closure(REV, true)),
|
||||
..Self::default()
|
||||
}
|
||||
}
|
||||
|
|
@ -64,12 +60,19 @@ impl FakeHost {
|
|||
Self {
|
||||
running: false,
|
||||
rev: Some(REV.to_owned()),
|
||||
built: Some(closure(REV, true)),
|
||||
..Self::default()
|
||||
}
|
||||
}
|
||||
|
||||
fn with_closure(mut self) -> Self {
|
||||
self.closure = Some(CLOSURE.to_owned());
|
||||
/// The builder recorded `meta_rev` for the agent, pushed or not.
|
||||
fn built(mut self, meta_rev: &str, cached: bool) -> Self {
|
||||
self.built = Some(closure(meta_rev, cached));
|
||||
self
|
||||
}
|
||||
|
||||
fn without_record(mut self) -> Self {
|
||||
self.built = None;
|
||||
self
|
||||
}
|
||||
|
||||
|
|
@ -98,10 +101,6 @@ impl FakeHost {
|
|||
fn calls(&self) -> Vec<String> {
|
||||
self.calls.lock().unwrap().clone()
|
||||
}
|
||||
|
||||
fn builds(&self) -> Vec<String> {
|
||||
self.builds.lock().unwrap().clone()
|
||||
}
|
||||
}
|
||||
|
||||
impl Host for FakeHost {
|
||||
|
|
@ -125,54 +124,38 @@ impl Host for FakeHost {
|
|||
.ok_or_else(|| anyhow::anyhow!("{name}: no internal/meta commit was recorded"))
|
||||
}
|
||||
|
||||
async fn cached_closure(&self, name: &str, rev: &str) -> anyhow::Result<String> {
|
||||
self.record(format!("cached_closure {name} {rev}"));
|
||||
self.closure
|
||||
.clone()
|
||||
.ok_or_else(|| anyhow::anyhow!("no-record: no build is recorded for the agent"))
|
||||
async fn closure_record(&self, name: &str) -> anyhow::Result<Option<AgentClosure>> {
|
||||
self.step("closure_record", format!("closure_record {name}"))?;
|
||||
Ok(self.built.clone())
|
||||
}
|
||||
|
||||
async fn pins_at(&self, name: &str, rev: &str) -> anyhow::Result<MetaBuild> {
|
||||
self.step("pins_at", format!("pins_at {name} {rev}"))?;
|
||||
Ok(MetaBuild {
|
||||
rev: rev.to_owned(),
|
||||
hyperhive: "dh".to_owned(),
|
||||
config: Some("dc".to_owned()),
|
||||
})
|
||||
}
|
||||
|
||||
async fn realise_closure(&self, name: &str, store_path: &str) -> anyhow::Result<()> {
|
||||
self.record(format!("realise {name} {store_path}"));
|
||||
if self.fails("realise") {
|
||||
if self.refusal {
|
||||
anyhow::bail!(
|
||||
"not available: {}: lacks a signature",
|
||||
hive_priv_sock::SIGNATURE_REFUSED
|
||||
);
|
||||
}
|
||||
anyhow::bail!("not available: Couldn't connect to server");
|
||||
}
|
||||
self.realised.lock().unwrap().push(store_path.to_owned());
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn prebuild_toplevel(
|
||||
&self,
|
||||
name: &str,
|
||||
rev: &str,
|
||||
node_id: u64,
|
||||
) -> anyhow::Result<Option<String>> {
|
||||
self.record(format!("prebuild {name} {rev} node={node_id}"));
|
||||
if self.fails("prebuild") {
|
||||
anyhow::bail!("prebuild failed");
|
||||
}
|
||||
let path = self.evaluates.clone().unwrap_or_else(|| CLOSURE.to_owned());
|
||||
if !self.realised.lock().unwrap().contains(&path) {
|
||||
self.builds.lock().unwrap().push(path.clone());
|
||||
}
|
||||
Ok(Some(path))
|
||||
}
|
||||
|
||||
async fn swap_update(
|
||||
&self,
|
||||
name: &str,
|
||||
_hive: &HiveEnv,
|
||||
_paths: &AgentPaths,
|
||||
rev: &str,
|
||||
toplevel: &str,
|
||||
node_id: u64,
|
||||
) -> anyhow::Result<()> {
|
||||
self.record(format!("swap_update {name} {rev} node={node_id}"));
|
||||
self.record(format!("swap_update {name} {toplevel} node={node_id}"));
|
||||
if self.fails("swap_update") {
|
||||
anyhow::bail!("nixos-container update failed");
|
||||
}
|
||||
|
|
@ -231,8 +214,8 @@ impl Host for FakeHost {
|
|||
self.step("provision", format!("provision {name}"))
|
||||
}
|
||||
|
||||
async fn create_only(&self, name: &str, rev: &str) -> anyhow::Result<()> {
|
||||
self.step("create_only", format!("create_only {name} {rev}"))
|
||||
async fn create_only(&self, name: &str, toplevel: &str) -> anyhow::Result<()> {
|
||||
self.step("create_only", format!("create_only {name} {toplevel}"))
|
||||
}
|
||||
|
||||
async fn write_dropins(
|
||||
|
|
@ -271,10 +254,6 @@ impl Host for FakeHost {
|
|||
self.purge.clone()
|
||||
}
|
||||
|
||||
async fn forge_ensure_all(&self) {
|
||||
self.record("forge_ensure_all".to_owned());
|
||||
}
|
||||
|
||||
async fn knowledge_pull(&self, _coord: &Arc<Coordinator>) -> anyhow::Result<()> {
|
||||
if self.hang {
|
||||
std::future::pending::<()>().await;
|
||||
|
|
@ -287,6 +266,17 @@ impl Host for FakeHost {
|
|||
}
|
||||
}
|
||||
|
||||
/// The builder's record of a closure built from `internal/meta` at `meta_rev`.
|
||||
fn closure(meta_rev: &str, cached: bool) -> AgentClosure {
|
||||
AgentClosure {
|
||||
meta_rev: meta_rev.to_owned(),
|
||||
store_path: CLOSURE.to_owned(),
|
||||
system: "x86_64-linux".to_owned(),
|
||||
built_unix: 1,
|
||||
cached,
|
||||
}
|
||||
}
|
||||
|
||||
fn build() -> MetaBuild {
|
||||
MetaBuild {
|
||||
rev: REV.to_owned(),
|
||||
|
|
@ -323,188 +313,185 @@ fn node_id() -> NodeId {
|
|||
|
||||
// --- Prebuild ---------------------------------------------------------------
|
||||
|
||||
const OTHER_REV: &str = "3333333333333333333333333333333333333333";
|
||||
|
||||
/// What a block logged, level and text: the message and each field.
|
||||
#[derive(Clone, Default)]
|
||||
struct Logged(Arc<Mutex<Vec<(tracing::Level, String)>>>);
|
||||
|
||||
impl tracing::Subscriber for Logged {
|
||||
fn enabled(&self, _: &tracing::Metadata<'_>) -> bool {
|
||||
true
|
||||
}
|
||||
fn new_span(&self, _: &tracing::span::Attributes<'_>) -> tracing::span::Id {
|
||||
tracing::span::Id::from_u64(1)
|
||||
}
|
||||
fn record(&self, _: &tracing::span::Id, _: &tracing::span::Record<'_>) {}
|
||||
fn record_follows_from(&self, _: &tracing::span::Id, _: &tracing::span::Id) {}
|
||||
fn event(&self, event: &tracing::Event<'_>) {
|
||||
use std::fmt::Write as _;
|
||||
|
||||
let mut text = String::new();
|
||||
event.record(
|
||||
&mut |field: &tracing::field::Field, value: &dyn std::fmt::Debug| {
|
||||
let _ = write!(text, "{}={value:?} ", field.name());
|
||||
},
|
||||
);
|
||||
self.0
|
||||
.lock()
|
||||
.unwrap()
|
||||
.push((*event.metadata().level(), text));
|
||||
}
|
||||
fn enter(&self, _: &tracing::span::Id) {}
|
||||
fn exit(&self, _: &tracing::span::Id) {}
|
||||
}
|
||||
|
||||
impl Logged {
|
||||
fn at(&self, level: tracing::Level) -> Vec<String> {
|
||||
self.0
|
||||
.lock()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.filter(|(l, _)| *l == level)
|
||||
.map(|(_, text)| text.clone())
|
||||
.collect()
|
||||
}
|
||||
}
|
||||
|
||||
/// `fut`'s output and what it logged. The test runtime is single-threaded, so
|
||||
/// the subscriber set here sees every event the future emits.
|
||||
async fn logging<T>(fut: impl std::future::Future<Output = T>) -> (T, Logged) {
|
||||
let logged = Logged::default();
|
||||
let _guard = tracing::subscriber::set_default(logged.clone());
|
||||
let out = fut.await;
|
||||
(out, logged)
|
||||
}
|
||||
|
||||
/// Every way the swarm's closure for `alice` cannot be had, with the text the
|
||||
/// error log must carry and the host calls that must not happen.
|
||||
fn without_a_closure() -> [(&'static str, FakeHost, &'static str); 4] {
|
||||
[
|
||||
("no record", FakeHost::up().without_record(), "no build"),
|
||||
(
|
||||
"not pushed",
|
||||
FakeHost::up().built(REV, false),
|
||||
"did not push",
|
||||
),
|
||||
(
|
||||
"download fails",
|
||||
FakeHost::up().failing("realise"),
|
||||
"Couldn't connect",
|
||||
),
|
||||
(
|
||||
"record unreadable",
|
||||
FakeHost::up().failing("closure_record"),
|
||||
"closure_record failed",
|
||||
),
|
||||
]
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn prebuild_warms_the_recorded_rev_when_the_container_is_up() {
|
||||
async fn prebuild_downloads_the_swarms_closure_when_the_container_is_up() {
|
||||
let host = FakeHost::up();
|
||||
let id = node_id();
|
||||
run_prebuild(&host, "alice", id).await.expect("prebuild ok");
|
||||
let (result, log) = logging(run_prebuild(&host, "alice")).await;
|
||||
result.expect("prebuild ok");
|
||||
assert_eq!(
|
||||
host.calls(),
|
||||
[
|
||||
"is_running alice".to_owned(),
|
||||
"recorded_rev alice".to_owned(),
|
||||
format!("cached_closure alice {REV}"),
|
||||
format!("prebuild alice {REV} node={}", id.get()),
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
/// The download is followed by the local build either way. When this hive
|
||||
/// evaluates the path the swarm built, the build finds it in the store.
|
||||
#[tokio::test]
|
||||
async fn a_downloaded_closure_the_hive_also_evaluates_is_found_not_built() {
|
||||
let host = FakeHost::up().with_closure();
|
||||
let id = node_id();
|
||||
run_prebuild(&host, "alice", id).await.expect("prebuild ok");
|
||||
assert_eq!(
|
||||
host.calls(),
|
||||
[
|
||||
"is_running alice".to_owned(),
|
||||
"recorded_rev alice".to_owned(),
|
||||
format!("cached_closure alice {REV}"),
|
||||
"closure_record alice".to_owned(),
|
||||
format!("realise alice {CLOSURE}"),
|
||||
format!("prebuild alice {REV} node={}", id.get()),
|
||||
]
|
||||
);
|
||||
assert_eq!(host.builds(), Vec::<String>::new());
|
||||
}
|
||||
|
||||
/// When it evaluates another path, the build happens in `Prebuild`, after the
|
||||
/// check that the agent is up and before `Swap` could stop it.
|
||||
#[tokio::test]
|
||||
async fn a_downloaded_closure_the_hive_does_not_evaluate_is_built_before_the_swap() {
|
||||
let other = "/nix/store/zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz-nixos-system-alice";
|
||||
let host = FakeHost {
|
||||
evaluates: Some(other.to_owned()),
|
||||
..FakeHost::up().with_closure()
|
||||
};
|
||||
let id = node_id();
|
||||
run_prebuild(&host, "alice", id).await.expect("prebuild ok");
|
||||
let calls = host.calls();
|
||||
assert_eq!(calls[0], "is_running alice", "{calls:?}");
|
||||
assert_eq!(
|
||||
calls.last().map(String::as_str),
|
||||
Some(format!("prebuild alice {REV} node={}", id.get()).as_str()),
|
||||
"{calls:?}"
|
||||
);
|
||||
assert_eq!(host.builds(), [other.to_owned()]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn only_a_different_path_is_drift() {
|
||||
assert_eq!(drift(CLOSURE, Some(CLOSURE)), None);
|
||||
let note = drift(CLOSURE, Some("/nix/store/other")).expect("drift");
|
||||
let info = log.at(tracing::Level::INFO);
|
||||
assert!(
|
||||
note.contains(CLOSURE) && note.contains("/nix/store/other"),
|
||||
"{note}"
|
||||
info.iter()
|
||||
.any(|l| l.contains("prebuild: substituted") && l.contains(CLOSURE)),
|
||||
"{info:?}"
|
||||
);
|
||||
assert!(drift(CLOSURE, None).expect("no path").contains(CLOSURE));
|
||||
assert!(log.at(tracing::Level::WARN).is_empty());
|
||||
assert!(log.at(tracing::Level::ERROR).is_empty());
|
||||
}
|
||||
|
||||
/// No closure to deploy fails the node before `StopForUpdate`, so the running
|
||||
/// container is untouched, and the log names the agent and why.
|
||||
#[tokio::test]
|
||||
async fn a_failed_download_builds_locally() {
|
||||
let host = FakeHost::up().with_closure().failing("realise");
|
||||
let id = node_id();
|
||||
run_prebuild(&host, "alice", id)
|
||||
.await
|
||||
.expect("the build covers it");
|
||||
let calls = host.calls();
|
||||
assert_eq!(calls[3], format!("realise alice {CLOSURE}"), "{calls:?}");
|
||||
assert_eq!(
|
||||
calls[4],
|
||||
format!("prebuild alice {REV} node={}", id.get()),
|
||||
"{calls:?}"
|
||||
);
|
||||
assert_eq!(calls.len(), 5, "{calls:?}");
|
||||
async fn a_rebuild_without_a_closure_fails_the_prebuild_and_logs_an_error() {
|
||||
for (case, host, why) in without_a_closure() {
|
||||
let (result, log) = logging(run_prebuild(&host, "alice")).await;
|
||||
let err = result.expect_err(case);
|
||||
assert!(format!("{err:#}").starts_with("alice: "), "{case}: {err:#}");
|
||||
let errors = log.at(tracing::Level::ERROR);
|
||||
assert_eq!(errors.len(), 1, "{case}: {errors:?}");
|
||||
assert!(
|
||||
errors[0].contains("alice") && errors[0].contains(why),
|
||||
"{case}: {errors:?}"
|
||||
);
|
||||
let calls = host.calls();
|
||||
assert!(
|
||||
!calls.iter().any(|c| c.starts_with("kill")),
|
||||
"{case}: {calls:?}"
|
||||
);
|
||||
assert!(
|
||||
!calls
|
||||
.iter()
|
||||
.any(|c| c.starts_with("swap_update") || c.starts_with("create_only")),
|
||||
"{case}: {calls:?}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// The fallback is the build, so a build that fails still fails the node.
|
||||
/// The builder's record is not for the commit this hive chose: its closure is
|
||||
/// deployed and the difference is a warning.
|
||||
#[tokio::test]
|
||||
async fn a_failed_download_and_a_failed_build_fail_the_node() {
|
||||
let host = FakeHost::up()
|
||||
.with_closure()
|
||||
.failing("realise")
|
||||
.failing("prebuild");
|
||||
let err = run_prebuild(&host, "alice", node_id())
|
||||
.await
|
||||
.expect_err("nothing produced the closure");
|
||||
assert!(format!("{err:#}").contains("prebuild failed"), "{err:#}");
|
||||
async fn a_closure_built_from_another_commit_is_deployed_with_a_warning() {
|
||||
let host = FakeHost::up().built(OTHER_REV, true);
|
||||
let (result, log) = logging(run_prebuild(&host, "alice")).await;
|
||||
result.expect("drift is not a refusal");
|
||||
assert!(
|
||||
host.calls().contains(&format!("realise alice {CLOSURE}")),
|
||||
"{:?}",
|
||||
host.calls()
|
||||
);
|
||||
let warnings = log.at(tracing::Level::WARN);
|
||||
assert_eq!(warnings.len(), 1, "{warnings:?}");
|
||||
assert!(
|
||||
[REV, OTHER_REV, CLOSURE]
|
||||
.iter()
|
||||
.all(|part| warnings[0].contains(part)),
|
||||
"{warnings:?}"
|
||||
);
|
||||
assert!(log.at(tracing::Level::ERROR).is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_stopped_container_looks_up_nothing() {
|
||||
let host = FakeHost::down().with_closure();
|
||||
run_prebuild(&host, "alice", node_id())
|
||||
.await
|
||||
.expect("skipped");
|
||||
assert_eq!(host.calls(), ["is_running alice"]);
|
||||
}
|
||||
|
||||
/// The error log is for a path the cache served unsigned, not for a cache that
|
||||
/// is down or a missing record.
|
||||
#[tokio::test]
|
||||
async fn only_a_signature_refusal_is_flagged() {
|
||||
let down = substitute(
|
||||
&FakeHost::up().with_closure().failing("realise"),
|
||||
"alice",
|
||||
REV,
|
||||
)
|
||||
.await
|
||||
.expect_err("down");
|
||||
assert!(!down.signature_refused, "{}", down.reason);
|
||||
assert!(down.reason.contains("Couldn't connect"), "{}", down.reason);
|
||||
|
||||
let no_record = substitute(&FakeHost::up(), "alice", REV)
|
||||
.await
|
||||
.expect_err("no record");
|
||||
assert!(!no_record.signature_refused);
|
||||
assert!(
|
||||
no_record.reason.starts_with("no-record"),
|
||||
"{}",
|
||||
no_record.reason
|
||||
);
|
||||
|
||||
let refused = FakeHost {
|
||||
refusal: true,
|
||||
..FakeHost::up().with_closure().failing("realise")
|
||||
};
|
||||
let refused = substitute(&refused, "alice", REV)
|
||||
.await
|
||||
.expect_err("refused");
|
||||
assert!(refused.signature_refused, "{}", refused.reason);
|
||||
|
||||
let ok = substitute(&FakeHost::up().with_closure(), "alice", REV).await;
|
||||
assert_eq!(ok.ok().as_deref(), Some(CLOSURE));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn prebuild_is_skipped_when_the_container_is_down() {
|
||||
let host = FakeHost::down();
|
||||
run_prebuild(&host, "alice", node_id())
|
||||
.await
|
||||
.expect("a skipped prebuild is ok");
|
||||
run_prebuild(&host, "alice").await.expect("skipped");
|
||||
assert_eq!(host.calls(), ["is_running alice"]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn prebuild_without_a_recorded_rev_fails_before_building() {
|
||||
async fn prebuild_without_a_recorded_rev_fails_before_looking_up_a_closure() {
|
||||
let host = FakeHost {
|
||||
rev: None,
|
||||
..FakeHost::up()
|
||||
};
|
||||
let err = run_prebuild(&host, "alice", node_id())
|
||||
let err = run_prebuild(&host, "alice")
|
||||
.await
|
||||
.expect_err("nothing to build from");
|
||||
.expect_err("nothing to deploy");
|
||||
assert!(
|
||||
format!("{err:#}").contains("no internal/meta commit"),
|
||||
"{err:#}"
|
||||
);
|
||||
assert!(!host.calls().iter().any(|c| c.starts_with("prebuild")));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn prebuild_failure_fails_the_node() {
|
||||
let host = FakeHost::up().failing("prebuild");
|
||||
let err = run_prebuild(&host, "alice", node_id())
|
||||
.await
|
||||
.expect_err("build failure propagates");
|
||||
assert!(format!("{err:#}").contains("prebuild failed"), "{err:#}");
|
||||
assert!(!host.calls().iter().any(|c| c.starts_with("closure_record")));
|
||||
}
|
||||
|
||||
// --- Swap -------------------------------------------------------------------
|
||||
|
||||
#[tokio::test]
|
||||
async fn swap_updates_at_the_recorded_rev() {
|
||||
async fn swap_applies_the_swarms_closure() {
|
||||
let (_dir, coord) = coordinator();
|
||||
let host = FakeHost::down();
|
||||
let id = node_id();
|
||||
|
|
@ -513,11 +500,82 @@ async fn swap_updates_at_the_recorded_rev() {
|
|||
host.calls(),
|
||||
[
|
||||
"recorded_rev alice".to_owned(),
|
||||
format!("swap_update alice {REV} node={}", id.get()),
|
||||
"closure_record alice".to_owned(),
|
||||
format!("realise alice {CLOSURE}"),
|
||||
format!("swap_update alice {CLOSURE} node={}", id.get()),
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
/// A swap with no closure to deploy never updates: the container stays as it
|
||||
/// was, and the log names the agent and why.
|
||||
#[tokio::test]
|
||||
async fn a_swap_without_a_closure_never_updates_and_logs_an_error() {
|
||||
for (case, host, why) in without_a_closure() {
|
||||
let (_dir, coord) = coordinator();
|
||||
let (result, log) = logging(run_swap(&host, &coord, "alice", node_id())).await;
|
||||
result.expect_err(case);
|
||||
let errors = log.at(tracing::Level::ERROR);
|
||||
assert_eq!(errors.len(), 1, "{case}: {errors:?}");
|
||||
assert!(
|
||||
errors[0].contains("alice") && errors[0].contains(why),
|
||||
"{case}: {errors:?}"
|
||||
);
|
||||
assert!(
|
||||
!host.calls().iter().any(|c| c.starts_with("swap_update")),
|
||||
"{case}: {:?}",
|
||||
host.calls()
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn swap_applies_a_closure_built_from_another_commit_with_a_warning() {
|
||||
let (_dir, coord) = coordinator();
|
||||
let host = FakeHost::down().built(OTHER_REV, true);
|
||||
let id = node_id();
|
||||
let (result, log) = logging(run_swap(&host, &coord, "alice", id)).await;
|
||||
result.expect("drift is not a refusal");
|
||||
// The pins of the commit the closure came from are read before the apply
|
||||
// and recorded after it, in place of the ones this hive chose; the
|
||||
// bookkeeping node then marks those.
|
||||
assert_eq!(
|
||||
host.calls(),
|
||||
[
|
||||
"recorded_rev alice".to_owned(),
|
||||
"closure_record alice".to_owned(),
|
||||
format!("realise alice {CLOSURE}"),
|
||||
format!("pins_at alice {OTHER_REV}"),
|
||||
format!("swap_update alice {CLOSURE} node={}", id.get()),
|
||||
format!("record_meta alice {OTHER_REV}"),
|
||||
]
|
||||
);
|
||||
assert_eq!(log.at(tracing::Level::WARN).len(), 1);
|
||||
assert!(log.at(tracing::Level::ERROR).is_empty());
|
||||
}
|
||||
|
||||
/// Pins that can't be read, or an apply that fails, leave the record alone.
|
||||
#[tokio::test]
|
||||
async fn a_drifted_swap_records_pins_only_for_an_apply_that_happened() {
|
||||
for failing in ["pins_at", "swap_update"] {
|
||||
let (_dir, coord) = coordinator();
|
||||
let host = FakeHost::down().built(OTHER_REV, true).failing(failing);
|
||||
run_swap(&host, &coord, "alice", node_id())
|
||||
.await
|
||||
.expect_err(failing);
|
||||
let calls = host.calls();
|
||||
assert!(
|
||||
!calls.iter().any(|c| c.starts_with("record_meta")),
|
||||
"{failing}: {calls:?}"
|
||||
);
|
||||
assert_eq!(
|
||||
calls.iter().any(|c| c.starts_with("swap_update")),
|
||||
failing == "swap_update",
|
||||
"{failing}: {calls:?}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// A failed update fails the node and nothing else: the swap never restarts or
|
||||
/// stops the container itself — recovery is the DAG's tail `Reconcile`.
|
||||
#[tokio::test]
|
||||
|
|
@ -529,8 +587,8 @@ async fn a_failed_swap_reports_the_error_and_touches_no_container() {
|
|||
.expect_err("update failure propagates");
|
||||
assert!(format!("{err:#}").contains("update failed"), "{err:#}");
|
||||
let calls = host.calls();
|
||||
assert_eq!(calls.len(), 2, "{calls:?}");
|
||||
assert!(calls[1].starts_with("swap_update alice"), "{calls:?}");
|
||||
assert_eq!(calls.len(), 4, "{calls:?}");
|
||||
assert!(calls[3].starts_with("swap_update alice"), "{calls:?}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
|
@ -824,84 +882,82 @@ async fn provision_and_write_dropin_forward_and_propagate_failure() {
|
|||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn create_without_a_record_builds_locally_at_the_resolved_rev() {
|
||||
async fn create_downloads_the_swarms_closure_and_creates_from_it() {
|
||||
let host = FakeHost::up();
|
||||
run_create(&host, "alice", node_id()).await.expect("ok");
|
||||
let (result, log) = logging(run_create(&host, "alice")).await;
|
||||
result.expect("ok");
|
||||
assert_eq!(
|
||||
host.calls(),
|
||||
[
|
||||
"resolve_meta alice".to_owned(),
|
||||
format!("cached_closure alice {REV}"),
|
||||
format!("create_only alice {REV}"),
|
||||
format!("mark_deployed alice {REV}"),
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn create_downloads_a_cached_closure_and_builds_nothing() {
|
||||
let host = FakeHost::up().with_closure();
|
||||
let id = node_id();
|
||||
run_create(&host, "alice", id).await.expect("ok");
|
||||
assert_eq!(
|
||||
host.calls(),
|
||||
[
|
||||
"resolve_meta alice".to_owned(),
|
||||
format!("cached_closure alice {REV}"),
|
||||
"closure_record alice".to_owned(),
|
||||
format!("realise alice {CLOSURE}"),
|
||||
format!("prebuild alice {REV} node={}", id.get()),
|
||||
format!("create_only alice {REV}"),
|
||||
format!("create_only alice {CLOSURE}"),
|
||||
format!("mark_deployed alice {REV}"),
|
||||
]
|
||||
);
|
||||
assert_eq!(host.builds(), Vec::<String>::new());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn create_builds_locally_when_the_download_fails() {
|
||||
let host = FakeHost::up().with_closure().failing("realise");
|
||||
run_create(&host, "alice", node_id()).await.expect("ok");
|
||||
assert_eq!(
|
||||
host.calls(),
|
||||
[
|
||||
"resolve_meta alice".to_owned(),
|
||||
format!("cached_closure alice {REV}"),
|
||||
format!("realise alice {CLOSURE}"),
|
||||
format!("create_only alice {REV}"),
|
||||
format!("mark_deployed alice {REV}"),
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn create_builds_the_evaluated_path_when_it_differs_from_the_download() {
|
||||
let other = "/nix/store/zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz-nixos-system-alice";
|
||||
let host = FakeHost {
|
||||
evaluates: Some(other.to_owned()),
|
||||
..FakeHost::up().with_closure()
|
||||
};
|
||||
run_create(&host, "alice", node_id()).await.expect("ok");
|
||||
assert_eq!(host.builds(), [other.to_owned()]);
|
||||
let calls = host.calls();
|
||||
assert!(
|
||||
calls
|
||||
log.at(tracing::Level::INFO)
|
||||
.iter()
|
||||
.any(|c| c == &format!("create_only alice {REV}")),
|
||||
"{calls:?}"
|
||||
.any(|l| l.contains("create: substituted") && l.contains(CLOSURE))
|
||||
);
|
||||
assert!(log.at(tracing::Level::WARN).is_empty());
|
||||
}
|
||||
|
||||
/// A create with no closure to deploy is refused: no container is made, none is
|
||||
/// marked deployed, and the log names the agent and why.
|
||||
#[tokio::test]
|
||||
async fn a_create_without_a_closure_is_refused_and_logs_an_error() {
|
||||
for (case, host, why) in without_a_closure() {
|
||||
let (result, log) = logging(run_create(&host, "alice")).await;
|
||||
let err = result.expect_err(case);
|
||||
assert!(format!("{err:#}").starts_with("alice: "), "{case}: {err:#}");
|
||||
let errors = log.at(tracing::Level::ERROR);
|
||||
assert_eq!(errors.len(), 1, "{case}: {errors:?}");
|
||||
assert!(
|
||||
errors[0].contains("alice") && errors[0].contains(why),
|
||||
"{case}: {errors:?}"
|
||||
);
|
||||
let calls = host.calls();
|
||||
assert!(
|
||||
!calls
|
||||
.iter()
|
||||
.any(|c| c.starts_with("create_only") || c.starts_with("mark_deployed")),
|
||||
"{case}: {calls:?}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn create_uses_a_closure_built_from_another_commit_with_a_warning() {
|
||||
let host = FakeHost::up().built(OTHER_REV, true);
|
||||
let (result, log) = logging(run_create(&host, "alice")).await;
|
||||
result.expect("drift is not a refusal");
|
||||
let calls = host.calls();
|
||||
// Created from the swarm's closure, and marked as the commit it was built
|
||||
// from, not the one this hive chose.
|
||||
assert_eq!(
|
||||
calls,
|
||||
[
|
||||
"resolve_meta alice".to_owned(),
|
||||
"closure_record alice".to_owned(),
|
||||
format!("realise alice {CLOSURE}"),
|
||||
format!("pins_at alice {OTHER_REV}"),
|
||||
format!("create_only alice {CLOSURE}"),
|
||||
format!("mark_deployed alice {OTHER_REV}"),
|
||||
]
|
||||
);
|
||||
assert_eq!(log.at(tracing::Level::WARN).len(), 1);
|
||||
assert!(log.at(tracing::Level::ERROR).is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_failed_create_is_never_marked_deployed() {
|
||||
let host = FakeHost::up().failing("create_only");
|
||||
run_create(&host, "alice", node_id())
|
||||
.await
|
||||
.expect_err("create fails");
|
||||
assert_eq!(host.calls().len(), 3);
|
||||
run_create(&host, "alice").await.expect_err("create fails");
|
||||
assert_eq!(host.calls().len(), 4);
|
||||
let host = FakeHost::up().failing("resolve_meta");
|
||||
run_create(&host, "alice", node_id())
|
||||
.await
|
||||
.expect_err("resolve fails");
|
||||
run_create(&host, "alice").await.expect_err("resolve fails");
|
||||
assert_eq!(host.calls(), ["resolve_meta alice"]);
|
||||
}
|
||||
|
||||
|
|
@ -1071,13 +1127,6 @@ async fn emit_rebuilt_reports_success_and_failure_under_one_key() {
|
|||
|
||||
// --- Sweeps and pulls ---------------------------------------------------------
|
||||
|
||||
#[tokio::test]
|
||||
async fn forge_sweep_runs_once_and_never_fails() {
|
||||
let host = FakeHost::up();
|
||||
run_forge_sweep(&host).await.expect("ok");
|
||||
assert_eq!(host.calls(), ["forge_ensure_all"]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn knowledge_pull_passes_the_result_through_and_times_out() {
|
||||
let (_dir, coord) = coordinator();
|
||||
|
|
@ -1165,11 +1214,19 @@ async fn dispatch_routes_each_host_backed_kind_to_its_executor() {
|
|||
),
|
||||
(
|
||||
alice(|agent| NodeKind::Prebuild { agent }),
|
||||
&["is_running alice", "recorded_rev alice", "prebuild alice"],
|
||||
&[
|
||||
"is_running alice",
|
||||
"recorded_rev alice",
|
||||
"closure_record alice",
|
||||
],
|
||||
),
|
||||
(
|
||||
alice(|agent| NodeKind::Swap { agent }),
|
||||
&["recorded_rev alice", "swap_update alice"],
|
||||
&[
|
||||
"recorded_rev alice",
|
||||
"closure_record alice",
|
||||
"swap_update alice",
|
||||
],
|
||||
),
|
||||
(
|
||||
alice(|agent| NodeKind::RebuildBookkeeping { agent }),
|
||||
|
|
@ -1183,6 +1240,7 @@ async fn dispatch_routes_each_host_backed_kind_to_its_executor() {
|
|||
alice(|agent| NodeKind::Create { agent }),
|
||||
&[
|
||||
"resolve_meta alice",
|
||||
"closure_record alice",
|
||||
"create_only alice",
|
||||
"mark_deployed alice",
|
||||
],
|
||||
|
|
@ -1236,7 +1294,6 @@ async fn dispatch_routes_each_host_backed_kind_to_its_executor() {
|
|||
async fn dispatch_routes_the_agentless_and_hostless_kinds() {
|
||||
let host = FakeHost::up();
|
||||
for kind in [
|
||||
NodeKind::ForgeSweep,
|
||||
NodeKind::WantedPull,
|
||||
NodeKind::KnowledgePull,
|
||||
alice(|agent| NodeKind::AgentWindow { agent }),
|
||||
|
|
@ -1256,7 +1313,7 @@ async fn dispatch_routes_the_agentless_and_hostless_kinds() {
|
|||
);
|
||||
}
|
||||
let calls = host.calls();
|
||||
for want in ["forge_ensure_all", "wanted_pull", "knowledge_pull"] {
|
||||
for want in ["wanted_pull", "knowledge_pull"] {
|
||||
assert!(calls.iter().any(|c| c == want), "{want} in {calls:?}");
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -32,9 +32,9 @@ pub enum NodeKind {
|
|||
/// The rebuild's preamble: runtime-dir prep and the `internal/meta`
|
||||
/// commit the build reads.
|
||||
MetaSync { agent: String },
|
||||
/// Out-of-band toplevel build while the container keeps serving.
|
||||
/// Download of the swarm's closure while the container keeps serving.
|
||||
Prebuild { agent: String },
|
||||
/// `nixos-container update` profile-swap.
|
||||
/// Profile-swap onto the swarm's closure.
|
||||
Swap { agent: String },
|
||||
/// The post-`Swap` bookkeeping tail as a first-class node.
|
||||
RebuildBookkeeping { agent: String },
|
||||
|
|
@ -96,9 +96,6 @@ pub enum NodeKind {
|
|||
/// Write the agent's durable power intent (`wanted = Up` when `up`,
|
||||
/// else `Offline`) as the head node of a power-op DAG.
|
||||
SetWanted { agent: String, up: bool },
|
||||
/// One-shot boot-time publish of the hive's forge read token for nix
|
||||
/// (`forge::ensure_all`). Agentless.
|
||||
ForgeSweep,
|
||||
/// `/knowledge` pull (`knowledge::pull`): at boot, on the swarm
|
||||
/// knowledge-changed event, and hourly. Agentless.
|
||||
KnowledgePull,
|
||||
|
|
@ -165,7 +162,7 @@ impl NodeKind {
|
|||
| NodeKind::AgentWindow { agent }
|
||||
| NodeKind::EmitRebuilt { agent, .. }
|
||||
| NodeKind::SetWanted { agent, .. } => agent,
|
||||
NodeKind::ForgeSweep | NodeKind::KnowledgePull | NodeKind::WantedPull => "",
|
||||
NodeKind::KnowledgePull | NodeKind::WantedPull => "",
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -190,7 +190,7 @@ fn rebuild_subtree<'a>(
|
|||
.needs(Resource::Agent(a()))
|
||||
.after_ok(meta_sync);
|
||||
|
||||
// Siblings under the brace: the build and the quiesce chain run
|
||||
// Siblings under the brace: the download and the quiesce chain run
|
||||
// concurrently, borrowing the brace's grants rather than declaring their
|
||||
// own. Declaring the lease on both would make them mutually exclusive —
|
||||
// it is single-unit — which is exactly what this shape exists to avoid.
|
||||
|
|
@ -200,8 +200,8 @@ fn rebuild_subtree<'a>(
|
|||
|
||||
let drain = graceful.then(|| quiesce(builder, agent, agent_window));
|
||||
|
||||
// The container goes down here, not earlier: `AfterOk` the build so a
|
||||
// failed build never stops a healthy container, and `AfterOk` the drain so
|
||||
// The container goes down here, not earlier: `AfterOk` the download so a
|
||||
// missing closure never stops a healthy container, and `AfterOk` the drain so
|
||||
// the agent has checkpointed.
|
||||
let mut stop_for_update = builder
|
||||
.node(NodeKind::StopForUpdate { agent: a() })
|
||||
|
|
|
|||
|
|
@ -10,9 +10,8 @@ use anyhow::{Context, Result, bail};
|
|||
use tokio::io::{AsyncRead, AsyncReadExt};
|
||||
use tokio::process::Command;
|
||||
|
||||
/// Deadline for a child that talks to a remote (forge fetch/push, `nix flake
|
||||
/// update`/`lock`): a healthy one takes seconds, so ten minutes only ever
|
||||
/// fires on a hang.
|
||||
/// Deadline for a child that talks to a remote (forge fetch/push): a healthy
|
||||
/// one takes seconds, so ten minutes only ever fires on a hang.
|
||||
pub const FETCH_DEADLINE: Duration = Duration::from_mins(10);
|
||||
|
||||
/// Deadline for a git child that touches only the local repo.
|
||||
|
|
@ -21,7 +20,7 @@ pub const GIT_LOCAL_DEADLINE: Duration = Duration::from_mins(2);
|
|||
/// Run `cmd` to completion and capture its output, failing once `deadline`
|
||||
/// passes. `what` names the command in the error. The child leads its own
|
||||
/// process group and the whole group is `SIGKILL`ed on timeout, so a
|
||||
/// grandchild (a `git` or credential helper under `nix`) dies with it instead
|
||||
/// grandchild (a `git` or credential helper it spawned) dies with it instead
|
||||
/// of outliving the lock.
|
||||
pub async fn run_with_deadline(
|
||||
cmd: &mut Command,
|
||||
|
|
|
|||
|
|
@ -234,9 +234,9 @@ pub async fn provision_container(name: &str, paths: &AgentPaths) -> Result<()> {
|
|||
|
||||
/// The `nixos-container create` proper — the job queue's `Create` node.
|
||||
/// Assumes `provision_container` already registered the agent.
|
||||
/// `meta_rev` is the `internal/meta` commit to build from (`crate::meta_rev`).
|
||||
pub async fn create_only(name: &str, meta_rev: &str) -> Result<()> {
|
||||
priv_run_inner("create", name, Some(meta_rev), None).await
|
||||
/// `toplevel` is the store path of the system closure to create it from.
|
||||
pub async fn create_only(name: &str, toplevel: &str) -> Result<()> {
|
||||
priv_run_inner("create", name, Some(toplevel), None).await
|
||||
}
|
||||
|
||||
/// Rebuild-path preamble, run by the job queue's `Prebuild` node: fail fast on
|
||||
|
|
@ -285,20 +285,20 @@ pub async fn prepare_rebuild_dirs(name: &str, paths: &AgentPaths) -> Result<()>
|
|||
}
|
||||
|
||||
/// Profile-swap for an existing, stopped container: re-apply the
|
||||
/// drop-ins, then `nixos-container update`. The job queue's `Swap`
|
||||
/// node. Requires the container stopped (the queue's `StopForUpdate`
|
||||
/// upstream); does NOT start it — the DAG's tail `Reconcile` owns
|
||||
/// bringing the agent back to its wanted power state. `meta_rev` is as for
|
||||
/// drop-ins, then point the container at the closure `toplevel`. The job
|
||||
/// queue's `Swap` node. Requires the container stopped (the queue's
|
||||
/// `StopForUpdate` upstream); does NOT start it — the DAG's tail `Reconcile` owns
|
||||
/// bringing the agent back to its wanted power state. `toplevel` is as for
|
||||
/// [`create_only`].
|
||||
pub async fn swap_update(
|
||||
name: &str,
|
||||
hive: &HiveEnv,
|
||||
paths: &AgentPaths,
|
||||
meta_rev: &str,
|
||||
toplevel: &str,
|
||||
node_id: Option<u64>,
|
||||
) -> Result<()> {
|
||||
write_dropins(name, hive, paths).await?;
|
||||
priv_run_inner("update", name, Some(meta_rev), node_id).await
|
||||
priv_run_inner("update", name, Some(toplevel), node_id).await
|
||||
}
|
||||
|
||||
/// Build the `AgentSpec` list for [`crate::agent_set::refresh`] from
|
||||
|
|
@ -720,195 +720,6 @@ fn confirm_gone_after_failed_destroy(
|
|||
}
|
||||
}
|
||||
|
||||
/// Wall-clock bound on [`prebuild_toplevel`]'s `nix build`. Four times CI's
|
||||
/// observed cold-cache `nix flake check` (~15 min, `.forgejo/workflows/ci.yml`),
|
||||
/// which compiles this workspace's crates, the part of an agent toplevel no
|
||||
/// binary cache holds. A stalled download is bounded by nix's own
|
||||
/// `stalled-download-timeout`; this covers the rest, such as a wedged
|
||||
/// nix-daemon holding the rebuild's build slot forever.
|
||||
const PREBUILD_TIMEOUT: std::time::Duration = std::time::Duration::from_hours(1);
|
||||
|
||||
/// Wait for `child` until `limit`. `None` means it was still running, and its
|
||||
/// whole process group was `SIGKILL`ed and the child reaped, so a grandchild
|
||||
/// holding the daemon connection dies with it. The child must have been
|
||||
/// spawned with `process_group(0)`.
|
||||
async fn wait_bounded(
|
||||
child: &mut tokio::process::Child,
|
||||
limit: std::time::Duration,
|
||||
) -> std::io::Result<Option<std::process::ExitStatus>> {
|
||||
// Group leader, so its pid is the pgid. Read before waiting: a reaped
|
||||
// child has no id.
|
||||
let pgid = child.id().and_then(|pid| i32::try_from(pid).ok());
|
||||
if let Ok(status) = tokio::time::timeout(limit, child.wait()).await {
|
||||
return status.map(Some);
|
||||
}
|
||||
if let Some(pgid) = pgid {
|
||||
// SAFETY: plain `kill(2)`; `-pgid` targets the child's own process
|
||||
// group, which `process_group(0)` made it lead.
|
||||
unsafe {
|
||||
libc::kill(-pgid, libc::SIGKILL);
|
||||
}
|
||||
}
|
||||
child.wait().await?;
|
||||
Ok(None)
|
||||
}
|
||||
|
||||
/// Log and record a prebuild's stdout; the last non-empty line is the
|
||||
/// toplevel `nix build --print-out-paths` printed.
|
||||
async fn pump_prebuild_stdout(
|
||||
stdout: tokio::process::ChildStdout,
|
||||
cmdline: String,
|
||||
logs: Option<std::sync::Arc<crate::build_logs::BuildLogs>>,
|
||||
log_id: Option<i64>,
|
||||
) -> Option<String> {
|
||||
use tokio::io::{AsyncBufReadExt, BufReader};
|
||||
let mut lines = BufReader::new(stdout).lines();
|
||||
let mut out_path = None;
|
||||
while let Ok(Some(line)) = lines.next_line().await {
|
||||
tracing::info!(target: "nix-prebuild", cmdline = %cmdline, "{line}");
|
||||
if !line.trim().is_empty() {
|
||||
out_path = Some(line.trim().to_owned());
|
||||
}
|
||||
if let (Some(h), Some(id)) = (&logs, log_id) {
|
||||
h.append_stdout(id, &line);
|
||||
}
|
||||
}
|
||||
out_path
|
||||
}
|
||||
|
||||
/// Pre-build `system.build.toplevel` against `meta#<name>` so the
|
||||
/// subsequent `nixos-container update` finds the result cached and
|
||||
/// skips straight to the profile-swap. Store-warming only — container
|
||||
/// is untouched. See `docs/scheduler/coordinator.md::Rebuild path` for why
|
||||
/// the prebuild happens before stop, and `docs/scheduler/coordinator.md::Prebuild
|
||||
/// attr path` for why the explicit nixosConfigurations attr is required.
|
||||
///
|
||||
/// `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.
|
||||
///
|
||||
/// Returns the toplevel `nix build` printed, `None` if it printed nothing.
|
||||
pub async fn prebuild_toplevel(
|
||||
name: &str,
|
||||
flake_ref: &str,
|
||||
node_id: Option<u64>,
|
||||
) -> Result<Option<String>> {
|
||||
use tokio::io::{AsyncBufReadExt, BufReader};
|
||||
// Split `<root>#<name>` so we can re-emit with the explicit
|
||||
// `nixosConfigurations.<name>` segment. The flake_ref shape is
|
||||
// constructed by the caller and always contains exactly one
|
||||
// `#`; `split_once` returning None here would be a programmer
|
||||
// error we'd want to surface loudly rather than paper over.
|
||||
let (flake_root, fragment) = flake_ref
|
||||
.split_once('#')
|
||||
.with_context(|| format!("flake_ref {flake_ref:?} missing '#<name>' fragment"))?;
|
||||
// Sanity-check the fragment matches the agent name we were
|
||||
// passed — guards against future calls that pass a divergent
|
||||
// pair (no current callsite does, but the pair is redundant
|
||||
// and worth checking once).
|
||||
if fragment != name {
|
||||
anyhow::bail!("prebuild_toplevel: flake_ref fragment '{fragment}' ≠ agent name '{name}'");
|
||||
}
|
||||
let attr = format!("{flake_root}#nixosConfigurations.{name}.config.system.build.toplevel");
|
||||
let args = vec![
|
||||
"--extra-experimental-features",
|
||||
"nix-command flakes",
|
||||
"build",
|
||||
"--no-link",
|
||||
"--print-out-paths",
|
||||
&attr,
|
||||
];
|
||||
let cmdline = format!("nix {}", args.join(" "));
|
||||
tracing::info!(%name, %cmdline, "prebuild: warming system toplevel");
|
||||
|
||||
// Open a build_logs row for this attempt (best-effort — None when
|
||||
// the global handle hasn't been installed, e.g. early startup
|
||||
// or standalone tests). Lines pumped from stdout/stderr append
|
||||
// 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, node_id)
|
||||
.map_err(|e| {
|
||||
tracing::warn!(error = ?e, "build_logs: start failed (prebuild log dropped)");
|
||||
})
|
||||
.ok()
|
||||
});
|
||||
|
||||
let mut child = Command::new("nix")
|
||||
.args(&args)
|
||||
.stdout(std::process::Stdio::piped())
|
||||
.stderr(std::process::Stdio::piped())
|
||||
.process_group(0)
|
||||
.spawn()
|
||||
.with_context(|| format!("spawn {cmdline}"))?;
|
||||
|
||||
let stdout = child.stdout.take().expect("piped stdout");
|
||||
let stderr = child.stderr.take().expect("piped stderr");
|
||||
|
||||
let pump_stdout = tokio::spawn(pump_prebuild_stdout(
|
||||
stdout,
|
||||
cmdline.clone(),
|
||||
logs.clone(),
|
||||
log_id,
|
||||
));
|
||||
|
||||
let stderr_cmdline = cmdline.clone();
|
||||
let stderr_logs = logs.clone();
|
||||
let pump_stderr = tokio::spawn(async move {
|
||||
let mut lines = BufReader::new(stderr).lines();
|
||||
while let Ok(Some(line)) = lines.next_line().await {
|
||||
if is_actionable_nix_line(&line) {
|
||||
tracing::warn!(target: "nix-prebuild", cmdline = %stderr_cmdline, "{line}");
|
||||
} else {
|
||||
tracing::info!(target: "nix-prebuild", cmdline = %stderr_cmdline, "{line}");
|
||||
}
|
||||
if let (Some(h), Some(id)) = (&stderr_logs, log_id) {
|
||||
h.append_stderr(id, &line);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
let Some(status) = wait_bounded(&mut child, PREBUILD_TIMEOUT)
|
||||
.await
|
||||
.with_context(|| format!("wait {cmdline}"))?
|
||||
else {
|
||||
// A process that left the group can still hold the pipes open.
|
||||
pump_stdout.abort();
|
||||
pump_stderr.abort();
|
||||
if let (Some(h), Some(id)) = (&logs, log_id) {
|
||||
h.finish(id, crate::build_logs::BuildStatus::Fail);
|
||||
}
|
||||
let secs = PREBUILD_TIMEOUT.as_secs();
|
||||
match log_id {
|
||||
Some(id) => {
|
||||
bail!("prebuild {cmdline} timed out after {secs}s; killed it; see build log #{id}")
|
||||
}
|
||||
None => bail!("prebuild {cmdline} timed out after {secs}s; killed it"),
|
||||
}
|
||||
};
|
||||
let out_path = pump_stdout.await.ok().flatten();
|
||||
let _ = pump_stderr.await;
|
||||
|
||||
let ok = status.success();
|
||||
if let (Some(h), Some(id)) = (&logs, log_id) {
|
||||
h.finish(
|
||||
id,
|
||||
if ok {
|
||||
crate::build_logs::BuildStatus::Ok
|
||||
} else {
|
||||
crate::build_logs::BuildStatus::Fail
|
||||
},
|
||||
);
|
||||
}
|
||||
if !ok {
|
||||
match log_id {
|
||||
Some(id) => bail!("prebuild {cmdline} failed ({status}); see build log #{id}"),
|
||||
None => bail!("prebuild {cmdline} failed ({status})"),
|
||||
}
|
||||
}
|
||||
Ok(out_path)
|
||||
}
|
||||
|
||||
pub async fn list() -> Result<Vec<String>> {
|
||||
let stdout = crate::priv_client::list_containers().await?;
|
||||
Ok(stdout
|
||||
|
|
@ -990,11 +801,11 @@ fn make_log_callback(
|
|||
}
|
||||
}
|
||||
|
||||
/// A `create` or `update` asked for without an `internal/meta` commit. The
|
||||
/// public entry points take the commit, so this guards the string dispatch
|
||||
/// A `create` or `update` asked for without a closure. The
|
||||
/// public entry points take the closure, so this guards the string dispatch
|
||||
/// below only.
|
||||
fn missing_meta_rev(kind: &str, name: &str) -> anyhow::Error {
|
||||
anyhow::anyhow!("{name}: {kind} needs an internal/meta commit to build from")
|
||||
fn missing_toplevel(kind: &str, name: &str) -> anyhow::Error {
|
||||
anyhow::anyhow!("{name}: {kind} needs a closure to apply")
|
||||
}
|
||||
|
||||
/// Execute a container operation via hive-priv and integrate with
|
||||
|
|
@ -1016,7 +827,7 @@ async fn priv_run(kind: &str, name: &str) -> Result<()> {
|
|||
async fn priv_run_inner(
|
||||
kind: &str,
|
||||
name: &str,
|
||||
meta_rev: Option<&str>,
|
||||
toplevel: Option<&str>,
|
||||
node_id: Option<u64>,
|
||||
) -> Result<()> {
|
||||
let container = container_name(name);
|
||||
|
|
@ -1034,27 +845,27 @@ async fn priv_run_inner(
|
|||
// For long-running ops use the streaming protocol so build_logs
|
||||
// receives lines in real time rather than as a batch at completion.
|
||||
let result: Result<()> = match kind {
|
||||
"create" => match meta_rev {
|
||||
Some(rev) => {
|
||||
"create" => match toplevel {
|
||||
Some(toplevel) => {
|
||||
crate::priv_client::create_container_streaming(
|
||||
name,
|
||||
rev,
|
||||
toplevel,
|
||||
make_log_callback(logs.clone(), log_id, cmdline.clone()),
|
||||
)
|
||||
.await
|
||||
}
|
||||
None => Err(missing_meta_rev(kind, name)),
|
||||
None => Err(missing_toplevel(kind, name)),
|
||||
},
|
||||
"update" => match meta_rev {
|
||||
Some(rev) => {
|
||||
"update" => match toplevel {
|
||||
Some(toplevel) => {
|
||||
crate::priv_client::update_container_streaming(
|
||||
name,
|
||||
rev,
|
||||
toplevel,
|
||||
make_log_callback(logs.clone(), log_id, cmdline.clone()),
|
||||
)
|
||||
.await
|
||||
}
|
||||
None => Err(missing_meta_rev(kind, name)),
|
||||
None => Err(missing_toplevel(kind, name)),
|
||||
},
|
||||
"start" => crate::priv_client::start_container(name).await,
|
||||
"stop" => crate::priv_client::stop_container(name).await,
|
||||
|
|
|
|||
|
|
@ -349,68 +349,6 @@ fn agent_names_propagates_an_unreadable_list() {
|
|||
assert!(format!("{err:#}").contains("hive-priv"), "{err:#}");
|
||||
}
|
||||
|
||||
fn is_dead(pid: &str) -> bool {
|
||||
// A killed process that is not our child stays a zombie until its own
|
||||
// parent reaps it, so a zombie counts as dead.
|
||||
match std::fs::read_to_string(format!("/proc/{pid}/stat")) {
|
||||
Err(_) => true,
|
||||
Ok(stat) => stat
|
||||
.rsplit_once(") ")
|
||||
.is_some_and(|(_, rest)| rest.starts_with('Z')),
|
||||
}
|
||||
}
|
||||
|
||||
/// A child still running at the limit comes back `None`, and the kill reaches
|
||||
/// its whole process group: the backgrounded grandchild dies too, not just the
|
||||
/// shell that started it.
|
||||
#[tokio::test]
|
||||
async fn a_prebuild_past_its_limit_is_killed_with_its_group() {
|
||||
use std::time::{Duration, Instant};
|
||||
let dir = tempfile::tempdir().expect("tempdir");
|
||||
let pidfile = dir.path().join("grandchild");
|
||||
let mut child = Command::new("sh")
|
||||
.arg("-c")
|
||||
.arg(format!("sleep 600 & echo $! > {}; wait", pidfile.display()))
|
||||
.process_group(0)
|
||||
.spawn()
|
||||
.expect("spawn sh");
|
||||
|
||||
let started = Instant::now();
|
||||
let status = wait_bounded(&mut child, Duration::from_secs(2))
|
||||
.await
|
||||
.expect("wait_bounded");
|
||||
assert!(status.is_none(), "must time out, got {status:?}");
|
||||
assert!(
|
||||
started.elapsed() < Duration::from_mins(1),
|
||||
"the limit did not bound the wait"
|
||||
);
|
||||
|
||||
let pid = std::fs::read_to_string(&pidfile).expect("grandchild pid");
|
||||
let pid = pid.trim();
|
||||
let deadline = Instant::now() + Duration::from_secs(5);
|
||||
while !is_dead(pid) && Instant::now() < deadline {
|
||||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||
}
|
||||
assert!(is_dead(pid), "grandchild {pid} outlived the timeout");
|
||||
}
|
||||
|
||||
/// The control: a child that exits inside its limit comes back with its
|
||||
/// status.
|
||||
#[tokio::test]
|
||||
async fn a_prebuild_inside_its_limit_returns_its_status() {
|
||||
let mut child = Command::new("sh")
|
||||
.arg("-c")
|
||||
.arg("exit 3")
|
||||
.process_group(0)
|
||||
.spawn()
|
||||
.expect("spawn sh");
|
||||
let status = wait_bounded(&mut child, std::time::Duration::from_mins(1))
|
||||
.await
|
||||
.expect("wait_bounded")
|
||||
.expect("a fast child must not time out");
|
||||
assert_eq!(status.code(), Some(3));
|
||||
}
|
||||
|
||||
/// The setup path creates the agent's `work/` dir beside `state/`, so the
|
||||
/// bind source exists before nspawn starts the container.
|
||||
#[test]
|
||||
|
|
|
|||
|
|
@ -115,7 +115,7 @@ enum Cmd {
|
|||
/// option.
|
||||
#[arg(long)]
|
||||
model_prices: Option<String>,
|
||||
/// Override: number of concurrent nix-heavy job-queue nodes
|
||||
/// Override: number of concurrent closure-fetching job-queue nodes
|
||||
/// (prebuild / profile-swap / create / meta lock). Set via the
|
||||
/// `services.hyperhive.deploy.hive-controller.buildSlots` NixOS option.
|
||||
#[arg(long)]
|
||||
|
|
@ -259,10 +259,6 @@ async fn cmd_serve(
|
|||
tracing::warn!(error = ?e, "auto-update task failed");
|
||||
}
|
||||
});
|
||||
// Forge credential publish: a `NodeKind::ForgeSweep` DAG node (see
|
||||
// `workers::auto_update::submit_startup_sweep_nodes`), submitted
|
||||
// unconditionally on every boot — moved off a bare `tokio::spawn` so it
|
||||
// shows as real work on the dashboard.
|
||||
// Knowledge periodic pull: hourly fallback in case a knowledge event from
|
||||
// the swarm is missed (e.g. hive-c0re was down when it was sent). Sleeps
|
||||
// before its first pull: the boot `NodeKind::KnowledgePull` DAG node
|
||||
|
|
|
|||
Some files were not shown because too many files have changed in this diff Show more
Loading…
Reference in a new issue