From eceb742ea35a8a53ec0ef050485cb9cbde77d0bc Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 06:26:04 +0200 Subject: [PATCH] B2 final: an in-memory challenge wait ends with the write that outlives it The wait spacing a challenge whose record failed to land was removed only when a challenge record for that target landed. It now carries the stamp of the target's stored records (generation, meters, transactions) as they were when the record failed, and the supervision pass drops it as soon as the stamp differs: any write for the target landing, or the target retired or withdrawn, returns the schedule to the stored value. A wait can no longer hold a drift repair after the disk recovers. Soul's probes become tests named for the rules: one reload per wait for a re-drifting fragment, a proved wait of one observation age, a failed wait that follows the stored doubling, a lost race leaving no wait, the wait ending on a landed write, the wait ending with its target. The reloads() helper is gated to unix (its use of the route stubs broke the Windows --lib --tests check). Comments on the wait field and the failed-challenge record now say what the code does. Co-Authored-By: Claude Sonnet 5.5 --- src/control_plane.rs | 260 ++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 244 insertions(+), 16 deletions(-) diff --git a/src/control_plane.rs b/src/control_plane.rs index 2b1e225..278ca36 100644 --- a/src/control_plane.rs +++ b/src/control_plane.rs @@ -2941,6 +2941,32 @@ impl ControlSnapshot { Ok(()) } + /// Every stored record that concerns `target`, as read: the generation, the + /// meters and its transactions. Two reads with equal stamps saw no write + /// for the target between them. + fn target_stamp(&self, target: &str) -> Vec { + let mut stamp = self + .admitted + .iter() + .filter(|stored| stored.value.target == target) + .map(|stored| stored.envelope.clone()) + .chain( + self.targets + .iter() + .filter(|stored| stored.value.target == target) + .map(|stored| stored.envelope.clone()), + ) + .chain( + self.transactions + .iter() + .filter(|stored| stored.value.target == target) + .map(|stored| stored.envelope.clone()), + ) + .collect::>(); + stamp.sort_by(|a, b| (&a.r#type, &a.key).cmp(&(&b.r#type, &b.key))); + stamp + } + fn admitted_for(&self, target: &str) -> Option<&Stored> { self.admitted .iter() @@ -4307,6 +4333,13 @@ impl Drop for ProcessLock { } } +/// One target's in-memory challenge wait: when it is due, and the stored +/// records of the target as they were when the record failed to land. +struct ChallengeWait { + due_unix_millis: u64, + stamp: Vec, +} + struct Engine { options: RuntimeOptions, idunn_signer: ServiceIdentitySigner, @@ -4325,11 +4358,13 @@ struct Engine { /// Process-local on purpose: a restart retries at once, and no schema /// carries a clock that only pacing reads. resume_backoff: Mutex>, - /// When each routed target's next challenge is due, held only while the - /// stored `RouteSupervisionState` cannot be written. The stored value owns - /// the schedule and survives restarts; this spaces attempts when the record - /// of the last one did not land. - challenge_wait: Mutex>, + /// When each routed target's next challenge is due, held only when the + /// stored `RouteSupervisionState` could not record the last one. The stored + /// value owns the schedule and survives restarts; this spaces attempts + /// while the record did not land. An entry ends the moment any write for + /// its target lands (its stamp no longer matches the store) or the target + /// is retired or withdrawn: it never outlives the episode that made it. + challenge_wait: Mutex>, } /// The route driver's door to the actuation ceiling: every group of host @@ -5384,6 +5419,13 @@ impl Engine { fn supervise_one_admitted_generation(&self) -> Result { let snapshot = ControlSnapshot::read(&self.options.state_store)?; + // A wait belongs to the failed record it spaces. Once any write for its + // target has landed, or the target is no longer admitted, the stored + // schedule owns the target again. + self.challenge_wait + .lock() + .expect("challenge wait mutex") + .retain(|target, wait| snapshot.target_stamp(target) == wait.stamp); let mut progressed = false; let mut admitted = snapshot.admitted.iter().collect::>(); admitted.sort_by_key(|stored| stored.value.target.as_str()); @@ -5412,7 +5454,7 @@ impl Engine { }) { continue; } - match self.supervise_admitted_route(current) { + match self.supervise_admitted_route(&snapshot, current) { Ok(true) => { progressed = true; continue; @@ -5748,7 +5790,11 @@ impl Engine { Ok(progressed) } - fn supervise_admitted_route(&self, current: &Stored) -> Result { + fn supervise_admitted_route( + &self, + snapshot: &ControlSnapshot, + current: &Stored, + ) -> Result { let Some(expected_route) = current.value.expected.route.as_ref() else { ensure!( matches!(¤t.value.routing, RoutingEvidence::SkippedUnrouted), @@ -5779,10 +5825,10 @@ impl Engine { .lock() .expect("challenge wait mutex") .get(¤t.value.target) - .is_some_and(|due| { + .is_some_and(|wait| { is_waiting( now, - *due, + wait.due_unix_millis, ROUTE_CHALLENGE_BACKOFF_CAP_MILLIS.max(self.options.topology_maximum_age_millis), ) }) @@ -5828,10 +5874,11 @@ impl Engine { } let route_key = format!("route:{}", current.value.target); // A failed repair or a failed proof changes observation state only: it - // marks the route degraded and widens the wait. It is a write, so it - // ends this target's supervision pass (`Ok(true)`), exactly as a - // proved challenge does: the caller's snapshot no longer matches the - // generation, and a later step in the same pass would lose its CAS. + // marks the route degraded and widens the wait. That record is written + // like a proved one: when it lands the pass ends (`Ok(true)`), because + // the caller's snapshot no longer matches the generation; when the store + // refuses it the pass reports no progress and the in-memory wait below + // spaces the next attempt. let challenge = (|| -> Result { // Actuates only when the fragment on disk differs from the // admitted membership; an exact fragment makes this a no-op. It is @@ -5922,7 +5969,13 @@ impl Engine { self.challenge_wait .lock() .expect("challenge wait mutex") - .insert(current.value.target.clone(), wait_until); + .insert( + current.value.target.clone(), + ChallengeWait { + due_unix_millis: wait_until, + stamp: snapshot.target_stamp(¤t.value.target), + }, + ); self.report_once( &record_key, format!( @@ -17043,6 +17096,7 @@ mod tests { set_meters(&self.world, &meters) } + #[cfg(unix)] fn reloads(&self) -> usize { self.world.route_stubs.count("systemctl reload") } @@ -17106,7 +17160,7 @@ mod tests { #[cfg(unix)] fn challenge_due(world: &EngineFixture) -> Option { - world.engine.challenge_wait.lock().unwrap().get("service").copied() + world.engine.challenge_wait.lock().unwrap().get("service").map(|wait| wait.due_unix_millis) } #[cfg(unix)] @@ -17193,12 +17247,186 @@ mod tests { routed.admit()?; routed.world.route_stubs.refuse_reloads()?; routed.drift()?; - routed.world.engine.challenge_wait.lock().unwrap().insert("service".into(), 0); + let snapshot = ControlSnapshot::read(&routed.world.state_store)?; + routed.world.engine.challenge_wait.lock().unwrap().insert( + "service".into(), + ChallengeWait { due_unix_millis: 0, stamp: snapshot.target_stamp("service") }, + ); assert!(routed.tick()?, "a landed record is progress"); assert_eq!(challenge_due(&routed.world), None); Ok(()) } + /// A fragment that re-drifts after every restore, with the store + /// unwritable, is reloaded once per wait, however many laps run. + #[cfg(unix)] + #[test] + fn a_re_drifting_fragment_is_reloaded_once_per_wait_when_the_store_is_unwritable() -> Result<()> { + let name = "a_re_drifting_fragment_is_reloaded_once_per_wait_when_the_store_is_unwritable"; + let Some(_) = in_unwritable_store_child(name)? else { + let routed = routed_world(Odin::Unreachable, 1, DeploymentPhase::Fencing, false)?; + routed.admit()?; + let before = routed.reloads(); + break_store_writes(&routed.world); + for _ in 0..30 { + routed.drift()?; + assert!(!routed.tick()?, "a refused write was reported as progress"); + } + assert_eq!(routed.reloads() - before, 1, "a re-drifting fragment reloaded every lap"); + for attempt in 2..=4 { + routed.world.engine.challenge_wait.lock().unwrap().clear(); + routed.drift()?; + assert!(!routed.tick()?); + routed.drift()?; + assert!(!routed.tick()?); + assert_eq!(routed.reloads() - before, attempt); + } + return Ok(()); + }; + Ok(()) + } + + /// The wait of a proved challenge whose record failed is one + /// observation age from the attempt. + #[cfg(unix)] + #[test] + fn a_proved_challenge_that_cannot_be_recorded_waits_one_observation_age() -> Result<()> { + let name = "a_proved_challenge_that_cannot_be_recorded_waits_one_observation_age"; + let Some(_) = in_unwritable_store_child(name)? else { + let routed = routed_world(Odin::Unreachable, 1, DeploymentPhase::Fencing, false)?; + routed.admit()?; + let before = routed.reloads(); + break_store_writes(&routed.world); + let began = now_millis()?; + routed.drift()?; + assert!(!routed.tick()?); + assert_eq!(routed.reloads() - before, 1, "the repair did not run"); + let due = challenge_due(&routed.world).expect("no wait"); + let age = routed.world.engine.options.topology_maximum_age_millis; + assert!(due >= began + age, "the wait is shorter than one age: {}", due - began); + return Ok(()); + }; + Ok(()) + } + + /// The stored schedule sets the in-memory wait: with three failures + /// recorded, the fourth failed challenge waits eight ages (capped), and + /// no repair runs inside that wait. + #[cfg(unix)] + #[test] + fn a_failed_challenge_that_cannot_be_recorded_waits_as_long_as_the_stored_schedule() -> Result<()> { + let name = "a_failed_challenge_that_cannot_be_recorded_waits_as_long_as_the_stored_schedule"; + let Some(_) = in_unwritable_store_child(name)? else { + let routed = routed_world(Odin::Unreachable, 1, DeploymentPhase::Fencing, false)?; + routed.admit()?; + routed.stub.hang_up.store(true, Ordering::SeqCst); + edit_incumbent(&routed.world, |generation| { + let state = generation.route_supervision.as_mut().unwrap(); + state.consecutive_failures = 3; + state.next_challenge_at_unix_millis = Some(1); + state.degraded_since_unix_millis = Some(1); + })?; + break_store_writes(&routed.world); + let began = now_millis()?; + routed.drift()?; + assert!(!routed.tick()?); + let due = challenge_due(&routed.world).expect("no wait"); + let age = routed.world.engine.options.topology_maximum_age_millis; + let want = (age * 8).min(ROUTE_CHALLENGE_BACKOFF_CAP_MILLIS); + assert!(due >= began + want, "the wait ignores the stored doubling: {}", due - began); + let reloads = routed.reloads(); + for _ in 0..10 { + routed.drift()?; + assert!(!routed.tick()?); + } + assert_eq!(routed.reloads(), reloads, "the route was repaired inside its wait"); + return Ok(()); + }; + Ok(()) + } + + /// A lost race is progress, and the next pass is governed by the + /// winner's record, not by an in-memory wait of the loser. + #[cfg(unix)] + #[test] + fn a_lost_race_leaves_no_in_memory_wait() -> Result<()> { + let routed = routed_world(Odin::Unreachable, 1, DeploymentPhase::Fencing, false)?; + routed.admit()?; + routed.world.route_stubs.refuse_reloads()?; + let old = ControlSnapshot::read(&routed.world.state_store)?; + let stale = old.admitted_for("service").context("no generation")?; + edit_incumbent(&routed.world, |generation| { + generation.route_supervision.as_mut().unwrap().last_challenge_at_unix_millis = Some(5); + })?; + routed.drift()?; + assert!(routed.world.engine.supervise_admitted_route(&old, stale)?); + assert_eq!(challenge_due(&routed.world), None, "a lost race left an in-memory wait"); + Ok(()) + } + + /// The wait left by a full disk does not outlive it: once any write for + /// the target lands, the stored schedule owns the target again and a + /// drift is repaired without waiting. + #[cfg(unix)] + #[test] + fn a_wait_left_by_a_full_disk_ends_when_any_write_for_the_target_lands() -> Result<()> { + let name = "a_wait_left_by_a_full_disk_ends_when_any_write_for_the_target_lands"; + let Some(_) = in_unwritable_store_child(name)? else { + let routed = routed_world(Odin::Unreachable, 1, DeploymentPhase::Fencing, false)?; + routed.admit()?; + let before = routed.reloads(); + break_store_writes(&routed.world); + routed.drift()?; + assert!(!routed.tick()?); + assert!(challenge_due(&routed.world).is_some(), "no wait to outlive"); + assert_eq!(routed.reloads() - before, 1); + + // Space is freed. Nothing has landed yet, so the wait still holds. + unsafe { + let limit = libc::rlimit { rlim_cur: libc::RLIM_INFINITY, rlim_max: libc::RLIM_INFINITY }; + assert_eq!(libc::setrlimit(libc::RLIMIT_FSIZE, &limit), 0); + } + routed.drift()?; + assert!(!routed.tick()?); + assert_eq!(routed.reloads() - before, 1, "the wait did not hold before any write landed"); + + // Something else writes for the target: the wait is over. + routed.ledger(1)?; + routed.drift()?; + assert!(routed.tick()?, "the drift was not repaired"); + assert_eq!(routed.reloads() - before, 2, "a stale wait held the repair"); + assert_eq!(challenge_due(&routed.world), None); + return Ok(()); + }; + Ok(()) + } + + /// A wait belongs to an admitted target: once the target is retired, + /// the wait is gone. + #[cfg(unix)] + #[test] + fn a_wait_ends_when_its_target_is_retired() -> Result<()> { + let routed = routed_world(Odin::Unreachable, 1, DeploymentPhase::Fencing, false)?; + routed.admit()?; + let snapshot = ControlSnapshot::read(&routed.world.state_store)?; + routed.world.engine.challenge_wait.lock().unwrap().insert( + "service".into(), + ChallengeWait { + due_unix_millis: now_millis()? + 500_000, + stamp: snapshot.target_stamp("service"), + }, + ); + assert!(!routed.tick()?); + assert!(challenge_due(&routed.world).is_some(), "an unchanged store dropped the wait"); + assert!( + SingleFileMessagePackBackingStore::new(&routed.world.state_store) + .delete_batch_if_unchanged(&[snapshot.admitted_for("service").unwrap().envelope.clone()])? + ); + routed.tick()?; + assert_eq!(challenge_due(&routed.world), None, "a retired target kept its wait"); + Ok(()) + } + #[cfg(unix)] #[test] fn a_full_ledger_still_restores_the_admitted_and_incumbent_route() -> Result<()> {