unify lock logic

25 files changed, 116 insertions, 134 deletions

diff --git a/src/api/peers.rs b/src/api/peers.rs
index a9694f7..fac3440 100644
--- a/src/api/peers.rs
+++ b/src/api/peers.rs
@@ -16,6 +16,7 @@ use utoipa::ToSchema;
 use super::store_status;
 use crate::document::replica::Epoch;
 use crate::document::store;
+use crate::locks::MutexExt;
 use crate::state::{AppState, HelloState, Incidents, Liveness, PeerState};
 use crate::swarm::peer::forgotten_path;
 use crate::transport::traffic::PeerTraffic;
@@ -141,8 +142,7 @@ pub(crate) async fn list_peers(
         let traffic = state.traffic.peer(&public_key).unwrap_or_default();
         let sync_throughput = state
             .throughput
-            .lock()
-            .unwrap_or_else(|poisoned| poisoned.into_inner())
+            .lock_or_recover()
             .get(&public_key)
             .map(|estimate| ThroughputInfo {
                 bytes_per_sec: estimate.bytes_per_sec.round(),
diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs
index 74c18e9..07a6cc8 100644
--- a/src/api/stats/prometheus.rs
+++ b/src/api/stats/prometheus.rs
@@ -7,6 +7,7 @@ use std::fmt::Write;
 use tokio::time::Instant;
 
 use super::StatsResponse;
+use crate::locks::MutexExt;
 use crate::state::{AppState, Incidents, Liveness};
 use crate::transport::traffic::{ByKind, Counter, PeerTraffic};
 
@@ -136,8 +137,7 @@ async fn peers(state: &AppState) -> Vec<Peer> {
             .map(|rtt| rtt.as_secs_f64());
         let throughput = state
             .throughput
-            .lock()
-            .unwrap_or_else(|poisoned| poisoned.into_inner())
+            .lock_or_recover()
             .get(&key)
             .map(|estimate| estimate.bytes_per_sec);
         let traffic = state.traffic.peer(&key).unwrap_or_default();
diff --git a/src/document/file.rs b/src/document/file.rs
index a773502..6e7a4b6 100644
--- a/src/document/file.rs
+++ b/src/document/file.rs
@@ -28,6 +28,7 @@ use tracing::{debug, error, warn};
 
 use crate::document::replica::Epoch;
 use crate::files::{FileKind, Files};
+use crate::locks::MutexExt;
 use crate::state::AppState;
 
 const MAGIC: &[u8] = b"continuum-document/1\n";
@@ -175,9 +176,7 @@ impl DocumentFile {
     }
 
     fn lock(&self) -> MutexGuard<'_, Option<Written>> {
-        self.written
-            .lock()
-            .unwrap_or_else(|poisoned| poisoned.into_inner())
+        self.written.lock_or_recover()
     }
 }
 
diff --git a/src/files.rs b/src/files.rs
index 58fd043..4b08469 100644
--- a/src/files.rs
+++ b/src/files.rs
@@ -8,6 +8,7 @@
 //! end and are flushed to disk before they count as done; a crash may leave the last one
 //! cut short, which the file's reader has to expect.
 
+use crate::locks::MutexExt;
 use std::io::Write;
 use std::path::{Path, PathBuf};
 use std::sync::{Mutex, MutexGuard};
@@ -67,9 +68,7 @@ pub(crate) struct Files {
 
 impl Files {
     fn lock(&self) -> MutexGuard<'_, [Counts; FileKind::ALL.len()]> {
-        self.counts
-            .lock()
-            .unwrap_or_else(|poisoned| poisoned.into_inner())
+        self.counts.lock_or_recover()
     }
 
     fn count(&self, kind: FileKind, f: impl FnOnce(&mut Counts)) {
diff --git a/src/locks.rs b/src/locks.rs
new file mode 100644
index 0000000..ad66e6f
--- /dev/null
+++ b/src/locks.rs
@@ -0,0 +1,53 @@
+//! Std locks without poisoning. A panic while one is held leaves its data as it was, and
+//! every lock of this kind here guards counters, tables or caches that stay usable: so
+//! the next user takes the data as it is instead of panicking too. (The tokio locks
+//! around the document and the peers don't poison.)
+
+use std::sync::{Mutex, MutexGuard, RwLock, RwLockReadGuard, RwLockWriteGuard};
+
+pub(crate) trait MutexExt<T> {
+    /// Locks, recovering from a poisoned mutex.
+    fn lock_or_recover(&self) -> MutexGuard<'_, T>;
+}
+
+impl<T> MutexExt<T> for Mutex<T> {
+    fn lock_or_recover(&self) -> MutexGuard<'_, T> {
+        self.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
+    }
+}
+
+pub(crate) trait RwLockExt<T> {
+    /// Locks for reading, recovering from a poisoned lock.
+    fn read_or_recover(&self) -> RwLockReadGuard<'_, T>;
+    /// Locks for writing, recovering from a poisoned lock.
+    fn write_or_recover(&self) -> RwLockWriteGuard<'_, T>;
+}
+
+impl<T> RwLockExt<T> for RwLock<T> {
+    fn read_or_recover(&self) -> RwLockReadGuard<'_, T> {
+        self.read().unwrap_or_else(|poisoned| poisoned.into_inner())
+    }
+
+    fn write_or_recover(&self) -> RwLockWriteGuard<'_, T> {
+        self.write()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn a_panic_while_locked_doesnt_spread() {
+        let mutex = std::sync::Arc::new(Mutex::new(1));
+        let poisoner = std::sync::Arc::clone(&mutex);
+        let _ = std::thread::spawn(move || {
+            let _guard = poisoner.lock().unwrap();
+            panic!("while holding the lock");
+        })
+        .join();
+        assert!(mutex.is_poisoned());
+        assert_eq!(*mutex.lock_or_recover(), 1);
+    }
+}
diff --git a/src/main.rs b/src/main.rs
index bfc45f0..a79c6a2 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -19,6 +19,7 @@ mod config;
 mod document;
 mod files;
 mod force;
+mod locks;
 mod logging;
 mod node;
 mod remake;
diff --git a/src/remake/local.rs b/src/remake/local.rs
index 517367b..8e9d8ba 100644
--- a/src/remake/local.rs
+++ b/src/remake/local.rs
@@ -16,6 +16,7 @@ use tokio::time::Instant;
 use super::records::Retention;
 use super::snapshot::Snapshot;
 use crate::document::replica::Epoch;
+use crate::locks::MutexExt;
 
 pub(crate) struct RemakeState {
     /// When this node started (Unix milliseconds): votes it cast before may be lost
@@ -185,9 +186,7 @@ impl RemakeState {
     }
 
     fn lock(&self) -> MutexGuard<'_, Inner> {
-        self.inner
-            .lock()
-            .unwrap_or_else(|poisoned| poisoned.into_inner())
+        self.inner.lock_or_recover()
     }
 
     pub(crate) fn store_prebuilt(&self, id: &str, snapshot: Snapshot) {
diff --git a/src/remake/switch.rs b/src/remake/switch.rs
index 65f296f..903baea 100644
--- a/src/remake/switch.rs
+++ b/src/remake/switch.rs
@@ -38,6 +38,7 @@ use super::vote;
 use crate::document::meta::{self, REMAKES};
 use crate::document::replica::Epoch;
 use crate::document::store;
+use crate::locks::MutexExt;
 use crate::swarm::sync::paths_related;
 
 const TICK: Duration = Duration::from_secs(1);
@@ -311,16 +312,10 @@ pub(super) async fn swap_in<H: Host>(
                     "Replayed late changes"
                 );
             }
-            *late_out
-                .lock()
-                .unwrap_or_else(|poisoned| poisoned.into_inner()) = late;
+            *late_out.lock_or_recover() = late;
             Ok(new_doc)
         })
         .await?;
-    let late = std::mem::take(
-        &mut *late_stats
-            .lock()
-            .unwrap_or_else(|poisoned| poisoned.into_inner()),
-    );
+    let late = std::mem::take(&mut *late_stats.lock_or_recover());
     Ok((old, late))
 }
diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs
index e8e6e5b..408f53e 100644
--- a/src/simtest/mod.rs
+++ b/src/simtest/mod.rs
@@ -24,6 +24,7 @@ use tokio::time::Instant;
 use crate::document::replica::Epoch;
 use crate::document::store;
 use crate::files::Files;
+use crate::locks::RwLockExt;
 use crate::node::{self, Node, NodeConfig};
 use crate::state::{AppState, PeerState, Timing};
 use crate::swarm::bootstrap::{self, Start};
@@ -198,14 +199,7 @@ impl Sim {
                 node.initial_sync_done
                     .load(std::sync::atomic::Ordering::Acquire),
             );
-            let _ = writeln!(
-                out,
-                "  view: {:?}",
-                *node
-                    .neighbours
-                    .read()
-                    .unwrap_or_else(|poisoned| poisoned.into_inner())
-            );
+            let _ = writeln!(out, "  view: {:?}", *node.neighbours.read_or_recover());
             let peers: Vec<_> = node
                 .peer_states
                 .read()
@@ -275,11 +269,7 @@ pub(crate) async fn heads(node: &AppState) -> Vec<ChangeHash> {
 /// Whether `node` has `peer` among its neighbours as computed from the member list.
 /// Before that list is known everyone counts as a neighbour, so this waits for it.
 pub(crate) fn has_neighbour(node: &AppState, peer: &str) -> bool {
-    node.neighbours
-        .read()
-        .unwrap_or_else(|poisoned| poisoned.into_inner())
-        .is_some()
-        && crate::swarm::topology::is_neighbour(node, peer)
+    node.neighbours.read_or_recover().is_some() && crate::swarm::topology::is_neighbour(node, peer)
 }
 
 /// Something about `node`'s state for `peer`; None if it doesn't know the peer.
diff --git a/src/state/host.rs b/src/state/host.rs
index 863cc70..9b149b8 100644
--- a/src/state/host.rs
+++ b/src/state/host.rs
@@ -13,6 +13,7 @@ use tracing::debug;
 
 use super::{AppState, HelloState, Liveness, NodeLifecycle, PeerState, StaleView, Standing};
 use crate::document::replica::Epoch;
+use crate::locks::MutexExt;
 use crate::remake::{Host, PeerSummary, RemakeState};
 use crate::swarm::sync::{ChangeSet, SyncEvent};
 
@@ -195,14 +196,8 @@ impl Host for AppState {
     fn reset_initial_sync(&self) {
         self.initial_sync_done
             .store(false, std::sync::atomic::Ordering::Release);
-        *self
-            .catch_up_source
-            .lock()
-            .unwrap_or_else(|poisoned| poisoned.into_inner()) = None;
-        *self
-            .catch_up_preference
-            .lock()
-            .unwrap_or_else(|poisoned| poisoned.into_inner()) = None;
+        *self.catch_up_source.lock_or_recover() = None;
+        *self.catch_up_preference.lock_or_recover() = None;
     }
 
     fn subscribe_changes(&self) -> broadcast::Receiver<Arc<ChangeSet>> {
diff --git a/src/state/mod.rs b/src/state/mod.rs
index c5a6885..3336b00 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -30,6 +30,7 @@ use tokio::time::Instant;
 use tracing::{debug, error, warn};
 
 use crate::document::replica::{Epoch, Replica};
+use crate::locks::MutexExt;
 use crate::remake::{Host, RemakeState};
 use crate::swarm::sync::{ChangeSet, SyncEvent};
 use crate::swarm::throughput::ThroughputTable;
@@ -174,9 +175,7 @@ impl AppState {
     }
 
     pub(crate) fn lock_rtt(&self) -> std::sync::MutexGuard<'_, crate::swarm::rtt::RttTable> {
-        self.rtt
-            .lock()
-            .unwrap_or_else(|poisoned| poisoned.into_inner())
+        self.rtt.lock_or_recover()
     }
 
     /// The local epoch. Code that labels or applies sync messages must use the epoch
diff --git a/src/state/read.rs b/src/state/read.rs
index b76582c..f8323a2 100644
--- a/src/state/read.rs
+++ b/src/state/read.rs
@@ -6,6 +6,7 @@ use automerge::{AutoCommit, ChangeHash};
 
 use super::AppState;
 use crate::document::replica::Epoch;
+use crate::locks::RwLockExt;
 
 /// A read-only view of the document the node served before it started rejoining.
 pub(crate) struct StaleView {
@@ -46,8 +47,7 @@ impl AppState {
 
         let stale = self
             .stale
-            .read()
-            .unwrap_or_else(|poisoned| poisoned.into_inner())
+            .read_or_recover()
             .as_ref()
             .map(|view| (view.epoch, Arc::clone(&view.doc)));
         let Some((epoch, doc)) = stale else {
@@ -92,9 +92,6 @@ impl AppState {
     }
 
     pub(super) fn set_stale(&self, view: Option<StaleView>) {
-        *self
-            .stale
-            .write()
-            .unwrap_or_else(|poisoned| poisoned.into_inner()) = view;
+        *self.stale.write_or_recover() = view;
     }
 }
diff --git a/src/swarm/auth.rs b/src/swarm/auth.rs
index 31230a5..1bca60d 100644
--- a/src/swarm/auth.rs
+++ b/src/swarm/auth.rs
@@ -19,6 +19,7 @@ use std::path::Path;
 use std::sync::Mutex;
 use std::time::{Duration, Instant};
 
+use crate::locks::MutexExt;
 use anyhow::{Context, Result, bail};
 use sha2::{Digest, Sha256};
 
@@ -67,10 +68,7 @@ impl SwarmKey {
     /// Records a rejected handshake. Returns how many went unlogged since the last log
     /// line if this one should be logged, None if it should be suppressed.
     pub(crate) fn note_rejection(&self) -> Option<u64> {
-        let mut rejections = self
-            .rejections
-            .lock()
-            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        let mut rejections = self.rejections.lock_or_recover();
         let due = rejections
             .last_logged
             .is_none_or(|last| last.elapsed() >= REJECTION_LOG_INTERVAL);
diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs
index 42cfbf3..58522fa 100644
--- a/src/swarm/blob.rs
+++ b/src/swarm/blob.rs
@@ -29,6 +29,7 @@ use tokio::sync::{Semaphore, mpsc};
 use tokio::time::Instant;
 use tracing::{debug, info, warn};
 
+use crate::locks::MutexExt;
 use crate::state::AppState;
 use crate::transport::{BlobBody, BlobId, BlobMessage, ParsedMessage};
 
@@ -80,25 +81,19 @@ impl Blobs {
     fn fetches(
         &self,
     ) -> MutexGuard<'_, HashMap<BlobId, mpsc::UnboundedSender<(String, BlobBody)>>> {
-        self.fetches
-            .lock()
-            .unwrap_or_else(|poisoned| poisoned.into_inner())
+        self.fetches.lock_or_recover()
     }
 
     fn cached(&self, id: BlobId) -> Option<Arc<Prepared>> {
         self.cache
-            .lock()
-            .unwrap_or_else(|poisoned| poisoned.into_inner())
+            .lock_or_recover()
             .as_ref()
             .filter(|(cached, _)| *cached == id)
             .map(|(_, prepared)| Arc::clone(prepared))
     }
 
     fn remember(&self, id: BlobId, prepared: Arc<Prepared>) {
-        *self
-            .cache
-            .lock()
-            .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some((id, prepared));
+        *self.cache.lock_or_recover() = Some((id, prepared));
     }
 }
 
diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs
index 651e067..ec6543b 100644
--- a/src/swarm/peer.rs
+++ b/src/swarm/peer.rs
@@ -14,6 +14,7 @@ use tracing::{debug, info, instrument, trace, warn};
 use crate::document::meta;
 use crate::document::replica::{Epoch, EpochRelation};
 use crate::document::store::{self, StoreError};
+use crate::locks::MutexExt;
 use crate::state::{AppState, HelloState, Liveness, PeerState, Standing};
 use crate::swarm::sync::{SyncEvent, handle_sync};
 use crate::swarm::topology;
@@ -543,8 +544,7 @@ pub(crate) async fn run_peer_worker(state: AppState) {
                 state.traffic.retain(|peer| known.contains_key(peer));
                 state
                     .throughput
-                    .lock()
-                    .unwrap_or_else(|poisoned| poisoned.into_inner())
+                    .lock_or_recover()
                     .retain(|peer| known.contains_key(peer));
             }
             // Fallback for what doesn't notify (`topology::run_worker`).
diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
index 703bc3f..5053be7 100644
--- a/src/swarm/relay.rs
+++ b/src/swarm/relay.rs
@@ -25,6 +25,7 @@ use std::time::Duration;
 use tokio::time::Instant;
 use tracing::debug;
 
+use crate::locks::MutexExt;
 use crate::state::{AppState, Standing};
 use crate::swarm::peer::send_ping;
 use crate::swarm::sync::SyncEvent;
@@ -45,10 +46,7 @@ pub(crate) async fn announce(state: &AppState, from: &str) {
         return;
     }
     let schedule = {
-        let mut pending = state
-            .announce_pending
-            .lock()
-            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        let mut pending = state.announce_pending.lock_or_recover();
         match pending.as_mut() {
             Some(senders) => {
                 senders.insert(from.to_owned());
@@ -68,8 +66,7 @@ pub(crate) async fn announce(state: &AppState, from: &str) {
         tokio::time::sleep(ANNOUNCE_DELAY).await;
         let senders = state
             .announce_pending
-            .lock()
-            .unwrap_or_else(|poisoned| poisoned.into_inner())
+            .lock_or_recover()
             .take()
             .unwrap_or_default();
         announce_now(&state, &senders).await;
@@ -128,10 +125,7 @@ pub(crate) async fn pull_if_behind(state: &AppState, peer: &str, remote: &transp
 
     let now = Instant::now();
     {
-        let mut pulling = state
-            .pulling
-            .lock()
-            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        let mut pulling = state.pulling.lock_or_recover();
         let busy = matches!(
             &*pulling,
             Some((other, since)) if other != peer && now < *since + PULL_TIMEOUT
@@ -169,10 +163,7 @@ pub(crate) async fn pull_if_behind(state: &AppState, peer: &str, remote: &transp
 
 /// A sync message from `peer` was applied: a pull from it is answered.
 pub(crate) fn pulled(state: &AppState, peer: &str) {
-    let mut pulling = state
-        .pulling
-        .lock()
-        .unwrap_or_else(|poisoned| poisoned.into_inner());
+    let mut pulling = state.pulling.lock_or_recover();
     if pulling.as_ref().is_some_and(|(p, _)| p == peer) {
         *pulling = None;
     }
diff --git a/src/swarm/source.rs b/src/swarm/source.rs
index b1701b3..e7f06fb 100644
--- a/src/swarm/source.rs
+++ b/src/swarm/source.rs
@@ -35,6 +35,7 @@ use std::time::Duration;
 use tokio::time::Instant;
 use tracing::{debug, info, warn};
 
+use crate::locks::MutexExt;
 use crate::state::{AppState, CatchUpSource, PeerState};
 use crate::swarm::sync::SyncEvent;
 
@@ -55,17 +56,11 @@ const REPLY_TIMEOUTS: u32 = 3;
 const STALL_WARNING: Duration = Duration::from_secs(60);
 
 fn lock_preference(state: &AppState) -> std::sync::MutexGuard<'_, Option<Instant>> {
-    state
-        .catch_up_preference
-        .lock()
-        .unwrap_or_else(|poisoned| poisoned.into_inner())
+    state.catch_up_preference.lock_or_recover()
 }
 
 fn lock_avoided(state: &AppState) -> std::sync::MutexGuard<'_, Option<(String, Instant)>> {
-    state
-        .catch_up_avoided
-        .lock()
-        .unwrap_or_else(|poisoned| poisoned.into_inner())
+    state.catch_up_avoided.lock_or_recover()
 }
 
 fn catching_up(state: &AppState) -> bool {
@@ -73,10 +68,7 @@ fn catching_up(state: &AppState) -> bool {
 }
 
 fn lock(state: &AppState) -> std::sync::MutexGuard<'_, Option<CatchUpSource>> {
-    state
-        .catch_up_source
-        .lock()
-        .unwrap_or_else(|poisoned| poisoned.into_inner())
+    state.catch_up_source.lock_or_recover()
 }
 
 /// Returns how long to hold back sync with `peer`, if at all: while catching up, with
diff --git a/src/swarm/stability.rs b/src/swarm/stability.rs
index 9d79b17..1ee308f 100644
--- a/src/swarm/stability.rs
+++ b/src/swarm/stability.rs
@@ -22,6 +22,7 @@ use std::time::Duration;
 use tokio::time::Instant;
 use tracing::{info, warn};
 
+use crate::locks::MutexExt;
 use crate::state::{AppState, HelloState};
 use crate::swarm::members::{self, Tier};
 
@@ -129,10 +130,7 @@ impl Stability {
 }
 
 fn lock(state: &AppState) -> std::sync::MutexGuard<'_, Stability> {
-    state
-        .stability
-        .lock()
-        .unwrap_or_else(|poisoned| poisoned.into_inner())
+    state.stability.lock_or_recover()
 }
 
 /// The fraction of the observed time (up to a day) we had a live peer, and how much
diff --git a/src/swarm/state_file.rs b/src/swarm/state_file.rs
index 8b883fc..ba9192e 100644
--- a/src/swarm/state_file.rs
+++ b/src/swarm/state_file.rs
@@ -15,6 +15,7 @@ use tokio::time::Instant;
 use tracing::{error, info, warn};
 
 use crate::files::{FileKind, Files};
+use crate::locks::MutexExt;
 use crate::state::AppState;
 use crate::swarm::members;
 
@@ -90,9 +91,7 @@ impl StateFile {
     }
 
     fn lock(&self) -> MutexGuard<'_, Inner> {
-        self.inner
-            .lock()
-            .unwrap_or_else(|poisoned| poisoned.into_inner())
+        self.inner.lock_or_recover()
     }
 
     fn write(&self, remembered: &Remembered) -> std::io::Result<()> {
diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs
index eb34276..f9a5b7b 100644
--- a/src/swarm/stats.rs
+++ b/src/swarm/stats.rs
@@ -9,6 +9,7 @@ use tokio::time::Instant;
 use tracing::{debug, warn};
 
 use crate::document::replica::Epoch;
+use crate::locks::MutexExt;
 use crate::state::AppState;
 
 /// Saving takes time proportional to the document; we spend at most about 1% of it on
@@ -120,9 +121,7 @@ impl Default for NodeStats {
 
 impl NodeStats {
     fn lock(&self) -> MutexGuard<'_, Inner> {
-        self.inner
-            .lock()
-            .unwrap_or_else(|poisoned| poisoned.into_inner())
+        self.inner.lock_or_recover()
     }
 
     pub(crate) fn caught_up(&self, sync_bytes: u64) {
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index a05e346..fc6dc43 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -11,6 +11,7 @@ use tokio::sync::mpsc;
 use tokio::time::Instant;
 use tracing::{debug, info, trace, warn};
 
+use crate::locks::MutexExt;
 use crate::state::{AppState, Liveness, Sending, Standing};
 use crate::swarm::peer::{peer_swarm, send_hello_once};
 use crate::swarm::source;
@@ -355,8 +356,7 @@ pub(crate) async fn handle_sync(
             peer_state.last_sync_rx = Some(Instant::now());
             if let Some(wait) = &peer_state.sync.reply {
                 throughput
-                    .lock()
-                    .unwrap_or_else(|poisoned| poisoned.into_inner())
+                    .lock_or_recover()
                     .exchange(&peer, wait.started, arrived, rtt);
             }
             peer_state.sync.reply_settled();
diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs
index 0dae7e9..73c2f2c 100644
--- a/src/swarm/topology.rs
+++ b/src/swarm/topology.rs
@@ -23,6 +23,7 @@ use sha2::{Digest, Sha256};
 use tokio::time::Instant;
 use tracing::info;
 
+use crate::locks::RwLockExt;
 use crate::state::AppState;
 use crate::swarm::members;
 
@@ -42,8 +43,7 @@ pub(crate) struct NeighbourView {
 pub(crate) fn is_full_mesh(state: &AppState) -> bool {
     state
         .neighbours
-        .read()
-        .unwrap_or_else(|poisoned| poisoned.into_inner())
+        .read_or_recover()
         .as_ref()
         .is_none_or(|view| view.full_mesh)
 }
@@ -65,8 +65,7 @@ pub(crate) fn summary(state: &AppState) -> Option<Summary> {
     let local = state.local_id();
     state
         .neighbours
-        .read()
-        .unwrap_or_else(|poisoned| poisoned.into_inner())
+        .read_or_recover()
         .as_ref()
         .map(|view| Summary {
             members: view.members.len(),
@@ -91,11 +90,7 @@ pub(crate) fn summary(state: &AppState) -> Option<Summary> {
 /// any member registered), so joining works as in a full mesh. We always *answer*
 /// anyone who contacts us.
 pub(crate) fn is_neighbour(state: &AppState, peer: &str) -> bool {
-    match &*state
-        .neighbours
-        .read()
-        .unwrap_or_else(|poisoned| poisoned.into_inner())
-    {
+    match &*state.neighbours.read_or_recover() {
         Some(view) => view.neighbours.contains(peer),
         None => true,
     }
@@ -154,10 +149,7 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
         }
     }
     let previous = {
-        let view = state
-            .neighbours
-            .read()
-            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        let view = state.neighbours.read_or_recover();
         if view.as_ref().is_some_and(|view| {
             view.members == current
                 && view.k == k
@@ -207,10 +199,7 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
         .iter()
         .filter(|m| m.as_str() != state.local_id() && !unreachable.contains(*m))
         .count();
-    *state
-        .neighbours
-        .write()
-        .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(NeighbourView {
+    *state.neighbours.write_or_recover() = Some(NeighbourView {
         full_mesh: computed.len() >= reachable_others,
         members: current,
         k,
diff --git a/src/transport/firewall.rs b/src/transport/firewall.rs
index b8e05cd..c8592bd 100644
--- a/src/transport/firewall.rs
+++ b/src/transport/firewall.rs
@@ -6,6 +6,7 @@ use async_trait::async_trait;
 use serde::{Deserialize, Serialize};
 use utoipa::ToSchema;
 
+use crate::locks::RwLockExt;
 use crate::transport::EncodedMessage;
 use crate::transport::{MessageHandler, ParsedMessage, Transport};
 
@@ -27,9 +28,7 @@ pub struct Firewall {
 
 impl Firewall {
     fn read(&self) -> RwLockReadGuard<'_, FirewallRules> {
-        self.rules
-            .read()
-            .unwrap_or_else(|poisoned| poisoned.into_inner())
+        self.rules.read_or_recover()
     }
 
     pub fn rules(&self) -> FirewallRules {
@@ -37,10 +36,7 @@ impl Firewall {
     }
 
     pub fn set_rules(&self, rules: FirewallRules) {
-        *self
-            .rules
-            .write()
-            .unwrap_or_else(|poisoned| poisoned.into_inner()) = rules;
+        *self.rules.write_or_recover() = rules;
     }
 
     pub fn blocks_inbound(&self, peer: &str) -> bool {
diff --git a/src/transport/sim.rs b/src/transport/sim.rs
index ce5977f..675f382 100644
--- a/src/transport/sim.rs
+++ b/src/transport/sim.rs
@@ -26,6 +26,7 @@ use tracing::{debug, warn};
 
 use super::mycelium::{decode_message, encode_message};
 use super::{EncodedMessage, MessageHandler, MessageKind, ParsedMessage, Transport};
+use crate::locks::MutexExt;
 use crate::testrand::Rng;
 
 /// How one direction of a link behaves.
@@ -102,9 +103,7 @@ impl SimNet {
     }
 
     fn lock(&self) -> MutexGuard<'_, NetInner> {
-        self.inner
-            .lock()
-            .unwrap_or_else(|poisoned| poisoned.into_inner())
+        self.inner.lock_or_recover()
     }
 
     /// A transport for a new node `id` on this net.
diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs
index 9dcd91a..3d17860 100644
--- a/src/transport/traffic.rs
+++ b/src/transport/traffic.rs
@@ -16,6 +16,7 @@ use async_trait::async_trait;
 use serde::Serialize;
 use utoipa::ToSchema;
 
+use crate::locks::MutexExt;
 use crate::transport::{EncodedMessage, MessageHandler, MessageKind, ParsedMessage, Transport};
 
 #[derive(Debug, Default, Clone, Copy, Serialize, ToSchema)]
@@ -79,9 +80,7 @@ pub struct Traffic {
 
 impl Traffic {
     fn lock(&self) -> MutexGuard<'_, Table> {
-        self.table
-            .lock()
-            .unwrap_or_else(|poisoned| poisoned.into_inner())
+        self.table.lock_or_recover()
     }
 
     fn count(&self, peer: &str, bytes: usize, pick: impl Fn(&mut PeerTraffic) -> &mut Counter) {