diff --git a/crates/didbot-reconcile/src/lib.rs b/crates/didbot-reconcile/src/lib.rs index aa7c75cc..02be28c1 100644 --- a/crates/didbot-reconcile/src/lib.rs +++ b/crates/didbot-reconcile/src/lib.rs @@ -1336,6 +1336,30 @@ mod tests { fn tick(&self) -> TickReport { self.reconciler.tick() } + + /// A brand new reconciler over the same provider, view and intent — + /// the process restarted, with everything `Progress` held lost. + fn restarted(&self) -> ZoneReconciler { + 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() + } + } + ZoneReconciler::new( + ZONE, + ZoneAuthority::Manages, + Box::new(ViewHandle(Arc::clone(&self.view))), + Box::new(IntentHandle(Arc::clone(&self.intent))), + Arc::clone(&self.provider) as Arc, + ) + } } #[test] @@ -1838,4 +1862,127 @@ mod tests { assert_eq!(schedule.next_delay(0), Schedule::MIN_DELAY); assert_eq!(schedule.next_delay(9), Schedule::MIN_DELAY); } + + #[test] + fn a_restart_loses_the_backoff_but_cannot_write_before_its_second_read() { + // Everything `Progress` holds — the confirmation counts, the attempt + // budgets, the backoff — is in memory, so a crash loop starts each + // of them over. That makes the backoff *not* survive a restart, and + // this is the property that keeps that safe rather than expensive: a + // repair needs two consecutive reads from one process, so a process + // that never survives two reads never writes at all. + let f = Fixture::managing(&[("a.agents.example", ip(1))]); + for _ in 0..20 { + let report = f.restarted().tick(); + assert_eq!(report.state, ZoneState::Drifted); + assert!(!report.wrote()); + } + assert_eq!( + f.provider.publishes.load(Ordering::SeqCst), + 0, + "a crash loop hammered the provider with reads and no writes" + ); + // And one process that lives long enough still repairs it. + let survivor = f.restarted(); + survivor.tick(); + assert_eq!(survivor.tick().applied.len(), 1); + + // The other half of the same fact, stated so it is a decision rather + // than an oversight: a restart *does* discard the backoff. A zone + // that has been unreadable for hours is being asked about hourly, + // and a process that crash-loops through that is back to asking on + // the plain interval every time it comes up. + let schedule = Schedule { + interval: Duration::from_secs(10), + max_interval: Duration::from_secs(600), + }; + let g = + Fixture::managing(&[("a.agents.example", ip(1))]).with(|r| r.with_schedule(schedule)); + *g.view.fail.lock().unwrap() = Some(ViewError::Unreadable("down".into())); + let mut backed_off = Duration::ZERO; + for _ in 0..5 { + backed_off = g.tick().next_delay; + } + assert_eq!(backed_off, Duration::from_secs(320)); + assert_eq!( + g.restarted().with_schedule(schedule).tick().next_delay, + Duration::from_secs(20), + "the restarted process asks again as if this were the first failure" + ); + } + + #[test] + fn a_hostile_zone_converges_in_two_passes_and_then_writes_nothing() { + // Every kind of wrong at once: one intended name gone, one pointing + // at somebody else, one foreign name that must survive all of it. + let f = Fixture::managing(&[ + ("gone.agents.example", ip(1)), + ("moved.agents.example", ip(2)), + ("fine.agents.example", ip(3)), + ]); + f.provider.publish("moved.agents.example", &ip(9)).unwrap(); + f.provider.publish("fine.agents.example", &ip(3)).unwrap(); + f.provider.publish("theirs.agents.example", &ip(9)).unwrap(); + let setup = f.provider.publishes.load(Ordering::SeqCst); + + assert_eq!(f.tick().state, ZoneState::Drifted); + let repaired = f.tick(); + assert_eq!(repaired.state, ZoneState::Reconciling); + assert_eq!(repaired.applied.len(), 2); + assert!(repaired.failed.is_empty()); + assert_eq!(f.tick().state, ZoneState::InSync); + + // Idempotent from here: further passes are reads and nothing else. + for _ in 0..5 { + let report = f.tick(); + assert_eq!(report.state, ZoneState::InSync); + assert!(!report.wrote()); + assert_eq!(report.survey.unattributed.len(), 1); + } + assert_eq!( + f.provider.publishes.load(Ordering::SeqCst) - setup, + 2, + "one publish and one replace, and no write after convergence" + ); + assert_eq!(f.provider.withdrawn(), vec!["moved.agents.example"]); + assert_eq!(f.provider.target("theirs.agents.example"), Some(ip(9))); + assert_eq!(f.provider.target("gone.agents.example"), Some(ip(1))); + assert_eq!(f.provider.target("moved.agents.example"), Some(ip(2))); + } + + #[test] + fn a_replace_interrupted_between_its_two_writes_is_resumable() { + // `apply` withdraws and then publishes, and nothing makes the pair + // atomic. Interrupting between them leaves the name *absent* rather + // than answering with somebody else's address — the direction this + // crate already says it prefers — and the run afterwards, with every + // in-memory count lost, has to be able to finish the job. + let f = Fixture::managing(&[("a.agents.example", ip(1))]); + f.provider.publish("a.agents.example", &ip(9)).unwrap(); + f.tick(); + f.provider.refuse_publish("interrupted"); + let torn = f.tick(); + assert_eq!(torn.applied, vec![]); + assert_eq!(torn.failed.len(), 1); + assert_eq!( + f.provider.target("a.agents.example"), + None, + "half-applied: withdrawn, not yet republished" + ); + + *f.provider.refuse_publish.lock().unwrap() = None; + let after = f.restarted(); + assert_eq!(after.tick().state, ZoneState::Drifted); + let report = after.tick(); + assert_eq!( + report.applied, + vec![Repair::Publish { + host: "a.agents.example".into(), + want: ip(1) + }], + "the torn write is finished as a plain missing record" + ); + assert_eq!(f.provider.target("a.agents.example"), Some(ip(1))); + assert_eq!(after.tick().state, ZoneState::InSync); + } }