mara/hyperhive
mara/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

Compare commits

base: mara:701efadfe46b2460d758096a5a1f43263dfa4c15
Branches Tags
mara:main
..
compare: mara:c28ca7aab58b67a39b484899bc07b9c848666f6f
Branches Tags
mara:main
7 changed files with 23 additions and 766 deletions
Download patch file Download diff file Expand all files Collapse all files

12
docs/swarm/builder.md
View file

@ -225,18 +225,6 @@ is the same case: the old builder has no subscriber for the new subject, so the
controller gets no responders at once, and the deploys go out after the builder
updates. Each round adds a build and deploy node per failing agent.
When the build belongs to a merged config PR and `nix` exited with an error or hit
its deadline, the controller also comments once on that PR, with its forge
account: the agent, the `internal/meta` commit of the first red build, `nix`'s last lines (at most 4000
bytes), and that the deploy waits and the swarm builds it again every round. A comment
carries a marker line with the agent, and the controller posts only if the PR
holds no comment of its own with that marker; a copy of the marker in someone else's comment is neither counted nor edited. The retry rounds stay quiet whatever commit
they build, and the latest error stays on the Jobs page. A green build of the
agent for that PR rewrites the marker and appends the build's commit to the comment, so a later red comments again.
A `no verdict` failure, and a build that isn't a
merged PR's (a new agent, the pin fan-out), leave no comment. When the forge
refuses the comment, the controller logs a warning and the node ends as before.
The controller exports the gauge `swarm.deploys.held`: the agents whose latest
build failed, with the label `reason` set to `failed` or `unavailable`. The Grafana
alert `hh-deploys-held` fires when the sum stays above zero for 30 minutes, so

6
swarm-controller/src/config_pr.rs
View file

@ -73,13 +73,11 @@ struct WebhookBranch {
sha: Option<String>,
}
/// A config PR that was just merged: whose config, the commit to deploy, and
/// the PR.
/// A config PR that was just merged: whose config, and the commit to deploy.
#[derive(Debug, PartialEq, Eq)]
pub struct MergedConfigPr {
pub agent: String,
pub rev: String,
pub pr: u64,
}
/// The merge a verified `ConfigPr` delivery reports, if it reports one.
@ -110,7 +108,6 @@ pub fn merged(body: &[u8]) -> Option<MergedConfigPr> {
Some(MergedConfigPr {
agent: payload.repository.name,
rev,
pr: payload.pull_request.number,
})
}
@ -376,7 +373,6 @@ mod tests {
Some(super::MergedConfigPr {
agent: "damocles".to_owned(),
rev: "abc123".to_owned(),
pr: 5,
})
);
}

1
swarm-controller/src/forge.rs
View file

@ -55,7 +55,6 @@ pub mod legacy_tokens;
mod meta_repo;
pub mod objects;
pub mod operator_profiles;
pub(crate) mod pr_comment;
pub mod pull_diff;
pub mod review_request;
pub mod runner_token;

330
swarm-controller/src/forge/pr_comment.rs
View file

@ -1,330 +0,0 @@
//! Comments on a pull request, posted once per marker.
use anyhow::{Context, Result};
use forgejo_api::structs::{
Comment, CreateIssueCommentOption, EditIssueCommentOption, IssueGetCommentsQuery,
};
use super::Client;
impl Client {
/// Every comment on pull request `pr` of `owner/repo` written by this
/// client's own forge login. A comment by anyone else can carry a marker
/// (a quote-reply copies it) and is neither counted nor edited.
async fn own_comments(&self, owner: &str, repo: &str, pr: u64) -> Result<Vec<Comment>> {
let me = self
.api
.call(|api| api.user_get_current().into_future())
.await
.context("reading this client's own forge login")?
.login
.context("the forge returned no login for this client")?;
Ok(self
.comments(owner, repo, pr)
.await?
.into_iter()
.filter(|c| c.user.as_ref().and_then(|u| u.login.as_deref()) == Some(me.as_str()))
.collect())
}
/// Every comment on pull request `pr` of `owner/repo`.
async fn comments(&self, owner: &str, repo: &str, pr: u64) -> Result<Vec<Comment>> {
let index = i64::try_from(pr).context("PR number out of range")?;
self.api
.call(|api| {
Box::pin(
api.issue_get_comments(
owner,
repo,
index,
IssueGetCommentsQuery {
since: None,
before: None,
},
)
.all(),
)
})
.await
.with_context(|| format!("listing the comments on {owner}/{repo}#{pr}"))
}
/// Comment `body` on pull request `pr` of `owner/repo` unless a comment
/// there already contains `marker`. Whether a comment was posted.
///
/// The marker is what the comment carries, so the check holds across a
/// restart and across every caller. It reads the comments first, so two
/// concurrent calls with one marker can both post.
pub(crate) async fn comment_once(
&self,
owner: &str,
repo: &str,
pr: u64,
marker: &str,
body: &str,
) -> Result<bool> {
let index = i64::try_from(pr).context("PR number out of range")?;
let existing = self.own_comments(owner, repo, pr).await?;
if existing
.iter()
.any(|c| c.body.as_deref().is_some_and(|b| b.contains(marker)))
{
return Ok(false);
}
let option = CreateIssueCommentOption {
body: body.to_owned(),
updated_at: None,
};
self.api
.call(|api| {
api.issue_create_comment(owner, repo, index, option.clone())
.into_future()
})
.await
.with_context(|| format!("commenting on {owner}/{repo}#{pr}"))?;
Ok(true)
}
/// Rewrite every comment of this client's own on pull request `pr` of
/// `owner/repo` that contains `marker`: the marker becomes `cleared`, which
/// lets [`Self::comment_once`] post that marker again, and `resolved` is
/// appended on a line of its own. A comment the forge refuses to edit is
/// logged and skipped, so one refusal leaves the rest cleared. How many
/// comments were rewritten.
///
/// # Errors
///
/// The comments could not be listed.
pub(crate) async fn clear_marker(
&self,
owner: &str,
repo: &str,
pr: u64,
marker: &str,
cleared: &str,
resolved: &str,
) -> Result<usize> {
let mut rewritten = 0;
for comment in self.own_comments(owner, repo, pr).await? {
let (Some(id), Some(body)) = (comment.id, comment.body.as_deref()) else {
continue;
};
if !body.contains(marker) {
continue;
}
let option = EditIssueCommentOption {
body: format!("{}\n\n{resolved}", body.replace(marker, cleared)),
updated_at: None,
};
let edited = self
.api
.call(|api| {
api.issue_edit_comment(owner, repo, id, option.clone())
.into_future()
})
.await;
match edited {
Ok(_) => rewritten += 1,
Err(e) => tracing::warn!(
id, %owner, %repo, pr, error = %format!("{e:#}"),
"swarm forge: could not rewrite a resolved build comment"
),
}
}
Ok(rewritten)
}
}
#[cfg(test)]
pub(crate) mod tests {
use std::sync::{Arc, Mutex};
use axum::http::{Method, StatusCode, Uri};
use super::*;
/// The comments a stand-in forge holds, as the issue path they were
/// posted to and their body.
pub(crate) type Posted = Arc<Mutex<Vec<(String, String)>>>;
/// A comment whose body starts with this is written by someone other than
/// the client's own login.
pub(crate) const HUMAN: &str = "HUMAN:";
/// A comment whose body contains this is one the forge refuses to edit.
pub(crate) const REFUSED: &str = "REFUSED";
/// A stand-in forge that stores the comments posted to any issue path and
/// lists them back.
pub(crate) async fn stand_in() -> (Client, Posted) {
let held: Posted = Arc::default();
let state = Arc::clone(&held);
let app = axum::Router::new().fallback(
move |method: Method, uri: Uri, body: axum::body::Bytes| {
let state = Arc::clone(&state);
async move {
let path = uri.path().to_owned();
let mut held = state.lock().unwrap();
match method.as_str() {
"GET" if path == "/api/v1/user" => (
StatusCode::OK,
[("x-total-count", String::new())],
r#"{"login":"bot"}"#.to_owned(),
),
"GET" => {
let list: Vec<_> = held
.iter()
.enumerate()
.filter(|(_, (p, _))| *p == path)
.map(|(id, (_, body))| {
let author = if body.starts_with(HUMAN) { "human" } else { "bot" };
serde_json::json!({ "id": id, "body": body, "user": { "login": author } })
})
.collect();
let count = list.len().to_string();
(
StatusCode::OK,
[("x-total-count", count)],
serde_json::Value::from(list).to_string(),
)
}
"PATCH" => {
let sent: serde_json::Value = serde_json::from_slice(&body).unwrap();
let id: usize = path.rsplit('/').next().unwrap().parse().unwrap();
if held[id].1.contains(REFUSED) {
return (
StatusCode::FORBIDDEN,
[("x-total-count", String::new())],
r#"{"message":"refused"}"#.to_owned(),
);
}
held[id].1 = sent["body"].as_str().unwrap().to_owned();
(
StatusCode::OK,
[("x-total-count", String::new())],
serde_json::json!({ "id": id, "body": held[id].1 }).to_string(),
)
}
"POST" => {
let sent: serde_json::Value = serde_json::from_slice(&body).unwrap();
let text = sent["body"].as_str().unwrap().to_owned();
held.push((path, text.clone()));
(
StatusCode::CREATED,
[("x-total-count", String::new())],
serde_json::json!({ "id": held.len() - 1, "body": text })
.to_string(),
)
}
_ => (
StatusCode::NOT_FOUND,
[("x-total-count", String::new())],
r#"{"message":"no"}"#.to_owned(),
),
}
}
},
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let url = url::Url::parse(&format!("http://{}/", listener.local_addr().unwrap())).unwrap();
tokio::spawn(async move { axum::serve(listener, app).await });
(Client::for_test_forge(url), held)
}
#[tokio::test]
async fn a_marker_is_commented_once_and_a_new_marker_again() {
let (client, held) = stand_in().await;
let once = |marker: &'static str| {
let client = &client;
async move {
client
.comment_once("o", "r", 7, marker, &format!("{marker} text"))
.await
.unwrap()
}
};
assert!(once("<!-- a -->").await, "first comment is posted");
assert!(!once("<!-- a -->").await, "same marker is not posted again");
assert!(once("<!-- b -->").await, "control: another marker posts");
let held = held.lock().unwrap();
assert_eq!(held.len(), 2, "{held:?}");
assert_eq!(
held[0].0,
format!("/api/v1/repos/o/r/{}/7/comments", "issues")
);
}
fn issue_path() -> String {
format!("/api/v1/repos/o/r/{}/7/comments", "issues")
}
#[tokio::test]
async fn a_cleared_marker_can_be_posted_again_and_the_old_comment_says_resolved() {
let (client, held) = stand_in().await;
let (marker, cleared) = ("<!-- a -->", "<!-- a cleared -->");
assert!(
client
.comment_once("o", "r", 7, marker, marker)
.await
.unwrap()
);
let clear = || client.clear_marker("o", "r", 7, marker, cleared, "Resolved.");
assert_eq!(clear().await.unwrap(), 1);
assert_eq!(clear().await.unwrap(), 0);
assert!(
client
.comment_once("o", "r", 7, marker, marker)
.await
.unwrap()
);
let held = held.lock().unwrap();
assert_eq!(held[0].1, format!("{cleared}\n\nResolved."));
assert_eq!(held[1].1, marker);
}
/// A human's comment can carry the marker (a quote-reply copies it): it
/// neither counts as the bot's report nor is rewritten.
#[tokio::test]
async fn a_comment_by_someone_else_with_the_marker_is_neither_counted_nor_edited() {
let (client, held) = stand_in().await;
let marker = "<!-- a -->";
let quoted = format!("{HUMAN} > {marker}");
held.lock().unwrap().push((issue_path(), quoted.clone()));
assert!(
client
.comment_once("o", "r", 7, marker, marker)
.await
.unwrap(),
"the human's copy does not silence the bot"
);
let cleared = client
.clear_marker("o", "r", 7, marker, "<!-- c -->", "Resolved.")
.await
.unwrap();
assert_eq!(cleared, 1, "only the bot's own comment");
let held = held.lock().unwrap();
assert_eq!(held[0].1, quoted, "the human's comment is untouched");
assert!(held[1].1.contains("<!-- c -->"), "{:?}", held[1]);
}
/// The forge refusing one edit does not stop the next comment's clear.
#[tokio::test]
async fn a_refused_edit_still_clears_the_next_comment() {
let (client, held) = stand_in().await;
let marker = "<!-- a -->";
let refused = format!("{marker} {REFUSED}");
{
let mut held = held.lock().unwrap();
held.push((issue_path(), refused.clone()));
held.push((issue_path(), marker.to_owned()));
}
let cleared = client
.clear_marker("o", "r", 7, marker, "<!-- c -->", "Resolved.")
.await
.unwrap();
assert_eq!(cleared, 1);
let held = held.lock().unwrap();
assert_eq!(held[0].1, refused, "control: the refused one is as it was");
assert!(held[1].1.contains("<!-- c -->"), "{:?}", held[1]);
}
}

1
swarm-controller/src/held_deploys.rs
View file

@ -96,7 +96,6 @@ mod tests {
.append(
SwarmNodeKind::BuildMeta {
agent: (*agent).to_owned(),
pr: None,
},
Vec::new(),
None,

87
swarm-controller/src/jobq.rs
View file

@ -116,10 +116,7 @@ pub(crate) enum SwarmNodeKind {
///
/// Carries no rev: it reads the head when it runs, which in most DAGs is
/// after the `UpdateMeta` that commits it.
///
/// `pr` is the merged config PR the build belongs to, if any: a red `nix`
/// build is reported on it once.
BuildMeta { agent: String, pr: Option<u64> },
BuildMeta { agent: String },
/// Re-mint `agent`'s queue secret if it is still stored and old enough.
/// See `agent_renewal`.
///
@ -205,12 +202,10 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind {
| SwarmNodeKind::MintAgentIdentity { agent }
| SwarmNodeKind::MintAgentForgeToken { agent }
| SwarmNodeKind::MintAgentMatrixAccount { agent }
| SwarmNodeKind::BuildMeta { agent }
| SwarmNodeKind::RenewAgentQueueCredential { agent } => {
serde_json::json!({ "agent": agent })
}
SwarmNodeKind::BuildMeta { agent, pr } => {
serde_json::json!({ "agent": agent, "pr": pr })
}
SwarmNodeKind::MintHiveSenderToken { hive }
| SwarmNodeKind::MintHiveForgeToken { hive } => serde_json::json!({ "hive": hive }),
SwarmNodeKind::MintClassifierForgeToken
@ -419,7 +414,7 @@ async fn run_node_work(
SwarmNodeKind::BuildAgent { agent, rev, pr } => {
build_agent(&deps, forge, agent, rev, pr).await
}
SwarmNodeKind::BuildMeta { agent, pr } => build_meta(&deps, forge, agent, pr).await,
SwarmNodeKind::BuildMeta { agent } => build_meta(&deps, forge, agent).await,
SwarmNodeKind::RenewAgentQueueCredential { agent } => renew_queue_credential(&agent).await,
SwarmNodeKind::SetAgentWanted { hive, agent } => {
set_agent_wanted(deps.wanted.as_deref(), &hive, &agent).await
@ -607,9 +602,8 @@ async fn build_meta(
deps: &WorkerDeps,
forge: Option<Arc<forge::Client>>,
agent: String,
pr: Option<u64>,
) -> hive_jobq::scheduler::Outcome {
meta_build::run(forge.as_deref(), deps.queue.as_ref(), agent, pr).await
meta_build::run(forge.as_deref(), deps.queue.as_ref(), agent).await
}
/// What an `UpdateMeta` pass decided: the deploys to send, and the placements
@ -752,13 +746,11 @@ fn gated_deploy<'a>(
hive: &str,
agent: &str,
rev: Option<String>,
pr: Option<u64>,
update: Option<hive_jobq::builder::NodeGuid>,
) -> hive_jobq::builder::NodeRef<'a, SwarmNodeKind, SwarmResourceKind> {
let mut build = b
.node(SwarmNodeKind::BuildMeta {
agent: agent.to_owned(),
pr,
})
.needs(SwarmResourceKind::Builder);
if let Some(update) = update {
@ -793,7 +785,7 @@ fn declare_fan_out(
}
let deploys: Vec<hive_jobq::builder::NodeGuid> = placements
.iter()
.map(|(hive, agent)| gated_deploy(grown, hive, agent, None, None, None).guid())
.map(|(hive, agent)| gated_deploy(grown, hive, agent, None, None).guid())
.collect();
if let FanOut::Record { pin } = fan_out {
let mut record = grown.node(SwarmNodeKind::RecordFanout { pin: pin.clone() });
@ -1181,7 +1173,7 @@ pub(crate) async fn queue_merged_config_deploy(
state: &AppState,
merged: config_pr::MergedConfigPr,
) {
let config_pr::MergedConfigPr { agent, rev, pr } = merged;
let config_pr::MergedConfigPr { agent, rev } = merged;
let declared = match read_declarations(state, state.hives.iter()).await {
Ok(declared) => declared,
Err((hive, e)) => {
@ -1205,7 +1197,7 @@ pub(crate) async fn queue_merged_config_deploy(
.jobq
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.insert_job(None, |b| merged_deploy_job(b, &hive, &agent, &rev, pr));
.insert_job(None, |b| merged_deploy_job(b, &hive, &agent, &rev));
match inserted {
Ok(_) => tracing::info!(%agent, %rev, %hive, "config-pr merge: deploy queued"),
Err(e) => {
@ -1226,13 +1218,12 @@ fn merged_deploy_job(
hive: &str,
agent: &str,
rev: &str,
pr: u64,
) -> Vec<hive_jobq::builder::NodeGuid> {
let update_meta = b
.node(SwarmNodeKind::UpdateMeta)
.needs(SwarmResourceKind::Meta);
let root = update_meta.guid();
let _deploy = gated_deploy(b, hive, agent, Some(rev.to_owned()), Some(pr), Some(root));
let _deploy = gated_deploy(b, hive, agent, Some(rev.to_owned()), Some(root));
vec![root]
}
@ -1356,7 +1347,7 @@ pub(crate) fn declare_agent_job(
// queue at all fails this node — but `TriggerDeploy` has nothing to
// publish to on such a host either, so the strict edge cancels a node
// that could not have succeeded anyway.
let trigger_deploy = gated_deploy(b, hive, agent, None, None, Some(update_meta.guid()))
let trigger_deploy = gated_deploy(b, hive, agent, None, Some(update_meta.guid()))
.after_any(mint_identity)
.after_any(mint_forge_token)
// `after_any` for the same reason: the container reads this token
@ -1570,9 +1561,6 @@ struct SkippedDeploy {
hive: String,
agent: String,
rev: Option<String>,
/// The merged config PR the failed build belonged to, so the build
/// declared again reports on it too.
pr: Option<u64>,
cause: SkipCause,
}
@ -1604,14 +1592,13 @@ fn skipped_deploys(
let SwarmNodeKind::TriggerDeploy { hive, agent, rev } = &n.payload else {
return None;
};
let dep_nodes = || {
n.deps.iter().filter_map(|d| match d {
hive_jobq::Dep::Node { id, .. } => graph.node(*id),
hive_jobq::Dep::Resource { .. } => None,
})
};
let failed = |is: fn(&SwarmNodeKind) -> bool| {
dep_nodes().any(|dep| is(&dep.payload) && dep.state == hive_jobq::State::Failed)
n.deps.iter().any(|d| match d {
hive_jobq::Dep::Node { id, .. } => graph.node(*id).is_some_and(|dep| {
is(&dep.payload) && dep.state == hive_jobq::State::Failed
}),
hive_jobq::Dep::Resource { .. } => false,
})
};
let cause = if failed(|p| matches!(p, SwarmNodeKind::UpdateMeta)) {
SkipCause::FailedUpdate
@ -1620,15 +1607,10 @@ fn skipped_deploys(
} else {
return None;
};
let pr = dep_nodes().find_map(|dep| match &dep.payload {
SwarmNodeKind::BuildMeta { pr, .. } => *pr,
_ => None,
});
Some(SkippedDeploy {
hive: hive.clone(),
agent: agent.clone(),
rev: rev.clone(),
pr,
cause,
})
})
@ -1684,7 +1666,7 @@ fn queue_meta_update(
});
let deploys: Vec<_> = skipped
.into_iter()
.map(|s| gated_deploy(b, &s.hive, &s.agent, s.rev, s.pr, update).guid())
.map(|s| gated_deploy(b, &s.hive, &s.agent, s.rev, update).guid())
.collect();
update.map_or(deploys, |update| vec![update])
})
@ -2312,7 +2294,7 @@ mod tests {
}
fn merged_dag(b: &TestBuilder) {
let _ = super::merged_deploy_job(b, "pr1ma", "atlas", "abc", 7);
let _ = super::merged_deploy_job(b, "pr1ma", "atlas", "abc");
}
/// A deploy needs the meta update to succeed: its edge refuses a failed
@ -2908,7 +2890,7 @@ mod tests {
.filter(|n| n.payload.label() == label)
.filter(|n| n.state == State::Pending)
.map(|n| match &n.payload {
SwarmNodeKind::BuildMeta { agent, .. }
SwarmNodeKind::BuildMeta { agent }
| SwarmNodeKind::TriggerDeploy { agent, .. } => agent.clone(),
other => panic!("not a build or a deploy: {other:?}"),
})
@ -3829,27 +3811,12 @@ mod tests {
nodes.into_iter().map(|(_, rev)| rev).collect()
}
/// The `pr` of `agent`'s builds, oldest first.
fn build_prs(sched: &SharedSched, agent: &str) -> Vec<Option<u64>> {
let guard = sched.lock().unwrap();
let mut nodes: Vec<_> = guard
.graph()
.nodes()
.filter_map(|n| match &n.payload {
SwarmNodeKind::BuildMeta { agent: a, pr } if a == agent => Some((n.id, *pr)),
_ => None,
})
.collect();
nodes.sort_by_key(|(id, _)| *id);
nodes.into_iter().map(|(_, pr)| pr).collect()
}
fn merged_atlas() -> SharedSched {
let sched = fresh_sched();
sched
.lock()
.unwrap()
.insert_job(None, |b| merged_deploy_job(b, "pr1ma", "atlas", "abc", 7))
.insert_job(None, |b| merged_deploy_job(b, "pr1ma", "atlas", "abc"))
.expect("the graph inserts");
sched
}
@ -3898,27 +3865,11 @@ mod tests {
.iter()
.all(|rev| rev.as_deref() == Some("abc"))
);
// So does the PR it belongs to: every retried build reports on it too.
assert_eq!(build_prs(&sched, "atlas"), [Some(7); 4]);
assert!(
!queue_meta_update(&sched).expect("tick"),
"the success ends the retries"
);
}
/// Only a merged config PR's build carries a PR: the creation DAG's does
/// not, so its failure has nobody to comment on.
#[test]
fn only_a_merged_prs_build_carries_its_pr() {
let sched = merged_atlas();
sched
.lock()
.unwrap()
.insert_job(None, |b| declare_agent_job(b, "iris", "pr1ma", false))
.expect("the graph inserts");
assert_eq!(build_prs(&sched, "atlas"), [Some(7)]);
assert_eq!(build_prs(&sched, "iris"), [None]);
}
}
#[cfg(test)]

352
swarm-controller/src/meta_build.rs
View file

@ -19,12 +19,6 @@
//!
//! Both are built again on every round of the meta timer
//! (`jobq::queue_meta_update`) until a build succeeds.
//!
//! A `nix` failure of a merged config PR's build (`pr` set) also leaves one
//! comment on that PR, deduped by a marker in the comment per agent: the retry
//! rounds stay quiet whatever rev they build, until a green build of the agent
//! clears the marker. A no-verdict failure says nothing about the config and
//! leaves none.
use std::fmt::Display;
use std::future::Future;
@ -57,26 +51,6 @@ pub(crate) trait MetaRepo {
fn head(&self) -> impl Future<Output = Result<Option<String>>> + Send;
/// Make sure the builder's forge account can read the repo.
fn grant_builder_read(&self) -> impl Future<Output = Result<()>> + Send;
/// Comment `body` on `pr` of `agent`'s config repo unless a comment there
/// already contains `marker`. Whether a comment was posted.
fn comment_once(
&self,
agent: &str,
pr: u64,
marker: &str,
body: &str,
) -> impl Future<Output = Result<bool>> + Send;
/// Rewrite the comments on `pr` of `agent`'s config repo that contain
/// `marker` so they contain `cleared` and end with `resolved`, which lets
/// `marker` be posted again. Only the controller's own comments count.
fn clear_marker(
&self,
agent: &str,
pr: u64,
marker: &str,
cleared: &str,
resolved: &str,
) -> impl Future<Output = Result<usize>> + Send;
}
impl MetaRepo for forge::Client {
@ -93,37 +67,6 @@ impl MetaRepo for forge::Client {
)
.await
}
async fn comment_once(&self, agent: &str, pr: u64, marker: &str, body: &str) -> Result<bool> {
crate::jobq::forge_node(
"build_meta",
forge::Client::comment_once(self, forge::CONFIG_ORG, agent, pr, marker, body),
)
.await
}
async fn clear_marker(
&self,
agent: &str,
pr: u64,
marker: &str,
cleared: &str,
resolved: &str,
) -> Result<usize> {
crate::jobq::forge_node(
"build_meta",
forge::Client::clear_marker(
self,
forge::CONFIG_ORG,
agent,
pr,
marker,
cleared,
resolved,
),
)
.await
}
}
/// Where a build is sent and its closure recorded: the queue in production, a
@ -176,99 +119,11 @@ impl MetaBuilds for async_nats::Client {
}
}
/// The most of `nix`'s tail a PR comment carries, in bytes: the builder's own cap
/// on the tail it sends.
const COMMENT_TAIL_BYTES: usize = 4000;
/// The marker a PR's failure comment carries for `agent`: one per PR and agent,
/// whatever `internal/meta` rev a retry builds, until a green build clears it.
fn failure_marker(agent: &str) -> String {
format!("<!-- swarm-build-failed agent={agent} -->")
}
/// What a green build leaves in place of [`failure_marker`], so the next red
/// build comments again. It does not contain the failure marker.
fn cleared_marker(agent: &str) -> String {
format!("<!-- swarm-build-cleared agent={agent} -->")
}
/// The comment a red build leaves on its merged config PR, first line the
/// failure marker.
fn failure_comment(agent: &str, meta_rev: &str, error_tail: &str) -> String {
let marker = failure_marker(agent);
let mut start = error_tail.len().saturating_sub(COMMENT_TAIL_BYTES);
while !error_tail.is_char_boundary(start) {
start += 1;
}
let tail = error_tail[start..].replace("```", "'''");
format!(
"{marker}\nThe swarm could not build `{agent}` from `internal/meta` at `{meta_rev}` \
after this PR merged. The deploy is held and built again on every round until a \
build succeeds. This is the only comment while it stays red; the latest error is on \
the Jobs page.\n\nLast lines of `nix`:\n\n```\n{}\n```",
tail.trim_end()
)
}
/// Tell `pr` that its merge's build failed, once per agent while the build
/// stays red. A comment that cannot be posted is logged and changes nothing
/// about the node.
async fn report_failure(
head: &impl MetaRepo,
pr: u64,
agent: &str,
meta_rev: &str,
error_tail: &str,
) {
let body = failure_comment(agent, meta_rev, error_tail);
match head
.comment_once(agent, pr, &failure_marker(agent), &body)
.await
{
Ok(posted) => {
tracing::info!(%agent, pr, %meta_rev, posted, "build_meta: reported the failed build on the PR");
}
Err(e) => tracing::warn!(
%agent, pr, %meta_rev, error = %format!("{e:#}"),
"build_meta: could not comment on the PR"
),
}
}
/// A green build ends the red spell: clear `pr`'s failure marker for `agent`,
/// so a later red comments again. A failure is logged and changes nothing
/// about the node.
async fn clear_failure(head: &impl MetaRepo, pr: u64, agent: &str, meta_rev: &str) {
let resolved = format!(
"Resolved: built at `{}`.",
meta_rev.get(..12).unwrap_or(meta_rev)
);
match head
.clear_marker(
agent,
pr,
&failure_marker(agent),
&cleared_marker(agent),
&resolved,
)
.await
{
Ok(cleared) => {
tracing::debug!(%agent, pr, cleared, "build_meta: cleared the failure marker");
}
Err(e) => tracing::warn!(
%agent, pr, error = %format!("{e:#}"),
"build_meta: could not clear the failure marker on the PR"
),
}
}
/// Build `agent` from `internal/meta`'s head and record the closure.
pub(crate) async fn run(
head: Option<&impl MetaRepo>,
queue: Option<&impl MetaBuilds>,
agent: String,
pr: Option<u64>,
) -> Outcome {
let Some(head) = head else {
return no_verdict(
@ -300,14 +155,8 @@ pub(crate) async fn run(
match queue.build(request).await {
Err(why) => no_verdict(why),
Ok(FromMetaReply::Unavailable { reason }) => no_verdict(reason),
Ok(FromMetaReply::Failed { error_tail }) => {
if let Some(pr) = pr {
report_failure(head, pr, &agent, &meta_rev, &error_tail).await;
}
Outcome::Failed(error_tail)
}
Ok(FromMetaReply::Failed { error_tail }) => Outcome::Failed(error_tail),
Ok(FromMetaReply::Built { out_path, system }) => {
let built_rev = meta_rev.clone();
let closure = AgentClosure {
meta_rev,
store_path: out_path.clone(),
@ -315,12 +164,7 @@ pub(crate) async fn run(
built_unix: unix_now(),
};
match queue.record(&agent, &closure).await {
Ok(()) => {
if let Some(pr) = pr {
clear_failure(head, pr, &agent, &built_rev).await;
}
Outcome::Done
}
Ok(()) => Outcome::Done,
Err(e) => no_verdict(format!("built {out_path}, but recording it failed: {e:#}")),
}
}
@ -340,8 +184,6 @@ mod tests {
struct FakeHead {
head: Result<Option<&'static str>, &'static str>,
grant_fails: bool,
/// The forge a PR comment goes to; `None` panics on a comment.
comments: Option<std::sync::Arc<forge::Client>>,
}
impl FakeHead {
@ -349,7 +191,6 @@ mod tests {
Self {
head,
grant_fails: false,
comments: None,
}
}
}
@ -366,33 +207,6 @@ mod tests {
anyhow::ensure!(!self.grant_fails, "the forge refused the collaborator");
Ok(())
}
async fn comment_once(
&self,
agent: &str,
pr: u64,
marker: &str,
body: &str,
) -> Result<bool> {
let client = self.comments.as_ref().expect("no comment expected");
client
.comment_once(forge::CONFIG_ORG, agent, pr, marker, body)
.await
}
async fn clear_marker(
&self,
agent: &str,
pr: u64,
marker: &str,
cleared: &str,
resolved: &str,
) -> Result<usize> {
let client = self.comments.as_ref().expect("no clear expected");
client
.clear_marker(forge::CONFIG_ORG, agent, pr, marker, cleared, resolved)
.await
}
}
/// A queue that answers every build with `reply`, and keeps what it was
@ -442,11 +256,7 @@ mod tests {
}
async fn run_with(head: &FakeHead, queue: &FakeQueue) -> Outcome {
run(Some(head), Some(queue), "atlas".to_owned(), None).await
}
async fn run_for_pr(head: &FakeHead, queue: &FakeQueue, pr: u64) -> Outcome {
run(Some(head), Some(queue), "atlas".to_owned(), Some(pr)).await
run(Some(head), Some(queue), "atlas".to_owned()).await
}
#[tokio::test]
@ -564,7 +374,6 @@ mod tests {
head.as_ref(),
with_queue.then_some(&queue),
"atlas".to_owned(),
None,
)
.await;
let Outcome::Failed(error) = outcome else {
@ -575,161 +384,6 @@ mod tests {
}
}
const REV2: &str = "fedcba9876543210fedcba9876543210fedcba98";
fn nix_failed(tail: &str) -> FakeQueue {
FakeQueue::answering(Ok(FromMetaReply::Failed {
error_tail: tail.to_owned(),
}))
}
async fn head_with_comments(
rev: &'static str,
) -> (FakeHead, crate::forge::pr_comment::tests::Posted) {
let (client, posted) = crate::forge::pr_comment::tests::stand_in().await;
let head = FakeHead {
comments: Some(std::sync::Arc::new(client)),
..FakeHead::reading(Ok(Some(rev)))
};
(head, posted)
}
/// A `nix` failure of a merged config PR's build is one comment on that PR
/// with the agent, the `internal/meta` rev, `nix`'s tail and the retry.
#[tokio::test]
async fn a_red_build_of_a_merged_pr_comments_once_on_it() {
let (head, posted) = head_with_comments(REV).await;
let outcome = run_for_pr(&head, &nix_failed("error: builder for foo failed"), 7).await;
assert_eq!(
outcome,
Outcome::Failed("error: builder for foo failed".to_owned())
);
let posted = posted.lock().unwrap();
let [(path, body)] = posted.as_slice() else {
panic!("one comment: {posted:?}")
};
assert_eq!(
path,
&format!("/api/v1/repos/agent-configs/atlas/{}/7/comments", "issues")
);
assert!(body.contains("`atlas`"), "{body}");
assert!(body.contains(REV), "{body}");
assert!(body.contains("error: builder for foo failed"), "{body}");
assert!(body.contains("held"), "{body}");
assert!(body.contains("every round"), "{body}");
}
/// The timer builds a failed merge again on every round, and `internal/meta`
/// moves between rounds: while the agent stays red the PR keeps its one
/// comment, whatever rev a retry builds.
#[tokio::test]
async fn a_build_that_stays_red_comments_once_across_retries_and_meta_revs() {
let (head, posted) = head_with_comments(REV).await;
for _ in 0..3 {
run_for_pr(&head, &nix_failed("error: boom"), 7).await;
}
let moved = FakeHead {
comments: head.comments.clone(),
..FakeHead::reading(Ok(Some(REV2)))
};
run_for_pr(&moved, &nix_failed("error: other"), 7).await;
let posted = posted.lock().unwrap();
assert_eq!(posted.len(), 1, "{posted:?}");
assert!(posted[0].1.contains(REV), "{:?}", posted[0]);
}
/// A green build ends the red spell, so a red after it comments again;
/// a red on another PR of the same agent is a different spell.
#[tokio::test]
async fn a_red_after_a_green_comments_again() {
let (head, posted) = head_with_comments(REV).await;
run_for_pr(&head, &nix_failed("error: boom"), 7).await;
let green = run_for_pr(&head, &FakeQueue::built(), 7).await;
assert_eq!(green, Outcome::Done);
let count = posted.lock().unwrap().len();
assert_eq!(count, 1, "a green build posts nothing");
run_for_pr(&head, &nix_failed("error: boom"), 7).await;
run_for_pr(&head, &nix_failed("error: boom"), 7).await;
let posted = posted.lock().unwrap();
assert_eq!(posted.len(), 2, "red, green, red: {posted:?}");
assert!(
!posted[0].1.contains(&failure_marker("atlas")),
"the first comment's marker is cleared: {:?}",
posted[0]
);
assert!(posted[0].1.contains(&cleared_marker("atlas")));
assert!(
posted[0].1.contains("Resolved: built at `0123456789ab`"),
"{:?}",
posted[0]
);
}
/// A build that got no verdict says nothing about the PR's config, so it
/// leaves no comment. The `nix` failure arm is the control.
#[tokio::test]
async fn a_no_verdict_failure_does_not_comment() {
for (name, reply) in [
("no responders", Err("no swarm builder".to_owned())),
(
"unavailable",
Ok(FromMetaReply::Unavailable {
reason: "busy".to_owned(),
}),
),
] {
let (head, posted) = head_with_comments(REV).await;
let outcome = run_for_pr(&head, &FakeQueue::answering(reply), 7).await;
assert!(
matches!(outcome, Outcome::Failed(e) if is_no_verdict(&e)),
"{name}"
);
assert!(posted.lock().unwrap().is_empty(), "{name}");
}
let (head, posted) = head_with_comments(REV).await;
run_for_pr(&head, &nix_failed("error: boom"), 7).await;
assert_eq!(
posted.lock().unwrap().len(),
1,
"control: a nix failure comments"
);
}
/// A build that is not a merged PR's (the pin fan-out, a created agent)
/// has nobody to comment on.
#[tokio::test]
async fn a_failed_build_that_is_not_from_a_pr_does_not_comment() {
let (head, posted) = head_with_comments(REV).await;
let outcome = run_with(&head, &nix_failed("error: boom")).await;
assert_eq!(outcome, Outcome::Failed("error: boom".to_owned()));
assert!(posted.lock().unwrap().is_empty());
}
#[tokio::test]
async fn the_comment_carries_the_end_of_a_long_tail_only() {
let tail = format!("START{}END", "x".repeat(10_000));
let (head, posted) = head_with_comments(REV).await;
run_for_pr(&head, &nix_failed(&tail), 7).await;
let posted = posted.lock().unwrap();
let body = &posted[0].1;
assert!(body.contains("END"), "the end of the tail is kept");
assert!(!body.contains("START"), "the start is cut");
assert!(body.len() < COMMENT_TAIL_BYTES + 1000, "{}", body.len());
}
/// A forge that does not take the comment leaves the node failed as it was.
#[tokio::test]
async fn a_comment_that_cannot_be_posted_does_not_change_the_outcome() {
let head = FakeHead {
comments: Some(std::sync::Arc::new(crate::forge::Client::for_test_forge(
url::Url::parse("http://127.0.0.1:1/").unwrap(),
))),
..FakeHead::reading(Ok(Some(REV)))
};
let outcome = run_for_pr(&head, &nix_failed("error: boom"), 7).await;
assert_eq!(outcome, Outcome::Failed("error: boom".to_owned()));
}
#[test]
fn the_repo_path_the_builder_fetches_is_the_one_the_controller_writes() {
assert_eq!(