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

report per agent whether a hive can pull its latest closure from the cache

Browse source 
for each agent the hive looks up its record in agent-closures, fetches the closure's narinfo from the cache with the pull token and checks that a signature by the cache's public key covers it. the result is a log line and the gauge hyperhive.agent.closure_pullable, with a reason label when it is 0. nothing is downloaded and deploys still build locally. part 3 of 3: the probe needs the cache facts and the closure read grant from parts 1 and 2.
This commit is contained in:
atlas 2026-10-27 17:33:57 +01:00 • committed by mara
commit 5486fc0333
1 changed files with 812 additions and 34 deletions
Download patch file Download diff file Expand all files Collapse all files

396
hive-c0re/src/attic/probe.rs Normal file
View file

@ -0,0 +1,396 @@
//! Whether this hive could pull the latest closure of each of its agents from
//! the swarm's binary cache, and the answer as a log line and a gauge.
//!
//! Every [`PASS_INTERVAL`] each agent's record in `agent-closures` is looked up
//! and its narinfo fetched with the pull token and checked against the cache's
//! public key. Nothing is downloaded and nothing is deployed from it.
//!
//! The verdict is the first of these that fails, in this order:
//! no credentials (`no-credentials`), no record (`no-record`), a record built
//! from another `internal/meta` commit than the head (`rev-mismatch`), a record
//! the builder did not push (`not-cached`), the cache not holding the narinfo
//! (`not-cached`), a request that failed or was refused (`fetch-failed`), a
//! narinfo no signature of the key covers (`bad-signature`).
use std::sync::Mutex;
use std::time::Duration;
use anyhow::{Context as _, Result};
use swarm_queue_client::agent_closures::{self, Read};
use swarm_wire::agent_closures::AgentClosure;
use super::narinfo;
use super::reader::{self, Facts};
use crate::stats::sweep_health::{self, SweepHealth};
/// How often every agent is probed.
pub const PASS_INTERVAL: Duration = Duration::from_mins(5);
/// At [`PASS_INTERVAL`] this is ~15 minutes of failing passes.
const FAILURES_BEFORE_WARNING: u32 = 3;
/// Why a closure cannot be pulled; the gauge's `reason` label.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Reason {
NoRecord,
RevMismatch,
NotCached,
NoCredentials,
FetchFailed,
BadSignature,
}
impl Reason {
/// The label value.
pub const fn label(self) -> &'static str {
match self {
Reason::NoRecord => "no-record",
Reason::RevMismatch => "rev-mismatch",
Reason::NotCached => "not-cached",
Reason::NoCredentials => "no-credentials",
Reason::FetchFailed => "fetch-failed",
Reason::BadSignature => "bad-signature",
}
}
}
/// A closure that cannot be pulled, and what the log line says about it.
#[derive(Debug)]
pub struct Unpullable {
pub reason: Reason,
pub detail: String,
}
fn no(reason: Reason, detail: impl Into<String>) -> Unpullable {
Unpullable {
reason,
detail: detail.into(),
}
}
/// The cache's narinfo for a store path, so a test can stand in for it.
pub(crate) trait Cache {
/// `None` when the cache does not hold the path.
async fn narinfo(&self, facts: &Facts, store_path: &str) -> Result<Option<String>>;
}
struct Http(reqwest::Client);
impl Cache for Http {
async fn narinfo(&self, facts: &Facts, store_path: &str) -> Result<Option<String>> {
narinfo::fetch(&self.0, facts, store_path).await
}
}
/// Whether the closure `record` names can be pulled with `facts`. `head` is the
/// `internal/meta` commit a deploy would build from.
pub(crate) async fn judge(
facts: Option<&Facts>,
record: Option<&AgentClosure>,
head: Option<&str>,
cache: &impl Cache,
) -> Result<(), Unpullable> {
let facts = facts.ok_or_else(|| {
no(
Reason::NoCredentials,
"the swarm has given this hive no cache facts",
)
})?;
let record =
record.ok_or_else(|| no(Reason::NoRecord, "no build is recorded for the agent"))?;
if head != Some(record.meta_rev.as_str()) {
return Err(no(
Reason::RevMismatch,
format!(
"built from {}, internal/meta is at {}",
record.meta_rev,
head.unwrap_or("nothing")
),
));
}
if !record.cached {
return Err(no(
Reason::NotCached,
"the builder did not push the closure",
));
}
let text = match cache.narinfo(facts, &record.store_path).await {
Ok(Some(text)) => text,
Ok(None) => return Err(no(Reason::NotCached, "the cache does not hold the narinfo")),
Err(e) => return Err(no(Reason::FetchFailed, format!("{e:#}"))),
};
narinfo::verify(&text, &record.store_path, facts.public_key())
.map_err(|e| no(Reason::BadSignature, format!("{e:#}")))
}
/// The last pass's verdict per agent: `None` when it can be pulled.
static VERDICTS: Mutex<Vec<(String, Option<Reason>)>> = Mutex::new(Vec::new());
/// The verdicts of the last pass, for the gauge.
pub fn verdicts() -> Vec<(String, Option<Reason>)> {
VERDICTS.lock().map(|g| g.clone()).unwrap_or_default()
}
/// Probe every agent now and every [`PASS_INTERVAL`] until shutdown.
pub async fn follow(client: async_nats::Client, mut shutdown: tokio::sync::watch::Receiver<bool>) {
let mut health = SweepHealth::new("attic_pull_probe", "warn", FAILURES_BEFORE_WARNING);
loop {
match pass(&client).await {
Ok(()) => health.record_ok(),
Err(e) => {
tracing::warn!(error = %format!("{e:#}"), "attic: the pull probe failed; the last verdicts stand");
let err = format!("{e:#}");
health.record_err(|ctx| {
let age = ctx.since_last_ok.map_or_else(
|| "no success this session".to_owned(),
|d| format!("last ok {} ago", sweep_health::fmt_age(d)),
);
format!(
"checking whether agents' closures can be pulled is failing ({} consecutive, {age}) \
— deploys are unaffected: {err}",
ctx.consecutive
)
});
}
}
tokio::select! {
() = tokio::time::sleep(PASS_INTERVAL) => {}
_ = shutdown.changed() => {
tracing::info!("attic: shutdown signal received");
return;
}
}
}
}
/// One pass: publish the facts, then judge every agent.
///
/// # Errors
/// A read that failed (the store, the queue, the forge), which leaves the last
/// verdicts as they are: the failure says nothing about any agent.
async fn pass(client: &async_nats::Client) -> Result<()> {
swarm_queue_client::ensure_connected(client)?;
let facts = reader::sync().await?;
let agents = crate::lifecycle::agents_for_meta_listing()
.await
.context("enumerating this hive's agents")?;
let head = match &facts {
Some(_) => meta_head().await?,
None => None,
};
let http = Http(narinfo::client()?);
let mut now = Vec::with_capacity(agents.len());
for spec in agents {
let record = match &facts {
Some(_) => read_record(client, &spec.name).await?,
None => None,
};
let verdict = judge(facts.as_ref(), record.as_ref(), head.as_deref(), &http).await;
now.push((spec.name, verdict.as_ref().err().map(|u| u.reason)));
report(&now[now.len() - 1], verdict.err().map(|u| u.detail));
}
if let Ok(mut g) = VERDICTS.lock() {
*g = now;
}
Ok(())
}
async fn read_record(client: &async_nats::Client, agent: &str) -> Result<Option<AgentClosure>> {
match agent_closures::read(client, agent)
.await
.map_err(|e| anyhow::anyhow!(e))
.with_context(|| format!("reading the closure record of {agent}"))?
{
Read::Current(record) => Ok(record),
Read::NoBucket => Ok(None),
}
}
/// The `internal/meta` head, or `None` when the swarm has not published it or
/// this hive has no forge token.
async fn meta_head() -> Result<Option<String>> {
use crate::meta_rev::MetaSource as _;
match crate::forge::internal_meta_source().await {
Some(source) => source
.meta_head()
.await
.context("reading internal/meta's head"),
None => Ok(None),
}
}
/// The log line: at info when the verdict differs from the last pass, else
/// debug.
fn report((agent, verdict): &(String, Option<Reason>), detail: Option<String>) {
let changed = verdicts()
.iter()
.find(|(a, _)| a == agent)
.is_none_or(|(_, v)| v != verdict);
let reason = verdict.map_or("", Reason::label);
let pullable = verdict.is_none();
let detail = detail.unwrap_or_default();
if changed {
tracing::info!(%agent, pullable, reason, %detail, "attic: can this hive pull the agent's latest closure");
} else {
tracing::debug!(%agent, pullable, reason, %detail, "attic: can this hive pull the agent's latest closure");
}
}
#[cfg(test)]
mod tests {
use base64::Engine as _;
use base64::engine::general_purpose::STANDARD;
use ed25519_dalek::{Signer as _, SigningKey};
use super::*;
const PATH: &str = "/nix/store/0123456789abcdfghijklmnpqrsvwxyz-nixos-system-atlas";
const HASH: &str = "sha256:0mdqa9w1p6cmli6976v4wi0sw9r4p5prkj7lzfd1877wk11c9c73";
const REV: &str = "a1b2c3";
fn key() -> SigningKey {
SigningKey::from_bytes(&[7; 32])
}
fn facts() -> Facts {
let public = format!(
"swarm:{}",
STANDARD.encode(key().verifying_key().to_bytes())
);
Facts::parse("tok", "https://cache.example", "swarm", &public).expect("facts")
}
fn record() -> AgentClosure {
AgentClosure {
meta_rev: REV.to_owned(),
store_path: PATH.to_owned(),
system: "x86_64-linux".to_owned(),
built_unix: 1,
cached: true,
}
}
fn narinfo_signed_by(key: &SigningKey) -> String {
let sig = key.sign(format!("1;{PATH};{HASH};120;").as_bytes());
format!(
"StorePath: {PATH}\nNarHash: {HASH}\nNarSize: 120\nReferences: \nSig: swarm:{}\n",
STANDARD.encode(sig.to_bytes())
)
}
/// What the cache answers for the narinfo.
struct Answers(Result<Option<String>, &'static str>);
impl Cache for Answers {
async fn narinfo(&self, _: &Facts, _: &str) -> Result<Option<String>> {
self.0.clone().map_err(|e| anyhow::anyhow!(e))
}
}
async fn reason(
facts: Option<&Facts>,
record: Option<&AgentClosure>,
head: Option<&str>,
cache: Answers,
) -> Option<Reason> {
judge(facts, record, head, &cache)
.await
.err()
.map(|u| u.reason)
}
fn good() -> Answers {
Answers(Ok(Some(narinfo_signed_by(&key()))))
}
#[tokio::test]
async fn a_signed_narinfo_for_the_head_build_is_pullable() {
assert_eq!(
reason(Some(&facts()), Some(&record()), Some(REV), good()).await,
None
);
}
#[tokio::test]
async fn no_facts_is_no_credentials_even_with_a_record() {
let r = reason(None, Some(&record()), Some(REV), good()).await;
assert_eq!(r, Some(Reason::NoCredentials));
}
#[tokio::test]
async fn no_record_is_no_record() {
let r = reason(Some(&facts()), None, Some(REV), good()).await;
assert_eq!(r, Some(Reason::NoRecord));
}
#[tokio::test]
async fn a_record_from_another_commit_is_a_rev_mismatch() {
let r = reason(Some(&facts()), Some(&record()), Some("other"), good()).await;
assert_eq!(r, Some(Reason::RevMismatch));
let r = reason(Some(&facts()), Some(&record()), None, good()).await;
assert_eq!(r, Some(Reason::RevMismatch), "no head to compare with");
}
#[tokio::test]
async fn an_unpushed_record_or_a_missing_narinfo_is_not_cached() {
let unpushed = AgentClosure {
cached: false,
..record()
};
let r = reason(Some(&facts()), Some(&unpushed), Some(REV), good()).await;
assert_eq!(r, Some(Reason::NotCached));
let r = reason(
Some(&facts()),
Some(&record()),
Some(REV),
Answers(Ok(None)),
)
.await;
assert_eq!(r, Some(Reason::NotCached));
}
#[tokio::test]
async fn a_failed_request_is_fetch_failed() {
let r = reason(
Some(&facts()),
Some(&record()),
Some(REV),
Answers(Err("401")),
)
.await;
assert_eq!(r, Some(Reason::FetchFailed));
}
#[tokio::test]
async fn a_narinfo_the_key_did_not_sign_is_a_bad_signature() {
let forged = Answers(Ok(Some(narinfo_signed_by(&SigningKey::from_bytes(
&[9; 32],
)))));
let r = reason(Some(&facts()), Some(&record()), Some(REV), forged).await;
assert_eq!(r, Some(Reason::BadSignature));
}
#[test]
fn the_labels_are_the_documented_ones() {
let labels = [
Reason::NoRecord,
Reason::RevMismatch,
Reason::NotCached,
Reason::NoCredentials,
Reason::FetchFailed,
Reason::BadSignature,
]
.map(Reason::label);
assert_eq!(
labels,
[
"no-record",
"rev-mismatch",
"not-cached",
"no-credentials",
"fetch-failed",
"bad-signature"
]
);
}
}