From 4a25ea9637e940e533af6be2268f6a4279cf6c8d Mon Sep 17 00:00:00 2001 From: "@permadeath.com" Date: Tue, 1 Sep 2026 23:35:36 -0400 Subject: [PATCH] test(reconcile): prove the three things it must never do Foreign records survive every state, a failed read writes nothing, and a name it cannot fix is quarantined rather than retried. --- crates/didbot-reconcile/src/lib.rs | 578 +++++++++++++++++++++++++++++ 1 file changed, 578 insertions(+) diff --git a/crates/didbot-reconcile/src/lib.rs b/crates/didbot-reconcile/src/lib.rs index 2e3729f7..01f6337e 100644 --- a/crates/didbot-reconcile/src/lib.rs +++ b/crates/didbot-reconcile/src/lib.rs @@ -1099,3 +1099,581 @@ impl ReconcileTask { let _ = self.handle.await; } } + +#[cfg(test)] +mod tests { + use super::*; + use didbot_dns::{DnsError, InMemoryDns}; + use didbot_fsm::GateStatus; + use std::net::Ipv4Addr; + use std::sync::atomic::{AtomicUsize, Ordering}; + + const ZONE: &str = "agents.example"; + + fn ip(last: u8) -> RecordTarget { + RecordTarget::Ipv4(Ipv4Addr::new(203, 0, 113, last)) + } + + /// An [`InMemoryDns`] that counts writes and can be made to refuse them. + /// + /// The withdrawal log is the evidence for the property that matters: no + /// test in this module ever sees a name in it that the intent did not + /// name. + #[derive(Default)] + struct CountingDns { + inner: InMemoryDns, + publishes: AtomicUsize, + withdrawn: Mutex>, + refuse_publish: Mutex>, + } + + impl CountingDns { + fn withdrawn(&self) -> Vec { + self.withdrawn.lock().unwrap().clone() + } + + fn refuse_publish(&self, message: &str) { + *self.refuse_publish.lock().unwrap() = Some(message.to_string()); + } + } + + impl DnsProvider for CountingDns { + fn name(&self) -> &str { + "counting" + } + + fn publish(&self, host: &str, target: &RecordTarget) -> Result<(), DnsError> { + self.publishes.fetch_add(1, Ordering::SeqCst); + if let Some(message) = self.refuse_publish.lock().unwrap().clone() { + return Err(DnsError::Backend { + host: host.to_owned(), + message, + }); + } + self.inner.publish(host, target) + } + + fn withdraw(&self, host: &str) -> Result<(), DnsError> { + self.withdrawn.lock().unwrap().push(host.to_owned()); + self.inner.withdraw(host) + } + + fn published(&self) -> Vec { + self.inner.published() + } + + fn target(&self, host: &str) -> Option { + self.inner.target(host) + } + + fn publish_txt(&self, host: &str, value: &str) -> Result<(), DnsError> { + self.inner.publish_txt(host, value) + } + + fn withdraw_txt(&self, host: &str, value: &str) -> Result<(), DnsError> { + self.inner.withdraw_txt(host, value) + } + + fn txt_values(&self, host: &str) -> Vec { + self.inner.txt_values(host) + } + } + + /// A view of what the provider holds, which can be made to fail, or + /// frozen at a fixed answer to model a zone somebody keeps changing back. + struct TestView { + provider: Arc, + fail: Mutex>, + frozen: Mutex>>, + reads: AtomicUsize, + } + + impl ZoneView for TestView { + fn observe(&self) -> Result, ViewError> { + self.reads.fetch_add(1, Ordering::SeqCst); + if let Some(err) = self.fail.lock().unwrap().clone() { + return Err(err); + } + if let Some(frozen) = self.frozen.lock().unwrap().clone() { + return Ok(frozen); + } + Ok(self + .provider + .published() + .into_iter() + .filter(|host| under_zone(host, ZONE)) + .filter_map(|host| self.provider.target(&host).map(|t| (host, t))) + .collect()) + } + } + + struct TestIntent(Mutex, String>>); + + impl ZoneIntent for TestIntent { + fn intended(&self) -> Result, String> { + self.0.lock().unwrap().clone() + } + } + + struct Fixture { + reconciler: Arc, + provider: Arc, + view: Arc, + intent: Arc, + } + + impl Fixture { + fn new(authority: ZoneAuthority, intended: &[(&str, RecordTarget)]) -> Self { + struct ViewHandle(Arc); + impl ZoneView for ViewHandle { + fn observe(&self) -> Result, ViewError> { + self.0.observe() + } + } + struct IntentHandle(Arc); + impl ZoneIntent for IntentHandle { + fn intended(&self) -> Result, String> { + self.0.intended() + } + } + + let provider = Arc::new(CountingDns::default()); + let view = Arc::new(TestView { + provider: Arc::clone(&provider), + fail: Mutex::new(None), + frozen: Mutex::new(None), + reads: AtomicUsize::new(0), + }); + let intent = Arc::new(TestIntent(Mutex::new(Ok(intended + .iter() + .map(|(host, target)| ((*host).to_string(), target.clone())) + .collect())))); + let reconciler = Arc::new(ZoneReconciler::new( + ZONE, + authority, + Box::new(ViewHandle(Arc::clone(&view))), + Box::new(IntentHandle(Arc::clone(&intent))), + Arc::clone(&provider) as Arc, + )); + Self { + reconciler, + provider, + view, + intent, + } + } + + fn managing(intended: &[(&str, RecordTarget)]) -> Self { + Self::new(ZoneAuthority::Manages, intended) + } + + fn with(mut self, f: impl FnOnce(ZoneReconciler) -> ZoneReconciler) -> Self { + let reconciler = Arc::try_unwrap(self.reconciler) + .map_err(|_| "sole owner") + .unwrap(); + self.reconciler = Arc::new(f(reconciler)); + self + } + + fn tick(&self) -> TickReport { + self.reconciler.tick() + } + } + + #[test] + fn a_first_observation_of_drift_is_propagation_not_a_fault() { + let f = Fixture::managing(&[("a.agents.example", ip(1))]); + let report = f.tick(); + assert_eq!(report.state, ZoneState::Drifted); + assert_eq!(report.survey.missing.len(), 1); + assert!(!report.wrote(), "nothing is written on first sight"); + assert_eq!(f.provider.publishes.load(Ordering::SeqCst), 0); + // Wrong, but not *failed*: nothing has been attempted. This is the + // third value, and a two-valued gate would back off from it. + assert_eq!( + report.gates.status(gate::PRESENT), + Some(&GateStatus::Pending) + ); + assert!(!report.gates.any_failed()); + assert_eq!(report.next_delay, Schedule::default().interval); + } + + #[test] + fn a_missing_record_is_republished_once_a_second_read_confirms_it() { + let f = Fixture::managing(&[("a.agents.example", ip(1))]); + f.tick(); + let report = f.tick(); + assert_eq!(report.state, ZoneState::Reconciling); + assert_eq!( + report.applied, + vec![Repair::Publish { + host: "a.agents.example".into(), + want: ip(1) + }] + ); + assert_eq!(f.provider.target("a.agents.example"), Some(ip(1))); + + let settled = f.tick(); + assert_eq!(settled.state, ZoneState::InSync); + assert!(!settled.wrote(), "a repaired zone is not repaired again"); + assert!(settled.gates.all_satisfied()); + } + + #[test] + fn a_record_pointing_elsewhere_is_replaced() { + let f = Fixture::managing(&[("a.agents.example", ip(1))]); + f.provider.publish("a.agents.example", &ip(9)).unwrap(); + f.tick(); + let report = f.tick(); + assert_eq!( + report.applied, + vec![Repair::Replace { + host: "a.agents.example".into(), + have: ip(9), + want: ip(1) + }] + ); + assert_eq!(f.provider.withdrawn(), vec!["a.agents.example"]); + assert_eq!(f.provider.target("a.agents.example"), Some(ip(1))); + } + + #[test] + fn a_record_this_deployment_did_not_publish_is_never_touched() { + // One name the intent knows and one it does not, side by side under + // the same zone, driven through every state this machine has. + let f = Fixture::managing(&[("mine.agents.example", ip(1))]); + f.provider.publish("theirs.agents.example", &ip(9)).unwrap(); + + let mut seen = std::collections::BTreeSet::new(); + // drifted, reconciling, in-sync + for _ in 0..4 { + seen.insert(f.tick().state); + } + // blocked + *f.view.fail.lock().unwrap() = Some(ViewError::Unreadable("down".into())); + seen.insert(f.tick().state); + *f.view.fail.lock().unwrap() = None; + for _ in 0..2 { + seen.insert(f.tick().state); + } + + assert_eq!( + seen, + ZoneState::ALL.iter().copied().collect(), + "every state was exercised while the foreign record sat there" + ); + assert_eq!( + f.provider.target("theirs.agents.example"), + Some(ip(9)), + "a record this deployment cannot attribute survives every state" + ); + assert!( + !f.provider.withdrawn().contains(&"theirs.agents.example".to_string()), + "and is never even withdrawn as part of a replace" + ); + } + + #[test] + fn an_unattributed_record_yields_no_repair_at_all() { + // The structural half of the property above: there is nothing to + // execute, not merely nothing that executes it. + let drift = Drift::Unattributed { + host: "theirs.agents.example".into(), + have: ip(9), + }; + assert_eq!(drift.repair(), None); + } + + #[test] + fn a_zone_that_cannot_be_read_blocks_and_writes_nothing() { + let f = Fixture::managing(&[("a.agents.example", ip(1))]); + f.provider.publish("stale.agents.example", &ip(9)).unwrap(); + *f.view.fail.lock().unwrap() = Some(ViewError::Unreadable("connection refused".into())); + + let report = f.tick(); + assert_eq!(report.state, ZoneState::Blocked); + assert!(!f.reconciler.policy().may_repair); + assert!(!f.reconciler.policy().observation_is_evidence); + assert_eq!( + report.gates.blocking().map(|g| g.name()), + Some(gate::READABLE) + ); + assert!(report.survey.unattributed.is_empty(), "it surveyed nothing"); + assert_eq!(f.provider.publishes.load(Ordering::SeqCst), 1, "only the test's own setup call"); + assert!(f.provider.withdrawn().is_empty()); + assert_eq!(f.provider.target("stale.agents.example"), Some(ip(9))); + } + + #[test] + fn a_missing_zone_blocks_rather_than_creating_one() { + let f = Fixture::managing(&[("a.agents.example", ip(1))]); + *f.view.fail.lock().unwrap() = Some(ViewError::ZoneAbsent); + let report = f.tick(); + assert_eq!(report.state, ZoneState::Blocked); + assert!(!report.wrote()); + } + + #[test] + fn an_underivable_intent_blocks_rather_than_treating_the_zone_as_foreign() { + let f = Fixture::managing(&[("a.agents.example", ip(1))]); + f.provider.publish("a.agents.example", &ip(1)).unwrap(); + *f.intent.0.lock().unwrap() = Err("account store unavailable".into()); + let report = f.tick(); + assert_eq!(report.state, ZoneState::Blocked); + assert_eq!( + report.gates.blocking().map(|g| g.name()), + Some(gate::INTENT) + ); + assert!( + report.survey.unattributed.is_empty(), + "an empty intent would have called every live record unattributed" + ); + assert!(f.provider.withdrawn().is_empty()); + } + + #[test] + fn a_zone_this_deployment_does_not_manage_is_not_even_read() { + let f = Fixture::new(ZoneAuthority::ReadOnly, &[("a.agents.example", ip(1))]); + let report = f.tick(); + assert_eq!(report.state, ZoneState::Blocked); + assert_eq!( + report.gates.blocking().map(|g| g.name()), + Some(gate::MANAGED) + ); + assert_eq!(f.view.reads.load(Ordering::SeqCst), 0); + assert_eq!(f.provider.publishes.load(Ordering::SeqCst), 0); + } + + #[test] + fn a_failed_read_is_not_evidence_in_either_direction() { + let f = Fixture::managing(&[("a.agents.example", ip(1))]); + assert_eq!(f.tick().state, ZoneState::Drifted); // one confirmation + *f.view.fail.lock().unwrap() = Some(ViewError::Unreadable("down".into())); + assert_eq!(f.tick().state, ZoneState::Blocked); + *f.view.fail.lock().unwrap() = None; + // The blocked tick neither confirmed the drift nor cleared it, so + // the next successful read is the second confirmation. + let report = f.tick(); + assert_eq!(report.state, ZoneState::Reconciling); + assert_eq!(report.applied.len(), 1); + } + + #[test] + fn a_write_that_keeps_failing_is_bounded_rather_than_retried_forever() { + let f = Fixture::managing(&[("a.agents.example", ip(1))]).with(|r| r.with_max_attempts(3)); + f.provider.refuse_publish("throttled"); + let mut states = Vec::new(); + for _ in 0..10 { + states.push(f.tick().state); + } + assert_eq!( + f.provider.publishes.load(Ordering::SeqCst), + 3, + "three attempts, then the name is left alone — not ten" + ); + assert_eq!(states.last(), Some(&ZoneState::Blocked)); + assert!(f + .reconciler + .tick() + .quarantined + .contains(&"a.agents.example".to_string())); + } + + #[test] + fn a_name_somebody_keeps_changing_back_is_reported_rather_than_fought() { + // The view is frozen at what a human keeps putting back, so every + // repair succeeds against the provider and the zone still reads + // wrong on the next tick. + let f = Fixture::managing(&[("a.agents.example", ip(1))]).with(|r| r.with_max_attempts(4)); + f.provider.publish("a.agents.example", &ip(9)).unwrap(); + *f.view.frozen.lock().unwrap() = Some( + [("a.agents.example".to_string(), ip(9))] + .into_iter() + .collect(), + ); + for _ in 0..12 { + f.tick(); + } + assert_eq!( + f.provider.withdrawn().len(), + 4, + "argued with four times, then stopped" + ); + let report = f.tick(); + assert_eq!(report.state, ZoneState::Blocked); + assert_eq!(report.quarantined, vec!["a.agents.example".to_string()]); + } + + #[test] + fn a_name_observed_correct_again_gets_its_budget_back() { + let f = Fixture::managing(&[("a.agents.example", ip(1))]).with(|r| r.with_max_attempts(2)); + f.provider.refuse_publish("throttled"); + f.tick(); + f.tick(); // one failed attempt + *f.provider.refuse_publish.lock().unwrap() = None; + f.provider.publish("a.agents.example", &ip(1)).unwrap(); + assert_eq!(f.tick().state, ZoneState::InSync); + // Withdraw it again: the earlier failure must not count against a + // fresh, unrelated drift months later. + f.provider.withdraw("a.agents.example").unwrap(); + f.tick(); + let report = f.tick(); + assert_eq!(report.applied.len(), 1); + } + + #[test] + fn one_tick_writes_no_more_than_its_budget() { + let intended: Vec<(String, RecordTarget)> = (0..30) + .map(|n| (format!("a{n}.agents.example"), ip(1))) + .collect(); + let borrowed: Vec<(&str, RecordTarget)> = intended + .iter() + .map(|(host, target)| (host.as_str(), target.clone())) + .collect(); + let f = Fixture::managing(&borrowed).with(|r| r.with_max_repairs_per_tick(5)); + f.tick(); + let report = f.tick(); + assert_eq!(report.applied.len(), 5); + assert_eq!(f.provider.published().len(), 5); + } + + #[test] + fn concurrent_ticks_publish_each_missing_record_exactly_once() { + use std::sync::Barrier; + + let intended: Vec<(String, RecordTarget)> = (0..8) + .map(|n| (format!("a{n}.agents.example"), ip(1))) + .collect(); + let borrowed: Vec<(&str, RecordTarget)> = intended + .iter() + .map(|(host, target)| (host.as_str(), target.clone())) + .collect(); + let f = Fixture::managing(&borrowed).with(|r| r.with_confirmations(1)); + + let barrier = Arc::new(Barrier::new(16)); + let handles: Vec<_> = (0..16) + .map(|_| { + let reconciler = Arc::clone(&f.reconciler); + let barrier = Arc::clone(&barrier); + std::thread::spawn(move || { + barrier.wait(); + reconciler.tick() + }) + }) + .collect(); + let reports: Vec = handles.into_iter().map(|h| h.join().unwrap()).collect(); + + assert_eq!( + f.provider.publishes.load(Ordering::SeqCst), + 8, + "sixteen concurrent reconcilers made eight writes between them" + ); + assert!(reports.iter().all(|r| r.failed.is_empty())); + assert_eq!(f.provider.published().len(), 8); + } + + #[test] + fn backoff_doubles_on_failing_ticks_and_resets_on_a_clean_one() { + let schedule = Schedule { + interval: Duration::from_secs(10), + max_interval: Duration::from_secs(60), + }; + assert_eq!(schedule.next_delay(0), Duration::from_secs(10)); + assert_eq!(schedule.next_delay(1), Duration::from_secs(20)); + assert_eq!(schedule.next_delay(2), Duration::from_secs(40)); + assert_eq!(schedule.next_delay(3), Duration::from_secs(60)); + assert_eq!(schedule.next_delay(99), Duration::from_secs(60)); + + let f = Fixture::managing(&[("a.agents.example", ip(1))]).with(|r| r.with_schedule(schedule)); + *f.view.fail.lock().unwrap() = Some(ViewError::Unreadable("down".into())); + assert_eq!(f.tick().next_delay, Duration::from_secs(20)); + assert_eq!(f.tick().next_delay, Duration::from_secs(40)); + *f.view.fail.lock().unwrap() = None; + assert_eq!( + f.tick().next_delay, + Duration::from_secs(10), + "one clean tick clears the backoff" + ); + } + + #[test] + fn serving_is_never_gated_on_this_machines_state() { + for state in ZoneState::ALL { + // Exhaustive destructuring: adding a capability to `ZonePolicy` + // that this machine could gate a request on stops compiling + // here, which is the point. + let ZonePolicy { + may_repair: _, + observation_is_evidence: _, + } = state.policy(); + } + assert!(!ZoneState::Blocked.policy().may_repair); + assert!(ZoneState::Drifted.policy().may_repair); + } + + #[test] + fn every_state_admits_staying_put_and_only_one_edge_is_refused() { + for state in ZoneState::ALL { + assert!(state.can_transition_to(*state)); + } + assert!( + !ZoneState::InSync.can_transition_to(ZoneState::Reconciling), + "a repair is never applied on the first tick that sees drift" + ); + assert!(ZoneState::InSync.can_transition_to(ZoneState::Drifted)); + assert!(ZoneState::Blocked.can_transition_to(ZoneState::InSync)); + } + + #[test] + fn a_survey_accounts_for_every_name_on_both_sides() { + let want: BTreeMap = [ + ("gone.agents.example".to_string(), ip(1)), + ("moved.agents.example".to_string(), ip(2)), + ("fine.agents.example".to_string(), ip(3)), + ] + .into_iter() + .collect(); + let live: BTreeMap = [ + ("moved.agents.example".to_string(), ip(9)), + ("fine.agents.example".to_string(), ip(3)), + ("theirs.agents.example".to_string(), ip(9)), + ] + .into_iter() + .collect(); + let survey = Survey::compare(&want, &live); + assert_eq!(survey.missing.len(), 1); + assert_eq!(survey.mismatched.len(), 1); + assert_eq!(survey.unattributed.len(), 1); + assert!(!survey.is_clean()); + assert_eq!(survey.actionable().count(), 2, "the foreign name is not one"); + } + + #[test] + fn a_zone_holding_only_foreign_records_is_in_sync() { + let f = Fixture::managing(&[]); + f.provider.publish("theirs.agents.example", &ip(9)).unwrap(); + let report = f.tick(); + assert_eq!(report.state, ZoneState::InSync); + assert_eq!(report.survey.unattributed.len(), 1); + assert!(report.gates.all_satisfied()); + } + + #[tokio::test(start_paused = true)] + async fn the_task_ticks_on_its_interval_and_stops_when_asked() { + let f = Fixture::managing(&[("a.agents.example", ip(1))]).with(|r| { + r.with_schedule(Schedule { + interval: Duration::from_secs(10), + max_interval: Duration::from_secs(60), + }) + }); + let task = ReconcileTask::spawn(Arc::clone(&f.reconciler)); + tokio::time::sleep(Duration::from_secs(25)).await; + task.stop().await; + assert!( + f.view.reads.load(Ordering::SeqCst) >= 3, + "ticked on its interval rather than once" + ); + assert_eq!(f.reconciler.state(), ZoneState::InSync); + assert_eq!(f.provider.target("a.agents.example"), Some(ip(1))); + } +} -- 2.51.2