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

cache-miss pass pushes a record the builder did not push

Browse source 
A meta build whose attic push failed (attic or its token not up yet) is
recorded with cached false and the deploy goes ahead. Nothing sent that
build again until internal/meta moved, so the create for that agent
failed on every hive and a rebuild was silently never pending.

The controller's cache-miss pass now also takes a record from the head
with cached false: it sends the same build without asking the cache, with
the pass's existing backoff (30 min doubling to 12 h), and rewrites the
record with cached true once the push lands, under the same
compare-and-set guard as a path change. The push.rs module doc no longer
claims the next build pushes again.

Mixed versions: controller-only, same wire type and builder request. An
old controller leaves an uncached record as before; a builder needs no
change.

Closes #6173
This commit is contained in:
atlas 2026-11-07 04:50:20 +01:00 • committed by mara
commit 64dfe0d205
1 changed files with 169 additions and 51 deletions
Download patch file Download diff file Expand all files Collapse all files

178
swarm-controller/src/cache_miss.rs
View file

@ -7,9 +7,15 @@
//! record and, on a 404, sends the builder the same `internal/meta` commit
//! again. The builder builds and pushes as for any meta build.
//!
//! Only a record a hive would use is checked: the one built from `internal/meta`'s
//! head, which says the builder pushed it. Only a 404 is a miss; an unreachable
//! cache or a refused token is "cannot tell" and rebuilds nothing.
//! Only a record built from `internal/meta`'s head is checked; an older one is
//! replaced by the next build. Only a 404 is a miss; an unreachable cache or a
//! refused token is "cannot tell" and rebuilds nothing.
//!
//! A record from the head with `cached` false (the builder's push failed, or the
//! cache or its token was not up yet) is not a hive's to use, and nothing else
//! pushes it before `internal/meta` moves. The pass sends its build without
//! asking the cache, and the record is rewritten with `cached` true once the
//! push lands.
//!
//! A pass sends at most [`MAX_REBUILDS_PER_PASS`] builds, one after the other.
//! An agent whose rebuild failed, or built without pushing, is left alone for
@ -17,10 +23,11 @@
//! lives in memory, so a restart tries every agent once more.
//!
//! The record names what hives deploy, so the pass changes it as little as it
//! can. A rebuild that lands on the record's own path writes nothing. One that
//! lands on another path (a build that is not reproducible) is logged at warn,
//! and replaces the record only if the record still holds the same commit and
//! no other write came in since it was read.
//! can. A rebuild of a cached record that lands on the record's own path writes
//! nothing. One that lands on another path (a build that is not reproducible)
//! is logged at warn. Both that and the push of an uncached record replace the
//! record only if it still holds the same commit and no other write came in
//! since it was read.
use std::collections::BTreeMap;
use std::future::Future;
@ -97,6 +104,10 @@ trait World {
enum Outcome {
/// Built and pushed, at the path the record names.
Rebuilt,
/// An uncached record's closure is pushed now, and the record says so.
Cached,
/// An uncached record's closure is pushed now, and the record was left as it was.
CachedKept,
/// Built and pushed at another path, and the record now names it.
PathChanged,
/// Built and pushed at another path, and the record was left as it was.
@ -113,6 +124,8 @@ impl Outcome {
const fn label(self) -> &'static str {
match self {
Self::Rebuilt => "rebuilt",
Self::Cached => "cached",
Self::CachedKept => "cached_kept",
Self::PathChanged => "path_changed",
Self::PathChangedKept => "path_changed_kept",
Self::NotPushed => "not_pushed",
@ -240,18 +253,20 @@ async fn pass(
}
};
let record = &stored.closure;
if !record.cached || record.meta_rev != head {
if record.meta_rev != head {
continue;
}
match world.holds(&record.store_path).await {
Ok(true) => {
backoff.cleared(&agent);
continue;
}
Ok(false) => {}
Err(e) => {
tracing::warn!(%agent, error = %format!("{e:#}"), "cache miss: asking the cache failed");
continue;
if record.cached {
match world.holds(&record.store_path).await {
Ok(true) => {
backoff.cleared(&agent);
continue;
}
Ok(false) => {}
Err(e) => {
tracing::warn!(%agent, error = %format!("{e:#}"), "cache miss: asking the cache failed");
continue;
}
}
}
if sent.len() >= MAX_REBUILDS_PER_PASS {
@ -261,10 +276,17 @@ async fn pass(
);
break;
}
tracing::warn!(
%agent, meta_rev = %record.meta_rev, store_path = %record.store_path,
"cache miss: the cache lost the agent's closure; rebuilding"
);
if record.cached {
tracing::warn!(
%agent, meta_rev = %record.meta_rev, store_path = %record.store_path,
"cache miss: the cache lost the agent's closure; rebuilding"
);
} else {
tracing::warn!(
%agent, meta_rev = %record.meta_rev, store_path = %record.store_path,
"cache miss: the record says the builder did not push the closure; rebuilding to push it"
);
}
let outcome = rebuild(world, &agent, stored).await;
metrics.count(outcome);
if outcome.backs_off() {
@ -309,14 +331,17 @@ async fn rebuild(world: &impl World, agent: &str, stored: Stored) -> Outcome {
tracing::warn!(%agent, %meta_rev, %out_path, "cache miss: rebuilt, but the builder did not push it");
return Outcome::NotPushed;
}
if out_path == stored.closure.store_path {
let same_path = out_path == stored.closure.store_path;
if same_path && stored.closure.cached {
tracing::info!(%agent, %meta_rev, %out_path, "cache miss: rebuilt and pushed");
return Outcome::Rebuilt;
}
tracing::warn!(
%agent, %meta_rev, recorded = %stored.closure.store_path, %out_path,
"cache miss: the rebuild is at another path than the record; the build is not reproducible"
);
if !same_path {
tracing::warn!(
%agent, %meta_rev, recorded = %stored.closure.store_path, %out_path,
"cache miss: the rebuild is at another path than the record; the build is not reproducible"
);
}
let now = AgentClosure {
meta_rev: meta_rev.clone(),
store_path: out_path,
@ -326,15 +351,20 @@ async fn rebuild(world: &impl World, agent: &str, stored: Stored) -> Outcome {
private_users,
nix_builds,
};
let (written, kept) = if same_path {
(Outcome::Cached, Outcome::CachedKept)
} else {
(Outcome::PathChanged, Outcome::PathChangedKept)
};
match replace_if_current(world, agent, &now).await {
Ok(true) => Outcome::PathChanged,
Ok(true) => written,
Ok(false) => {
tracing::warn!(%agent, %meta_rev, "cache miss: the record changed meanwhile and is left as it is");
Outcome::PathChangedKept
kept
}
Err(e) => {
tracing::warn!(%agent, %meta_rev, error = %format!("{e:#}"), "cache miss: the record could not be replaced");
Outcome::PathChangedKept
kept
}
}
}
@ -763,13 +793,93 @@ mod tests {
}
#[tokio::test]
async fn only_the_record_a_hive_would_use_is_checked() {
let world = Fake::missing(&["iris", "argus", "atlas", "muede"])
async fn only_the_record_from_the_head_is_checked() {
let world = Fake::missing(&["iris", "atlas", "muede"])
.with_record("iris", record(OLD, "iris", true))
.with_record("argus", record(HEAD, "argus", false));
.with_record("muede", record(OLD, "muede", false));
let sent = pass_at(&world, &mut Backoff::default(), Instant::now()).await;
// `iris` is from another commit, `argus` was never pushed.
assert_eq!(sent, [rebuilt("atlas"), rebuilt("muede")]);
// `iris` and `muede` are from another commit, whether or not pushed.
assert_eq!(sent, [rebuilt("atlas")]);
}
fn cached(agent: &str) -> (String, Outcome) {
(agent.to_owned(), Outcome::Cached)
}
#[tokio::test]
async fn a_record_the_builder_did_not_push_is_pushed_and_rewritten_as_cached() {
let mut world =
Fake::missing(&["iris", "argus"]).with_record("argus", record(HEAD, "argus", false));
// The cache is not asked about a record that says it was never pushed.
world.cache_down = true;
let mut backoff = Backoff::default();
let now = Instant::now();
assert_eq!(pass_at(&world, &mut backoff, now).await, [cached("argus")]);
assert_eq!(world.builds(), [("argus".to_owned(), HEAD.to_owned())]);
let writes = world.writes();
assert_eq!(writes.len(), 1);
assert_eq!(writes[0].0, "argus");
assert_eq!(writes[0].1.store_path, path("argus"));
assert!(writes[0].1.cached);
// Done: the record is cached, so the next pass sends nothing for it.
world.cache_down = false;
assert_eq!(pass_at(&world, &mut backoff, now).await, [rebuilt("iris")]);
assert_eq!(world.builds().len(), 2);
assert_eq!(pass_at(&world, &mut backoff, now).await, []);
}
#[tokio::test]
async fn a_push_that_fails_again_is_tried_after_the_wait_until_it_lands() {
let world = Fake::missing(&["argus"]).with_record("argus", record(HEAD, "argus", false));
world
.answers
.lock()
.unwrap()
.insert("argus".to_owned(), Ok(built(&path("argus"), false)));
let mut backoff = Backoff::default();
let t0 = Instant::now();
let not_pushed = [("argus".to_owned(), Outcome::NotPushed)];
assert_eq!(pass_at(&world, &mut backoff, t0).await, not_pushed);
assert!(world.writes().is_empty());
// No tight loop inside the wait.
let inside = t0 + BACKOFF_BASE.saturating_sub(Duration::from_secs(1));
assert_eq!(pass_at(&world, &mut backoff, inside).await, []);
assert_eq!(world.builds().len(), 1);
// After it the build is sent again, and a push that lands fixes the record.
let t1 = t0 + BACKOFF_BASE;
assert_eq!(pass_at(&world, &mut backoff, t1).await, not_pushed);
world.answers.lock().unwrap().clear();
let t2 = t1 + BACKOFF_BASE * 2;
assert_eq!(pass_at(&world, &mut backoff, t2).await, [cached("argus")]);
assert_eq!(world.builds().len(), 3);
assert!(world.records.lock().unwrap()["argus"].closure.cached);
}
#[tokio::test]
async fn an_uncached_record_rebuilt_at_another_path_names_the_new_one() {
let other = path("argus-other");
let world = Fake::missing(&["argus"])
.with_record("argus", record(HEAD, "argus", false))
.answering("argus", built(&other, true));
let sent = pass_at(&world, &mut Backoff::default(), Instant::now()).await;
assert_eq!(sent, [("argus".to_owned(), Outcome::PathChanged)]);
let writes = world.writes();
assert_eq!(writes.len(), 1);
assert_eq!(
(writes[0].1.store_path.as_str(), writes[0].1.cached),
(other.as_str(), true)
);
}
#[tokio::test]
async fn an_uncached_record_that_moved_on_is_left_alone() {
let world = Fake::missing(&["argus"]).with_record("argus", record(HEAD, "argus", false));
// A newer commit is recorded while the push runs.
*world.racing.lock().unwrap() =
Some(("argus".to_owned(), record(OLD, "argus-newer", true)));
let sent = pass_at(&world, &mut Backoff::default(), Instant::now()).await;
assert_eq!(sent, [("argus".to_owned(), Outcome::CachedKept)]);
assert!(world.writes().is_empty());
}
#[tokio::test]