From 2f40c81c3aeae9ed5844ad43347f72ef27ad606f Mon Sep 17 00:00:00 2001 From: Stephen Waits Date: Mon, 11 May 2026 16:13:35 -0600 Subject: [PATCH] release 0.2.1: fast recovery from PublishFailing --- CHANGELOG.md | 28 +++++++++ Cargo.lock | 2 +- Cargo.toml | 2 +- src/builder.rs | 24 ++++++++ src/protocol.rs | 147 +++++++++++++++++++++++++++++++++++++++++++++--- 5 files changed, 192 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4fa7655..e263603 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,34 @@ format is based on [Keep a Changelog], and this project adheres to ## [Unreleased] +## [0.2.1] — 2026-05-11 + +### Changed + +- **Fast recovery from `DhtStatus::PublishFailing`.** The write loop + previously slept the full `write_period` (default 5 min) between + every iteration and still applied the `1/(n+1)` probability gate, + so a node stuck in `PublishFailing` could wait many minutes — even + half an hour with a populated active view — before retrying. The + loop now checks `dht_status` before sleeping: while failing, it + uses the new `failure_retry_period` (default 15 s) and bypasses + the gate, so every tick attempts a write until one succeeds. On + success the steady-state cadence and gate are restored. + +### Added + +- `Builder::failure_retry_period(Duration)` — override the recovery + cadence (additive, non-breaking). Public constant + `DEFAULT_FAILURE_RETRY_PERIOD = 15 s`. + +### Rationale + +The probability gate exists to keep N healthy publishers from +storming the DHT with redundant writes — but once *our* publish has +failed, redundancy is irrelevant; we need to land *one* successful +write. Skipping the gate while failing turns the recovery time into +a function of `failure_retry_period`, not of `(n + 1) × write_period`. + ## [0.2.0] — 2026-05-11 ### Changed diff --git a/Cargo.lock b/Cargo.lock index ce249ee..755abeb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1963,7 +1963,7 @@ dependencies = [ [[package]] name = "iroh-gossip-rendezvous" -version = "0.2.0" +version = "0.2.1" dependencies = [ "anyhow", "arbitrary", diff --git a/Cargo.toml b/Cargo.toml index 5706994..a4f49e3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "iroh-gossip-rendezvous" -version = "0.2.0" +version = "0.2.1" edition = "2024" rust-version = "1.93.0" authors = ["Stephen Waits "] diff --git a/src/builder.rs b/src/builder.rs index 89bdfa2..b369e09 100644 --- a/src/builder.rs +++ b/src/builder.rs @@ -33,6 +33,12 @@ pub const DEFAULT_EPOCH_WRITES: u64 = 64; pub const DEFAULT_WRITE_PERIOD: Duration = Duration::from_secs(300); /// T_h — mean heal-loop period. pub const DEFAULT_HEAL_PERIOD: Duration = Duration::from_secs(30); +/// T_f — write-loop period while `DhtStatus::PublishFailing`. Used in +/// place of `T_w` until a publish succeeds; the probability gate is also +/// skipped during recovery so every tick attempts a write. Bounds the +/// recovery latency from a transient DHT failure to roughly this period +/// (± jitter) regardless of neighbor count. +pub const DEFAULT_FAILURE_RETRY_PERIOD: Duration = Duration::from_secs(15); /// σ — jitter factor (± fraction of the period). pub const DEFAULT_JITTER: f32 = 0.5; @@ -49,6 +55,7 @@ pub struct Builder { epoch_size: u64, write_period: Duration, heal_period: Duration, + failure_retry_period: Duration, jitter: f32, wait_for_first_neighbor: Option, /// Test-only: inject a custom DhtSlots backend (e.g., InMemoryDht). @@ -71,6 +78,7 @@ impl Default for Builder { epoch_size: DEFAULT_EPOCH_WRITES, write_period: DEFAULT_WRITE_PERIOD, heal_period: DEFAULT_HEAL_PERIOD, + failure_retry_period: DEFAULT_FAILURE_RETRY_PERIOD, jitter: DEFAULT_JITTER, wait_for_first_neighbor: None, #[cfg(feature = "test-support")] @@ -156,6 +164,21 @@ impl Builder { self } + /// Override `T_f` — the write-loop period used while + /// [`DhtStatus::PublishFailing`](crate::state::DhtStatus::PublishFailing). + /// Default 15 seconds. + /// + /// While failing, the probability gate `1/(n+1)` is bypassed and a + /// write is attempted every tick. Returning to + /// [`DhtStatus::Ready`](crate::state::DhtStatus::Ready) on a + /// successful publish restores the steady-state `write_period` + /// cadence and re-engages the gate. + #[must_use] + pub fn failure_retry_period(mut self, period: Duration) -> Self { + self.failure_retry_period = period; + self + } + /// Override `σ`. Default 0.5 (± 50%). #[must_use] pub fn jitter(mut self, sigma: f32) -> Self { @@ -257,6 +280,7 @@ impl Builder { epoch_size: self.epoch_size, write_period: self.write_period, heal_period: self.heal_period, + failure_retry_period: self.failure_retry_period, jitter: self.jitter, }; diff --git a/src/protocol.rs b/src/protocol.rs index 0325b4e..0388bc2 100644 --- a/src/protocol.rs +++ b/src/protocol.rs @@ -48,6 +48,12 @@ pub(crate) struct ProtocolConfig { pub epoch_size: u64, pub write_period: Duration, pub heal_period: Duration, + /// Sleep between write-loop ticks while `dht_status == PublishFailing`. + /// Used in place of `write_period` to recover from a failed publish + /// faster than the steady-state cadence allows. The probability gate + /// is also skipped while failing — every tick attempts a write until + /// one succeeds. + pub failure_retry_period: Duration, pub jitter: f32, } @@ -376,19 +382,39 @@ pub(crate) fn spawn_loops( first_iter = false; true } else { - let sleep = jittered(state.config.write_period, state.config.jitter); + // While the last publish is failing, retry on the + // shorter `failure_retry_period` cadence and bypass + // the `1/(n+1)` probability gate — the gate exists + // to avoid swarm-wide write storms when N is large + // and the DHT is healthy, but once *our* publish + // has failed we should retry unconditionally until + // it succeeds. + let failing = matches!( + state.observable.lock().map(|s| s.dht_status), + Ok(DhtStatus::PublishFailing) + ); + let period = if failing { + state.config.failure_retry_period + } else { + state.config.write_period + }; + let sleep = jittered(period, state.config.jitter); tokio::select! { () = tokio::time::sleep(sleep) => {} () = cancel.cancelled() => break, } - // Write with probability 1/(n+1). Scope the (non-Send) - // ThreadRng so it doesn't live across the await. - let n = gossip.neighbors().len(); - let r = { - let mut rng = rand::rng(); - rng.random::() - }; - should_fire_write(r, n) + if failing { + true + } else { + // Scope the (non-Send) ThreadRng so it doesn't + // live across the await. + let n = gossip.neighbors().len(); + let r = { + let mut rng = rand::rng(); + rng.random::() + }; + should_fire_write(r, n) + } }; if should_write { write_once(&state, dht.as_ref(), gossip.as_ref()).await; @@ -472,6 +498,61 @@ mod tests { } } + /// Fault-injecting `DhtSlots`. Fails the first `fail_writes` write calls + /// (counting attempts, not slots), then delegates to an inner + /// [`InMemoryDht`]. Reads always delegate. + struct FaultyDht { + inner: InMemoryDht, + remaining_failures: Mutex, + write_attempts: Arc, + } + + impl FaultyDht { + fn new(fail_writes: usize) -> (Self, Arc) { + let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + ( + Self { + inner: InMemoryDht::new(), + remaining_failures: Mutex::new(fail_writes), + write_attempts: attempts.clone(), + }, + attempts, + ) + } + } + + #[async_trait] + impl DhtSlots for FaultyDht { + async fn read( + &self, + slot: crate::dht::SlotKey, + ) -> Result, crate::dht::DhtError> { + self.inner.read(slot).await + } + + async fn write( + &self, + slot: crate::dht::SlotKey, + record: crate::dht::SlotRecord, + ) -> Result<(), crate::dht::DhtError> { + self.write_attempts + .fetch_add(1, std::sync::atomic::Ordering::SeqCst); + let should_fail = { + let mut r = self.remaining_failures.lock().unwrap(); + if *r > 0 { + *r -= 1; + true + } else { + false + } + }; + if should_fail { + return Err(crate::dht::DhtError::Transport("injected".into())); + } + self.inner.write(slot, record).await + } + } + fn test_config(self_id: [u8; 32], shards: usize) -> ProtocolConfig { ProtocolConfig { self_id, @@ -482,6 +563,7 @@ mod tests { epoch_size: 64, write_period: Duration::from_secs(1), heal_period: Duration::from_secs(1), + failure_retry_period: Duration::from_millis(50), jitter: 0.5, } } @@ -567,6 +649,53 @@ mod tests { } } + #[tokio::test] + async fn write_loop_recovers_from_publish_failing_on_short_period() { + // Loop should retry on `failure_retry_period` (not `write_period`) + // and bypass the `1/(n+1)` gate while `DhtStatus::PublishFailing`. + // Setup: write_period=10s, failure_retry_period=30ms, dht fails + // first 3 writes then succeeds. Within a few hundred ms we should + // see ≥4 attempts (3 fails + 1 success) and status == Ready. + + let self_id = [42u8; 32]; + let mut config = test_config(self_id, 3); + config.write_period = Duration::from_secs(10); + config.heal_period = Duration::from_secs(10); + config.failure_retry_period = Duration::from_millis(30); + // No jitter so the test is deterministic. + config.jitter = 0.0; + + let state = Arc::new(test_state(config)); + let (faulty, attempts) = FaultyDht::new(3); + let dht: Arc = Arc::new(faulty); + let (gossip_view, _joined) = MockGossip::new(vec![]); + let gossip: Arc = Arc::new(gossip_view); + let cancel = CancellationToken::new(); + + let mut tasks = spawn_loops(state.clone(), dht.clone(), gossip.clone(), cancel.clone()); + + // 4 × 30ms = 120ms minimum. Give 500ms — comfortably above the + // retry budget, comfortably below `write_period`. If the loop + // were waiting on `write_period` (10s) we'd see only 1 attempt. + tokio::time::sleep(Duration::from_millis(500)).await; + + let made = attempts.load(std::sync::atomic::Ordering::SeqCst); + let status = state.observable.lock().unwrap().dht_status; + + cancel.cancel(); + while tasks.join_next().await.is_some() {} + + assert!( + made >= 4, + "expected ≥4 write attempts (3 fails + 1 success) within 500ms; got {made}" + ); + assert_eq!( + status, + DhtStatus::Ready, + "status should flip to Ready after the 4th write succeeded" + ); + } + #[tokio::test] async fn seq_is_monotonic_across_writes() { let self_id = [1u8; 32];