diff --git a/docs/swarm/builder.md b/docs/swarm/builder.md index 0282af76..e3ad2e8e 100644 --- a/docs/swarm/builder.md +++ b/docs/swarm/builder.md @@ -46,13 +46,53 @@ Where it runs and how it's enabled: [`services.md`](services.md#builder). | the builder busy for 30 seconds | `failure` | why | failed | | no builder listening, or no reply in time | `error` | why | failed | - The full tail is in the failed node's error on the Jobs page and in the - builder's journal line for the build. + The controller records every build that gets a final status, and the + status links to that record: see [Build records](#build-records). The + full tail is also in the failed node's error on the Jobs page and in the + builder's journal line for the build, as the field `F_ERROR_TAIL`: + + ```console + $ swarm-logs query '_time:7d _SYSTEMD_UNIT:="swarm-builder.service" _msg:="agent config build failed" | fields _time, F_AGENT, F_PR, F_REV, F_ERROR_TAIL' --format json + ``` The builder holds the last 64 KiB of each of `nix`'s output streams while the build runs, and the reply is the last 30 lines of that (at most 4000 bytes), so a long build log never grows its memory. +## Build records + +`swarm-controller` records each build when it writes the final status: +built, failed, or no build ran or came back. The newest 100 are one JSON +array under the key `recent` in the `builder-runs` KV bucket on the swarm +queue, which only the controller reads or writes. It creates the bucket on +first use and adds each build with a compare-and-set write, so the records +survive a controller restart. Recording the 101st drops the oldest. A record +the controller can't decode stays in the bucket, out of the list, and counts +toward the 100. + +Each record has: + +- `id`: `--` +- `repo` (`agent-configs/`), `pr` and `head_sha` +- `started_unix` and `finished_unix` +- `outcome`: `built`, `failed` or `error`, the status's `success`, `failure` + or `error` +- `out_path` for `built`, `error_tail` (the last 30 lines `nix` printed) for + `failed`, and `detail` (why) for `error` + +`GET /api/builder/runs` lists them, newest first, and +`GET /api/builder/runs/` returns one, or 404 when it isn't among the +newest 100. Both are behind the swarm UI vhost's `admins` gate, and answer 503 +when the controller has no swarm queue or can't open the bucket, 500 when it +can't read it. + +The final status links to `https:///api/builder/runs/`, so +selecting **Details** on a red `swarm-builder` check opens the record, with +`nix`'s tail, in the browser. A status has no link when the record wasn't +stored (the controller logs the failed write) or when the controller doesn't +serve the swarm vhost (`SWARM_CONTROLLER_PUBLIC_URL` unset). The `pending` +status has no link. + ## What the check is, and isn't - It composes the config the way a hive's meta flake does @@ -62,7 +102,7 @@ Where it runs and how it's enabled: [`services.md`](services.md#builder). hive's. A green check can therefore still fail on a hive, and the reverse. - The agent base is the builder's host's, not the hive's. Hives on another hyperhive or nixpkgs pin build something else. -- The builder pushes nothing and records nothing. The closure stays in the builder +- The builder pushes nothing. The closure stays in the builder host's store until it's collected, so on a host that also runs the hive, the hive's own build can reuse whatever matches. - It doesn't gate anything. A merge deploys whatever the status says. diff --git a/swarm-controller/README.md b/swarm-controller/README.md index db38a8a1..6ac96d37 100644 --- a/swarm-controller/README.md +++ b/swarm-controller/README.md @@ -79,6 +79,12 @@ The swarm's control plane. The swarm UI and `swarmctl` are its clients. leaves the group drops out of the list. Every agent reads the key into its system prompt → [`docs/turn-loop/claude-invocation.md`](../docs/turn-loop/claude-invocation.md). +- **Builder runs** — the swarm builder's newest 100 finished builds of agent + config PRs, with `nix`'s tail for a failed one, one document in the + `builder-runs` KV bucket, which this daemon alone writes + (`GET /api/builder/runs`, `GET /api/builder/runs/{id}`, the link on each + `swarm-builder` commit status) → + [`docs/swarm/builder.md`](../docs/swarm/builder.md#build-records). - **Relays** — each agent's terminal and turn-state header as SSE, agent icons, the cross-repo issue report, and the UI's quick links. - **Alerts** — the swarm Grafana's current alerts (`GET /api/alerts`), read diff --git a/swarm-controller/src/agent_limits.rs b/swarm-controller/src/agent_limits.rs index d22096ca..d801db4c 100644 --- a/swarm-controller/src/agent_limits.rs +++ b/swarm-controller/src/agent_limits.rs @@ -509,6 +509,7 @@ pub(crate) mod tests { reviews: Arc::default(), try_outs: crate::review_classifier::TryOuts::default(), runs: crate::review_classifier::runs::Runs::Live(None), + builds: crate::builder::runs::Builds::Live(None), alerts: None, } } diff --git a/swarm-controller/src/builder.rs b/swarm-controller/src/builder.rs index fbb288b7..e79278a2 100644 --- a/swarm-controller/src/builder.rs +++ b/swarm-controller/src/builder.rs @@ -1,21 +1,26 @@ //! The `BuildAgent` node: send a config PR's head to the swarm builder over the -//! queue, and turn its answer into the [`CONTEXT`] commit status on that head. +//! queue, turn its answer into the [`CONTEXT`] commit status on that head, and +//! record the build, `nix`'s tail with it, in [`runs`], whose +//! `GET /api/builder/runs/{id}` the status links to. //! -//! The status is the check's only record besides the node itself: nothing is -//! pushed, and the closure the builder made stays in its host's store. A red -//! status blocks nothing; the merge deploys as before. +//! Nothing is pushed, and the closure the builder made stays in its host's +//! store. A red status blocks nothing; the merge deploys as before. //! //! Builds share one `SwarmResourceKind::Builder` slot, so one runs at a time, //! and a newer push replaces its PR's build while that build is still //! pending (`queue_build`). +pub(crate) mod runs; + use std::future::Future; use forgejo_api::structs::CommitStatusState; use hive_jobq::scheduler::Outcome; use swarm_queue_client::builder::{BuildAgent, REPLY_DEADLINE, Reply, SUBJECT}; +use crate::agent_renewal::unix_now; use crate::forge::{self, CONFIG_ORG}; +use runs::{BuildOutcome, BuildRecord}; /// The commit status context the check writes. pub const CONTEXT: &str = "swarm-builder"; @@ -59,6 +64,7 @@ pub(crate) trait StatusSink { rev: &str, state: CommitStatusState, description: &str, + target_url: Option<&str>, ) -> impl Future> + Send; } @@ -69,21 +75,36 @@ impl StatusSink for forge::Client { rev: &str, state: CommitStatusState, description: &str, + target_url: Option<&str>, ) -> anyhow::Result<()> { forge::within_forge_deadline( "build_agent status", - self.create_commit_status(CONFIG_ORG, agent, rev, CONTEXT, state, description), + self.create_commit_status( + CONFIG_ORG, + agent, + rev, + forge::CommitStatus { + context: CONTEXT, + state, + description, + target_url, + }, + ), ) .await .unwrap_or_else(|| Err(anyhow::anyhow!("no answer from the forge"))) } } -/// Build `job` and write its status. The node fails when the build does, so -/// the Jobs page shows red where the PR does. +/// Build `job`, record it in `builds`, and write its status, linking to the +/// recorded build under `link_base` (the swarm vhost's public URL) when there is +/// one. The node fails when the build does, so the Jobs page shows red where +/// the PR does. pub(crate) async fn run( queue: Option<&impl BuildSink>, statuses: Option<&impl StatusSink>, + builds: &runs::Builds, + link_base: Option<&str>, job: BuildAgent, ) -> Outcome { let Some(statuses) = statuses else { @@ -93,22 +114,30 @@ pub(crate) async fn run( .to_owned(), ); }; - let set = |state, description: String| { + let started_unix = unix_now(); + let set = |state, description: String, link: Option| { let job = &job; async move { statuses - .set(&job.agent, &job.rev, state, &description) + .set(&job.agent, &job.rev, state, &description, link.as_deref()) .await .map_err(|e| format!("{e:#}")) } }; + let finish = + |outcome, text: String| record(builds, link_base, &job, started_unix, outcome, text); let Some(queue) = queue else { let error = "no swarm queue is configured on this host, so no build can be sent"; - return settle(set(CommitStatusState::Error, error.to_owned()).await, error); + let link = finish(BuildOutcome::Error, error.to_owned()).await; + return settle( + set(CommitStatusState::Error, error.to_owned(), link).await, + error, + ); }; if let Err(e) = set( CommitStatusState::Pending, "building on the swarm builder".to_owned(), + None, ) .await { @@ -121,22 +150,74 @@ pub(crate) async fn run( match queue.build(request).await { Ok(Reply::Built { out_path }) => { tracing::info!(agent = %job.agent, rev = %job.rev, %out_path, "build_agent: built"); - match set(CommitStatusState::Success, out_path).await { + let link = finish(BuildOutcome::Built, out_path.clone()).await; + match set(CommitStatusState::Success, out_path, link).await { Ok(()) => Outcome::Done, Err(e) => Outcome::Failed(format!("built, but the status was not written: {e}")), } } Ok(Reply::Failed { error_tail }) => { - let last = last_line(&error_tail); + let last = last_line(&error_tail).to_owned(); + let link = finish(BuildOutcome::Failed, error_tail.clone()).await; settle( - set(CommitStatusState::Failure, last.to_owned()).await, + set(CommitStatusState::Failure, last, link).await, &format!("the build failed: {error_tail}"), ) } - Err(error) => settle(set(CommitStatusState::Error, error.clone()).await, &error), + Err(error) => { + let link = finish(BuildOutcome::Error, error.clone()).await; + settle( + set(CommitStatusState::Error, error.clone(), link).await, + &error, + ) + } } } +/// Store the build of `job` that ended as `outcome`, `text` being its store +/// path, `nix`'s tail or the reason. The link to it, when it was +/// stored and `link_base` is known. +async fn record( + builds: &runs::Builds, + link_base: Option<&str>, + job: &BuildAgent, + started_unix: i64, + outcome: BuildOutcome, + text: String, +) -> Option { + let mut build = BuildRecord { + id: format!( + "{}-{}-{started_unix}", + job.agent, + job.rev.get(..12).unwrap_or(&job.rev) + ), + repo: format!("{CONFIG_ORG}/{}", job.agent), + pr: job.pr, + head_sha: job.rev.clone(), + started_unix, + finished_unix: unix_now(), + outcome, + out_path: None, + error_tail: None, + detail: None, + }; + *match outcome { + BuildOutcome::Built => &mut build.out_path, + BuildOutcome::Failed => &mut build.error_tail, + BuildOutcome::Error => &mut build.detail, + } = Some(text); + if !builds.record(&build).await { + return None; + } + link_base.map(|base| { + format!( + "{}/api/builder/runs/{}", + base.trim_end_matches('/'), + build.id + ) + }) +} + /// A failed node carrying `error`, and the status write's own failure when /// there was one. fn settle(written: Result<(), String>, error: &str) -> Outcome { @@ -157,10 +238,13 @@ fn last_line(tail: &str) -> &str { #[cfg(test)] mod tests { - use std::sync::Mutex; + use std::sync::{Arc, Mutex}; + use super::runs::tests::FakeDoc; use super::*; + const BASE: &str = "https://swarm.example"; + struct FakeQueue(Result); impl BuildSink for FakeQueue { @@ -172,7 +256,7 @@ mod tests { } #[derive(Default)] - struct FakeStatuses(Mutex>); + struct FakeStatuses(Mutex)>>); impl StatusSink for FakeStatuses { async fn set( @@ -181,13 +265,22 @@ mod tests { rev: &str, state: CommitStatusState, description: &str, + target_url: Option<&str>, ) -> anyhow::Result<()> { assert_eq!((agent, rev), ("atlas", "abc")); - self.0.lock().unwrap().push((state, description.to_owned())); + self.0.lock().unwrap().push(( + state, + description.to_owned(), + target_url.map(str::to_owned), + )); Ok(()) } } + fn store() -> runs::Builds { + runs::Builds::Fake(Arc::new(FakeDoc::default())) + } + fn job() -> BuildAgent { BuildAgent { agent: "atlas".to_owned(), @@ -198,9 +291,9 @@ mod tests { async fn statuses_after(queue: Option) -> (Outcome, Vec) { let statuses = FakeStatuses::default(); - let outcome = run(queue.as_ref(), Some(&statuses), job()).await; + let outcome = run(queue.as_ref(), Some(&statuses), &store(), Some(BASE), job()).await; let states = statuses.0.into_inner().unwrap(); - (outcome, states.into_iter().map(|(s, _)| s).collect()) + (outcome, states.into_iter().map(|(s, _, _)| s).collect()) } #[tokio::test] @@ -216,19 +309,114 @@ mod tests { ); } - #[tokio::test] - async fn a_failed_build_goes_red_with_its_last_line_and_fails_the_node() { + /// The build, the doc it was recorded in, and the statuses written. + async fn recorded( + reply: Result, + doc: FakeDoc, + link_base: Option<&str>, + ) -> ( + Outcome, + Vec, + Vec<(CommitStatusState, String, Option)>, + ) { let statuses = FakeStatuses::default(); - let queue = FakeQueue(Ok(Reply::Failed { - error_tail: "evaluating\nerror: boom\n".to_owned(), - })); - let outcome = run(Some(&queue), Some(&statuses), job()).await; + let doc = Arc::new(doc); + let outcome = run( + Some(&FakeQueue(reply)), + Some(&statuses), + &runs::Builds::Fake(Arc::clone(&doc)), + link_base, + job(), + ) + .await; + (outcome, doc.builds(), statuses.0.into_inner().unwrap()) + } + + #[tokio::test] + async fn a_failed_build_goes_red_with_its_last_line_and_links_to_its_recorded_tail() { + let tail = "evaluating\nerror: boom\n"; + let (outcome, builds, written) = recorded( + Ok(Reply::Failed { + error_tail: tail.to_owned(), + }), + FakeDoc::default(), + Some("https://swarm.example/"), + ) + .await; assert!(matches!(&outcome, Outcome::Failed(e) if e.contains("error: boom"))); - let written = statuses.0.into_inner().unwrap(); + let [build] = builds.as_slice() else { + panic!("one build recorded: {builds:?}") + }; + assert_eq!(build.repo, "agent-configs/atlas"); + assert_eq!((build.pr, build.head_sha.as_str()), (3, "abc")); + assert_eq!(build.outcome, BuildOutcome::Failed); + assert_eq!(build.error_tail.as_deref(), Some(tail)); + assert_eq!((&build.out_path, &build.detail), (&None, &None)); + assert!(build.id.starts_with("atlas-abc-"), "{}", build.id); + assert!(build.finished_unix >= build.started_unix); + let link = format!("https://swarm.example/api/builder/runs/{}", build.id); + assert_eq!( + written, + [ + ( + CommitStatusState::Pending, + "building on the swarm builder".to_owned(), + None + ), + ( + CommitStatusState::Failure, + "error: boom".to_owned(), + Some(link) + ), + ] + ); + } + + #[tokio::test] + async fn a_built_config_and_a_missing_builder_are_recorded_and_linked_too() { + let (_, builds, written) = recorded( + Ok(Reply::Built { + out_path: "/nix/store/x".to_owned(), + }), + FakeDoc::default(), + Some(BASE), + ) + .await; + assert_eq!(builds[0].outcome, BuildOutcome::Built); + assert_eq!(builds[0].out_path.as_deref(), Some("/nix/store/x")); + assert!(written[1].2.is_some(), "{written:?}"); + + let (_, builds, written) = recorded( + Err("no swarm builder".to_owned()), + FakeDoc::default(), + Some(BASE), + ) + .await; + assert_eq!(builds[0].outcome, BuildOutcome::Error); + assert_eq!(builds[0].detail.as_deref(), Some("no swarm builder")); + assert_eq!(written[1].0, CommitStatusState::Error); + assert!(written[1].2.is_some(), "{written:?}"); + } + + /// No link to a build that was not stored, nor without the swarm's URL; + /// the status is written either way. + #[tokio::test] + async fn a_status_links_only_to_a_stored_build_under_a_known_url() { + let failed = || { + Ok(Reply::Failed { + error_tail: "error: boom".to_owned(), + }) + }; + let (_, builds, written) = recorded(failed(), FakeDoc::failing(), Some(BASE)).await; + assert!(builds.is_empty()); assert_eq!( written.last(), - Some(&(CommitStatusState::Failure, "error: boom".to_owned())) + Some(&(CommitStatusState::Failure, "error: boom".to_owned(), None)) ); + + let (_, builds, written) = recorded(failed(), FakeDoc::default(), None).await; + assert_eq!(builds.len(), 1, "recorded all the same"); + assert_eq!(written.last().map(|w| &w.2), Some(&None)); } /// A superseded build that finishes after the newer head's writes only on @@ -259,6 +447,7 @@ mod tests { rev: &str, state: CommitStatusState, _: &str, + _: Option<&str>, ) -> anyhow::Result<()> { self.0.lock().unwrap().push((rev.to_owned(), state)); Ok(()) @@ -270,8 +459,22 @@ mod tests { rev: rev.to_owned(), ..job() }; - run(Some(&ByRevQueue), Some(&statuses), at("new")).await; - run(Some(&ByRevQueue), Some(&statuses), at("old")).await; + run( + Some(&ByRevQueue), + Some(&statuses), + &store(), + None, + at("new"), + ) + .await; + run( + Some(&ByRevQueue), + Some(&statuses), + &store(), + None, + at("old"), + ) + .await; let written = statuses.0.into_inner().unwrap(); let last_on = |rev: &str| { written @@ -310,7 +513,14 @@ mod tests { panic!("a build with nowhere to report must not be sent") } } - let outcome = run(Some(&Unreachable), None::<&FakeStatuses>, job()).await; + let outcome = run( + Some(&Unreachable), + None::<&FakeStatuses>, + &store(), + None, + job(), + ) + .await; assert!(matches!(&outcome, Outcome::Failed(e) if e.contains("no forge access"))); } diff --git a/swarm-controller/src/builder/runs.rs b/swarm-controller/src/builder/runs.rs new file mode 100644 index 00000000..5a669a0f --- /dev/null +++ b/swarm-controller/src/builder/runs.rs @@ -0,0 +1,495 @@ +//! `GET /api/builder/runs`: the swarm builder's finished builds, newest +//! first, each with what `nix` printed when it failed; and +//! `GET /api/builder/runs/{id}`, one of them, which its commit status links to. +//! +//! The newest [`MAX_RUNS`] are one JSON array under [`KEY`] in [`BUCKET`], +//! which only this daemon reads or writes, kept the way +//! `review_classifier::runs` keeps the classifier's: one compare-and-set +//! document, with entries this build cannot decode kept as they were. + +use std::future::Future; + +use anyhow::Context as _; +use async_nats::jetstream::kv::Store; +use axum::Json; +use axum::extract::{Path, State}; +use axum::http::StatusCode; +use problem_details::ProblemDetails; +use serde::{Deserialize, Serialize}; +use utoipa::ToSchema; + +use crate::schedules::{read_with, write}; +use crate::{AppState, error_problem}; + +/// The bucket holding the builds. Agents hold no grant on it. +pub const BUCKET: &str = "builder-runs"; + +/// The one key in [`BUCKET`]. +pub const KEY: &str = "recent"; + +/// How many builds the list keeps; recording one more drops the oldest. +pub(crate) const MAX_RUNS: usize = 100; + +/// How many times a losing writer re-reads and re-applies. +const MAX_ATTEMPTS: usize = 5; + +/// How a build ended, as its commit status reads. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)] +#[serde(rename_all = "kebab-case")] +pub(crate) enum BuildOutcome { + /// The config built: `success`. + Built, + /// `nix` failed or timed out: `failure`. + Failed, + /// No build ran or none came back: `error`. + Error, +} + +/// One finished build. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema)] +pub(crate) struct BuildRecord { + /// `--`, the last segment + /// of the link on the build's status. + pub(crate) id: String, + /// `agent-configs/`. + pub(crate) repo: String, + pub(crate) pr: u64, + /// The commit built. + pub(crate) head_sha: String, + /// Unix seconds: when the node sent the build, and when its answer came. + pub(crate) started_unix: i64, + pub(crate) finished_unix: i64, + pub(crate) outcome: BuildOutcome, + /// The system closure's store path, for `built`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(crate) out_path: Option, + /// The last lines `nix` printed, for `failed`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(crate) error_tail: Option, + /// Why no build ran or came back, for `error`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(crate) detail: Option, +} + +/// One entry of the stored list. +#[derive(Debug, Clone, PartialEq, Serialize)] +#[serde(untagged)] +pub(crate) enum Stored { + Build(BuildRecord), + /// A record this build cannot decode, such as one a newer build wrote. + /// Left out of the list but kept, counted toward [`MAX_RUNS`] and written + /// back as it was read. + Unreadable(serde_json::Value), +} + +/// Where the builds are kept: [`BUCKET`] in production, a fake in the tests. +pub(crate) trait BuildDoc { + /// The entries, with the revision to write them back against. + fn read(&self) -> impl Future, Option)>> + Send; + /// `Ok(false)` when another write landed since `revision`. + fn write( + &self, + builds: Vec, + revision: Option, + ) -> impl Future> + Send; +} + +impl BuildDoc for Store { + async fn read(&self) -> anyhow::Result<(Vec, Option)> { + read_with(self, KEY, |bytes| Ok(decode(bytes))).await + } + + async fn write(&self, builds: Vec, revision: Option) -> anyhow::Result { + write(self, KEY, &builds, revision).await + } +} + +/// The entries in `bytes`. A document that is not a JSON array reads as no +/// entries, with a warning, and the next write replaces it. +fn decode(bytes: &[u8]) -> Vec { + let values: Vec = match serde_json::from_slice(bytes) { + Ok(values) => values, + Err(e) => { + tracing::warn!( + bucket = BUCKET, key = KEY, error = %e, + "the builder runs are not decodable; reading them as none" + ); + return Vec::new(); + } + }; + values + .into_iter() + .map(|value| match BuildRecord::deserialize(&value) { + Ok(build) => Stored::Build(build), + Err(e) => { + tracing::warn!( + bucket = BUCKET, key = KEY, error = %e, + "a builder run this build cannot decode is left out of the list and kept" + ); + Stored::Unreadable(value) + } + }) + .collect() +} + +/// Put `build` first in `doc`, keeping the newest [`MAX_RUNS`], re-reading and +/// re-applying when another write landed in between. +async fn push(doc: &impl BuildDoc, build: &BuildRecord) -> anyhow::Result<()> { + for _ in 0..MAX_ATTEMPTS { + let (mut builds, revision) = doc.read().await?; + builds.insert(0, Stored::Build(build.clone())); + builds.truncate(MAX_RUNS); + if doc.write(builds, revision).await? { + return Ok(()); + } + tracing::debug!("builder runs: write lost a race, retrying"); + } + anyhow::bail!("gave up recording the build after {MAX_ATTEMPTS} conflicting writes") +} + +/// The builds' store. +#[derive(Clone)] +pub(crate) enum Builds { + /// [`BUCKET`] on the swarm queue, when this host has one. + Live(Option), + /// An in-memory document a test reads back. + #[cfg(test)] + Fake(std::sync::Arc), +} + +/// Why the builds' store could not be reached: always 503. +struct Unreachable(String); + +impl Builds { + /// Add `build` to the list. `false`, with a warning, when the write + /// failed. + pub(crate) async fn record(&self, build: &BuildRecord) -> bool { + let pushed = match self { + Self::Live(client) => match open(client.as_ref()).await { + Ok(store) => push(&store, build).await, + Err(Unreachable(e)) => { + Err(anyhow::anyhow!(e).context(format!("opening the {BUCKET} bucket"))) + } + }, + #[cfg(test)] + Self::Fake(doc) => push(&**doc, build).await, + }; + match pushed { + Ok(()) => true, + Err(e) => { + tracing::warn!( + id = %build.id, repo = %build.repo, head_sha = %build.head_sha, + error = %format!("{e:#}"), + "storing a builder run failed; its status links to nothing" + ); + false + } + } + } + + async fn list(&self) -> Result, ProblemDetails> { + let read = match self { + Self::Live(client) => { + let store = open(client.as_ref()) + .await + .map_err(|Unreachable(e)| error_problem(StatusCode::SERVICE_UNAVAILABLE, &e))?; + BuildDoc::read(&store).await + } + #[cfg(test)] + Self::Fake(doc) => doc.read().await, + }; + read.map(|(builds, _)| { + builds + .into_iter() + .filter_map(|entry| match entry { + Stored::Build(build) => Some(build), + Stored::Unreadable(_) => None, + }) + .collect() + }) + .map_err(|e| { + let detail = format!("{e:#}"); + tracing::warn!(error = %detail, "reading the builder runs failed"); + error_problem(StatusCode::INTERNAL_SERVER_ERROR, &detail) + }) + } +} + +/// Open [`BUCKET`], creating it if nothing has yet. +async fn open(client: Option<&async_nats::Client>) -> Result { + let client = client + .ok_or_else(|| Unreachable("no swarm queue is configured on this host".to_owned()))?; + swarm_queue_client::ensure_connected(client) + .map_err(|e| Unreachable(swarm_queue_client::chain(&e)))?; + let js = async_nats::jetstream::new(client.clone()); + if let Ok(store) = js.get_key_value(BUCKET).await { + return Ok(store); + } + js.create_key_value(async_nats::jetstream::kv::Config { + bucket: BUCKET.to_owned(), + description: "The swarm builder's newest finished builds".to_owned(), + history: 1, + ..Default::default() + }) + .await + .with_context(|| format!("creating the {BUCKET} bucket")) + .map_err(|e| Unreachable(format!("{e:#}"))) +} + +/// List the swarm builder's finished builds. +#[utoipa::path( + get, + path = "/api/builder/runs", + responses( + (status = 200, description = "the newest 100 finished builds, newest first; empty \ + when there are none", body = Vec), + (status = 503, description = "no swarm queue is configured on this host, it is not \ + connected, or the builds' bucket could not be opened (problem+json)", body = String), + (status = 500, description = "the builds could not be read (problem+json)", + body = String), + ), + tag = "builder" +)] +pub(crate) async fn get_runs( + State(state): State, +) -> Result>, ProblemDetails> { + state.builds.list().await.map(Json) +} + +/// One finished build: the link a `swarm-builder` commit status carries. +#[utoipa::path( + get, + path = "/api/builder/runs/{id}", + params(("id" = String, Path, description = "the build's id, as the list names it")), + responses( + (status = 200, description = "the build, with `nix`'s tail when it failed", + body = BuildRecord), + (status = 404, description = "no build has this id among the newest 100 \ + (problem+json)", body = String), + (status = 503, description = "as for the list (problem+json)", body = String), + (status = 500, description = "as for the list (problem+json)", body = String), + ), + tag = "builder" +)] +pub(crate) async fn get_run( + State(state): State, + Path(id): Path, +) -> Result, ProblemDetails> { + state + .builds + .list() + .await? + .into_iter() + .find(|build| build.id == id) + .map(Json) + .ok_or_else(|| { + error_problem( + StatusCode::NOT_FOUND, + "no build has this id among the newest 100", + ) + }) +} + +#[cfg(test)] +pub(crate) mod tests { + use std::sync::{Arc, Mutex}; + + use axum::body::Body; + use axum::http::Request; + use tower::ServiceExt as _; + + use super::*; + + /// A document in memory. Each of the first `conflicts` writes loses to a + /// write that lands one build in between; a document set to `fail` fails + /// every read. + #[derive(Default)] + pub(crate) struct FakeDoc { + state: Mutex<(Vec, u64)>, + conflicts: Mutex, + fail: bool, + } + + impl FakeDoc { + pub(crate) fn failing() -> Self { + Self { + fail: true, + ..Self::default() + } + } + + /// The decodable builds, as the list shows them. + pub(crate) fn builds(&self) -> Vec { + self.state + .lock() + .expect("unpoisoned") + .0 + .iter() + .filter_map(|entry| match entry { + Stored::Build(build) => Some(build.clone()), + Stored::Unreadable(_) => None, + }) + .collect() + } + } + + impl BuildDoc for FakeDoc { + async fn read(&self) -> anyhow::Result<(Vec, Option)> { + anyhow::ensure!(!self.fail, "the store is down"); + let (builds, revision) = self.state.lock().expect("unpoisoned").clone(); + Ok((builds, Some(revision).filter(|r| *r > 0))) + } + + async fn write(&self, builds: Vec, revision: Option) -> anyhow::Result { + let mut state = self.state.lock().expect("unpoisoned"); + let mut conflicts = self.conflicts.lock().expect("unpoisoned"); + if *conflicts > 0 { + *conflicts -= 1; + state + .0 + .insert(0, Stored::Build(record(&format!("other-{conflicts}")))); + state.1 += 1; + } + if revision.unwrap_or(0) != state.1 { + return Ok(false); + } + *state = (builds, state.1 + 1); + Ok(true) + } + } + + pub(crate) fn record(id: &str) -> BuildRecord { + BuildRecord { + id: id.to_owned(), + repo: "agent-configs/atlas".to_owned(), + pr: 3, + head_sha: "abc".to_owned(), + started_unix: 1_000, + finished_unix: 1_060, + outcome: BuildOutcome::Failed, + out_path: None, + error_tail: Some("evaluating\nerror: boom".to_owned()), + detail: None, + } + } + + fn ids(builds: &[BuildRecord]) -> Vec<&str> { + builds.iter().map(|b| b.id.as_str()).collect() + } + + #[tokio::test] + async fn the_newest_hundred_survive_newest_first() { + let doc = FakeDoc::default(); + for i in 0..=MAX_RUNS { + push(&doc, &record(&format!("r{i}"))).await.expect("stored"); + } + let builds = doc.builds(); + assert_eq!(builds.len(), 100); + assert_eq!(builds[0].id, "r100"); + assert_eq!(builds[99].id, "r1", "r0, the oldest, is gone"); + } + + #[tokio::test] + async fn a_write_that_lost_a_race_is_redone_on_the_winner() { + let doc = FakeDoc { + conflicts: Mutex::new(2), + ..FakeDoc::default() + }; + push(&doc, &record("mine")).await.expect("stored"); + assert_eq!(ids(&doc.builds()), ["mine", "other-0", "other-1"]); + } + + /// The bucket on a real server: created on first use, written, read + /// back; a record from a newer build is left out and kept in its place. + #[tokio::test] + async fn builds_round_trip_through_the_bucket_and_keep_an_undecodable_one() { + let nats = crate::test_nats::TestNats::start(); + let client = nats.client().await; + let builds = Builds::Live(Some(client.clone())); + assert!(builds.record(&record("a")).await); + assert!(builds.record(&record("b")).await); + assert_eq!(ids(&builds.list().await.expect("listed")), ["b", "a"]); + + let Ok(store) = open(Some(&client)).await else { + panic!("the bucket did not open"); + }; + let mut newer = serde_json::to_value(record("newer")).expect("encodes"); + newer["outcome"] = "a-later-outcome".into(); + let doc = serde_json::json!([newer, record("b"), record("a")]); + store + .put(KEY, doc.to_string().into()) + .await + .expect("written"); + assert!(builds.record(&record("c")).await); + assert_eq!(ids(&builds.list().await.expect("listed")), ["c", "b", "a"]); + let entry = store.get(KEY).await.expect("readable").expect("a document"); + let after: Vec = serde_json::from_slice(&entry).expect("a JSON array"); + assert_eq!(after.len(), 4, "{after:?}"); + assert_eq!(after[1], newer, "kept untouched, in its place"); + } + + #[tokio::test] + async fn a_failed_write_is_reported() { + assert!( + !Builds::Fake(Arc::new(FakeDoc::failing())) + .record(&record("a")) + .await + ); + assert!(!Builds::Live(None).record(&record("a")).await); + } + + #[tokio::test] + async fn the_route_lists_the_builds_with_their_tail_and_a_down_store_is_an_error() { + let doc = Arc::new(FakeDoc::default()); + let mut built = record("a"); + built.outcome = BuildOutcome::Built; + built.out_path = Some("/nix/store/x".to_owned()); + built.error_tail = None; + push(&*doc, &built).await.expect("stored"); + push(&*doc, &record("b")).await.expect("stored"); + + let get_at = |builds: Builds, path: &'static str| async move { + let (mut state, _) = crate::tests::state_with_roster(); + state.builds = builds; + let res = crate::build_app(state) + .oneshot(Request::get(path).body(Body::empty()).expect("a request")) + .await + .expect("infallible"); + let status = res.status(); + let bytes = axum::body::to_bytes(res.into_body(), usize::MAX) + .await + .expect("a body"); + let body: serde_json::Value = serde_json::from_slice(&bytes).expect("JSON"); + (status, body) + }; + let get = |builds: Builds| get_at(builds, "/api/builder/runs"); + let (status, body) = get(Builds::Fake(Arc::clone(&doc))).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body[0]["id"], "b"); + assert_eq!(body[0]["repo"], "agent-configs/atlas"); + assert_eq!(body[0]["head_sha"], "abc"); + assert_eq!(body[0]["outcome"], "failed"); + assert_eq!(body[0]["error_tail"], "evaluating\nerror: boom"); + assert!(body[0].get("out_path").is_none(), "{body}"); + assert_eq!(body[1]["outcome"], "built"); + assert_eq!(body[1]["out_path"], "/nix/store/x"); + assert!(body[1].get("error_tail").is_none(), "{body}"); + + let one = |path| get_at(Builds::Fake(Arc::clone(&doc)), path); + let (status, body) = one("/api/builder/runs/b").await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["id"], "b"); + assert_eq!(body["error_tail"], "evaluating\nerror: boom"); + let (status, body) = one("/api/builder/runs/a").await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["outcome"], "built"); + let (status, body) = one("/api/builder/runs/nope").await; + assert_eq!(status, StatusCode::NOT_FOUND, "{body}"); + + let (status, body) = get(Builds::Fake(Arc::new(FakeDoc::failing()))).await; + assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR, "{body}"); + let (status, body) = get(Builds::Live(None)).await; + assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE, "{body}"); + assert_eq!(body["detail"], "no swarm queue is configured on this host"); + } +} diff --git a/swarm-controller/src/forge.rs b/swarm-controller/src/forge.rs index 94d6d4b6..75199cc7 100644 --- a/swarm-controller/src/forge.rs +++ b/swarm-controller/src/forge.rs @@ -62,6 +62,16 @@ pub mod site_admin; /// sends, in characters. The forge keeps more; its PR view shows one line. pub const STATUS_DESCRIPTION_MAX: usize = 200; +/// One commit status, as [`Client::create_commit_status`] writes it. +pub struct CommitStatus<'a> { + pub context: &'a str, + pub state: CommitStatusState, + /// Cut to [`STATUS_DESCRIPTION_MAX`] characters. + pub description: &'a str, + /// Where the status links to, if anywhere. + pub target_url: Option<&'a str>, +} + /// `description` cut to [`STATUS_DESCRIPTION_MAX`] characters, an ellipsis /// marking a cut. fn status_description(description: &str) -> String { @@ -1213,9 +1223,7 @@ impl Client { Ok(rows.into_iter().flatten().collect()) } - /// Set the commit status named `context` on `sha` of `owner/repo` to - /// `state`, with `description` cut to [`STATUS_DESCRIPTION_MAX`] - /// characters. + /// Set `status` on `sha` of `owner/repo`. /// /// # Errors /// The forge refused the status. @@ -1224,15 +1232,14 @@ impl Client { owner: &str, repo: &str, sha: &str, - context: &str, - state: CommitStatusState, - description: &str, + status: CommitStatus<'_>, ) -> Result<()> { + let context = status.context; let status = CreateStatusOption { context: Some(context.to_owned()), - description: Some(status_description(description)), - state: Some(state), - target_url: None, + description: Some(status_description(status.description)), + state: Some(status.state), + target_url: status.target_url.map(str::to_owned), }; self.api .call(|api| { @@ -2796,9 +2803,12 @@ mod tests { CONFIG_ORG, "atlas", "abc", - "swarm-builder", - CommitStatusState::Failure, - "error: boom", + CommitStatus { + context: "swarm-builder", + state: CommitStatusState::Failure, + description: "error: boom", + target_url: Some("https://swarm.example/api/builder/runs/r1"), + }, ) .await .expect("the stub accepts the status"); @@ -2809,5 +2819,9 @@ mod tests { assert_eq!(body["context"], "swarm-builder"); assert_eq!(body["state"], "failure"); assert_eq!(body["description"], "error: boom"); + assert_eq!( + body["target_url"], + "https://swarm.example/api/builder/runs/r1" + ); } } diff --git a/swarm-controller/src/forge_account.rs b/swarm-controller/src/forge_account.rs index f1262c63..0856130a 100644 --- a/swarm-controller/src/forge_account.rs +++ b/swarm-controller/src/forge_account.rs @@ -169,6 +169,7 @@ mod tests { try_outs: super::super::review_classifier::TryOuts::default(), reviews: std::sync::Arc::default(), runs: crate::review_classifier::runs::Runs::Live(None), + builds: crate::builder::runs::Builds::Live(None), alerts: None, } } diff --git a/swarm-controller/src/github_account.rs b/swarm-controller/src/github_account.rs index 857fe3cd..fb0d124f 100644 --- a/swarm-controller/src/github_account.rs +++ b/swarm-controller/src/github_account.rs @@ -139,6 +139,7 @@ mod tests { try_outs: super::super::review_classifier::TryOuts::default(), reviews: std::sync::Arc::default(), runs: crate::review_classifier::runs::Runs::Live(None), + builds: crate::builder::runs::Builds::Live(None), alerts: None, } } diff --git a/swarm-controller/src/linked_accounts.rs b/swarm-controller/src/linked_accounts.rs index 9650a657..1ad46944 100644 --- a/swarm-controller/src/linked_accounts.rs +++ b/swarm-controller/src/linked_accounts.rs @@ -978,6 +978,7 @@ pub(crate) mod tests { try_outs: super::super::review_classifier::TryOuts::default(), reviews: std::sync::Arc::default(), runs: crate::review_classifier::runs::Runs::Live(None), + builds: crate::builder::runs::Builds::Live(None), alerts: None, } } diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 0a09cbc3..d244e1e3 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -600,7 +600,16 @@ async fn build_agent( pr: u64, ) -> hive_jobq::scheduler::Outcome { let job = swarm_queue_client::builder::BuildAgent { agent, rev, pr }; - builder::run(deps.queue.as_ref(), forge.as_deref(), job).await + let builds = builder::runs::Builds::Live(deps.queue.clone()); + let link_base = std::env::var(PUBLIC_URL_ENV).ok(); + builder::run( + deps.queue.as_ref(), + forge.as_deref(), + &builds, + link_base.as_deref(), + job, + ) + .await } /// The `MintBuilderForgeToken` arm, lifted out for the same reason. @@ -887,6 +896,7 @@ fn socket_path() -> PathBuf { (name = "review-classifier", description = "trying review classifications against a PR (swarm-ui's try-out page), and the classifier's swarm-wide settings"), (name = "pulls", description = "open pull requests, for the try-out page's dropdowns"), (name = "alerts", description = "the swarm Grafana's current alerts (swarm-ui's alerts page)"), + (name = "builder", description = "the swarm builder's finished builds of agent config PRs, which their commit statuses link to"), ) )] struct ApiDoc; @@ -1022,6 +1032,8 @@ struct AppState { reviews: Arc, /// The review classifier's finished runs. runs: review_classifier::runs::Runs, + /// The swarm builder's finished builds. + builds: builder::runs::Builds, /// Grafana's alerts API, for `GET /api/alerts`. `None` when no Grafana URL /// is configured. alerts: Option>, @@ -3622,7 +3634,6 @@ async fn main() -> Result<()> { // what `policy::render_agent` grants today reaches them only from here. read_policy::ensure_agent_policies().await; - let runs = review_classifier::runs::Runs::Live(status.as_ref().map(|s| s.queue_client())); let state = AppState { hives: Arc::new(hives), links: Arc::new(load_links()), @@ -3631,6 +3642,8 @@ async fn main() -> Result<()> { agent_icons: agent_icon_reader(status.as_ref()), reported_inference: reported_inference(status.as_ref()), agent_login: agent_login(status.as_ref()), + runs: review_classifier::runs::Runs::Live(status.as_ref().map(|s| s.queue_client())), + builds: builder::runs::Builds::Live(status.as_ref().map(|s| s.queue_client())), status, jobq, webhook_secret, @@ -3642,7 +3655,6 @@ async fn main() -> Result<()> { accounts: linked_accounts::Accounts::Live, try_outs: review_classifier::TryOuts::default(), reviews: Arc::default(), - runs, alerts: alerts::from_env(), }; @@ -3734,6 +3746,8 @@ fn build_app(state: AppState) -> axum::Router { .routes(routes!(review_classifier::get_rules)) .routes(routes!(review_classifier::open_pr::post_open_pr)) .routes(routes!(review_classifier::runs::get_runs)) + .routes(routes!(builder::runs::get_runs)) + .routes(routes!(builder::runs::get_run)) .routes(routes!( review_classifier::settings::get_settings, review_classifier::settings::put_settings @@ -3862,6 +3876,7 @@ mod tests { try_outs: super::review_classifier::TryOuts::default(), reviews: std::sync::Arc::default(), runs: super::review_classifier::runs::Runs::Live(None), + builds: super::builder::runs::Builds::Live(None), alerts: None, }; (state, sched) diff --git a/swarm-controller/src/matrix_account.rs b/swarm-controller/src/matrix_account.rs index 5d3d6f22..a79082a1 100644 --- a/swarm-controller/src/matrix_account.rs +++ b/swarm-controller/src/matrix_account.rs @@ -473,6 +473,7 @@ mod tests { try_outs: super::super::review_classifier::TryOuts::default(), reviews: std::sync::Arc::default(), runs: crate::review_classifier::runs::Runs::Live(None), + builds: crate::builder::runs::Builds::Live(None), alerts: None, } }