server/lib/src/lock.rs
//! **The ONE mutex-poison rule of `server/lib`** (lane C3, 2026-10-20).
//!
//! A `Mutex` or `RwLock` is poisoned when a thread panicked while it held the guard.
//! holger had three answers to that: 32 `lock().expect(..)` sites that took the
//! panic to every later caller (a poisoned job-queue mutex killed every RPC after
//! it), 85 sites that recovered silently, and a handful that refused with a string.
//!
//! **The rule: a poisoned lock is RECOVERED, and the poisoning is LOGGED ONCE.**
//! Every guarded value in `server/lib` is either a cache, a board, a counter or a
//! store whose on-disk half is the truth, and every one of them is written in a way
//! that leaves it usable when a writer stops halfway (the panic was the bug, not
//! the state). So the guard is taken back with [`PoisonError::into_inner`], an
//! `error!` names the call site that met the poison, and the poison flag is cleared
//! so the same poisoning is not logged again by every later caller. A refusal by
//! name would only turn one panic into a permanent outage of that subsystem.
//!
//! Use [`LockOrRecover::lock_or_recover`], [`RwLockOrRecover::read_or_recover`] /
//! [`RwLockOrRecover::write_or_recover`] and [`recover`] (for a `Condvar` wait).
//! Never `lock().unwrap()`, `lock().expect(..)` or a hand-written
//! `unwrap_or_else(|p| p.into_inner())` in production code; the census test below
//! fails on any of them.
use std::sync::{
LockResult, Mutex, MutexGuard, PoisonError, RwLock, RwLockReadGuard, RwLockWriteGuard,
};
#[track_caller]
fn poisoned(kind: &str) {
let at = std::panic::Location::caller();
log::error!(
"holger: a {kind} was poisoned (a thread panicked while holding it); recovered at {}:{} and the poison cleared",
at.file(),
at.line()
);
}
/// [`Mutex::lock`] under the rule: recover, log once, clear the poison.
pub trait LockOrRecover<T: ?Sized> {
/// The guard, poisoned or not.
fn lock_or_recover(&self) -> MutexGuard<'_, T>;
}
impl<T: ?Sized> LockOrRecover<T> for Mutex<T> {
#[track_caller]
fn lock_or_recover(&self) -> MutexGuard<'_, T> {
match self.lock() {
Ok(g) => g,
Err(p) => {
poisoned("mutex");
self.clear_poison();
p.into_inner()
}
}
}
}
/// [`RwLock::read`] / [`RwLock::write`] under the rule.
pub trait RwLockOrRecover<T: ?Sized> {
/// The read guard, poisoned or not.
fn read_or_recover(&self) -> RwLockReadGuard<'_, T>;
/// The write guard, poisoned or not.
fn write_or_recover(&self) -> RwLockWriteGuard<'_, T>;
}
impl<T: ?Sized> RwLockOrRecover<T> for RwLock<T> {
#[track_caller]
fn read_or_recover(&self) -> RwLockReadGuard<'_, T> {
match self.read() {
Ok(g) => g,
Err(p) => {
poisoned("rwlock");
self.clear_poison();
p.into_inner()
}
}
}
#[track_caller]
fn write_or_recover(&self) -> RwLockWriteGuard<'_, T> {
match self.write() {
Ok(g) => g,
Err(p) => {
poisoned("rwlock");
self.clear_poison();
p.into_inner()
}
}
}
}
/// Any other [`LockResult`] (a `Condvar` wait) under the rule. The mutex the wait
/// returns through stays poisoned until the next `lock_or_recover` clears it.
#[track_caller]
pub fn recover<G>(r: LockResult<G>) -> G {
r.unwrap_or_else(|p: PoisonError<G>| {
poisoned("condvar wait");
p.into_inner()
})
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Arc;
/// A poisoned mutex is recovered with its data, and the poison is cleared so the
/// next caller does not meet it (and log it) again.
#[test]
fn a_poisoned_mutex_is_recovered_and_cleared() {
let m = Arc::new(Mutex::new(7u32));
let m2 = m.clone();
let _ = std::thread::spawn(move || {
let _g = m2.lock().unwrap();
panic!("poison it");
})
.join();
assert!(m.is_poisoned());
assert_eq!(*m.lock_or_recover(), 7);
assert!(!m.is_poisoned(), "the poison is cleared after one recovery");
}
#[test]
fn a_poisoned_rwlock_is_recovered_and_cleared() {
let l = Arc::new(RwLock::new(vec![1u8]));
let l2 = l.clone();
let _ = std::thread::spawn(move || {
let _g = l2.write().unwrap();
panic!("poison it");
})
.join();
assert!(l.is_poisoned());
l.write_or_recover().push(2);
assert!(!l.is_poisoned());
assert_eq!(*l.read_or_recover(), vec![1, 2]);
}
/// **The rule holds everywhere in `server/lib/src`**: no production line takes a
/// lock with `unwrap()`, `expect(..)`, a hand-written `into_inner` recovery or a
/// `map_err` refusal. Test items are stripped; they may unwrap freely.
#[test]
fn every_lock_in_server_lib_follows_the_one_rule() {
let src = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("src");
let mut stack = vec![src.clone()];
let mut bad = Vec::new();
let mut files = 0;
while let Some(dir) = stack.pop() {
for e in std::fs::read_dir(&dir).unwrap().flatten() {
let p = e.path();
if p.is_dir() {
if p.file_name().is_some_and(|n| n != "generated") {
stack.push(p);
}
continue;
}
if p.extension().is_none_or(|x| x != "rs")
|| p.file_name().is_some_and(|n| n == "tests.rs")
{
continue;
}
files += 1;
let rel = p.strip_prefix(&src).unwrap().display().to_string();
let text = std::fs::read_to_string(&p).unwrap();
// Join each statement onto one line so a call split over lines
// (`.lock()\n.unwrap_or_else(..)`) is still seen.
let prod = strip_tests(&text);
let flat: String = prod
.lines()
.map(|l| l.split("//").next().unwrap_or("").trim())
.collect::<Vec<_>>()
.join(" ");
if rel == "lock.rs" {
continue;
}
// Every std lock is taken through this module: a bare `.lock()`,
// `.read()` or `.write()` (not a tokio `.await`) is a site that
// chose its own answer to poison (unwrap, expect, map_err, `.ok()`,
// `if let Ok(..)`, a silent default).
for take in [".lock()", ".read()", ".write()"] {
let mut from = 0;
while let Some(at) = flat[from..].find(take) {
let at = from + at;
let after = flat[at + take.len()..].trim_start();
if !after.starts_with(".await") {
bad.push(format!("{rel}: …{}", &flat[at..floor(&flat, at + 90)]));
}
from = at + take.len();
}
}
// A `Condvar` wait hands back a `LockResult` too: it goes through `recover`.
for take in [
".wait(",
".wait_timeout(",
".wait_while(",
".wait_timeout_while(",
] {
let mut from = 0;
while let Some(at) = flat[from..].find(take) {
let at = from + at;
let before = &flat[floor(&flat, at.saturating_sub(60))..at];
if !before.contains("recover(") {
bad.push(format!("{rel}: …{}", &flat[at..floor(&flat, at + 90)]));
}
from = at + take.len();
}
}
}
}
assert!(files > 60, "the census read only {files} files");
assert!(
bad.is_empty(),
"{} lock site(s) break the one poison rule (use lock_or_recover / read_or_recover / write_or_recover / recover):\n {}",
bad.len(),
bad.join("\n ")
);
}
/// The largest char boundary of `s` at or below `at`.
fn floor(s: &str, at: usize) -> usize {
let mut at = at.min(s.len());
while !s.is_char_boundary(at) {
at -= 1;
}
at
}
fn strip_tests(text: &str) -> String {
let lines: Vec<&str> = text.lines().collect();
let (mut out, mut i) = (String::new(), 0usize);
while i < lines.len() {
let t = lines[i].trim_start();
if t.starts_with("#[cfg(test)]") || t.starts_with("#[cfg(all(test") {
let (mut depth, mut opened) = (0i32, false);
while i < lines.len() {
for c in lines[i].chars() {
match c {
'{' => {
depth += 1;
opened = true;
}
'}' => depth -= 1,
_ => {}
}
}
i += 1;
if (opened && depth <= 0) || (!opened && lines[i - 1].trim_end().ends_with(';'))
{
break;
}
}
continue;
}
out.push_str(lines[i]);
out.push('\n');
i += 1;
}
out
}
}