release 0.2.1: fast recovery from PublishFailing

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

Cargo.lock

@@ -1963,7 +1963,7 @@ dependencies = [
 
 [[package]]
 name = "iroh-gossip-rendezvous"
-version = "0.2.0"
+version = "0.2.1"
 dependencies = [
  "anyhow",
  "arbitrary",

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 <steve@waits.net>"]

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<Duration>,
     /// 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,
         };
 

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::<f32>()
-                    };
-                    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::<f32>()
+                        };
+                        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<usize>,
+        write_attempts: Arc<std::sync::atomic::AtomicUsize>,
+    }
+
+    impl FaultyDht {
+        fn new(fail_writes: usize) -> (Self, Arc<std::sync::atomic::AtomicUsize>) {
+            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<Option<crate::dht::SlotRecord>, 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<dyn DhtSlots> = Arc::new(faulty);
+        let (gossip_view, _joined) = MockGossip::new(vec![]);
+        let gossip: Arc<dyn GossipView> = 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];