scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-10-01 02:04:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Share one poison-tolerant lock helper from fabro-util

Browse source 
Seven crates' files each carried the same four-line lock function that
recovers a poisoned mutex. fabro_util::sync::lock is that function, once;
the copies in fabro-petri and fabro-server are gone. fabro-template's
helper panics on poison instead, a different policy, and is left as is.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-09-20 14:47:46 -04:00
parent 6d78cc4f47
commit 1f6e53c949
No known key found for this signature in database
9 changed files with 81 additions and 83 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
lib/apps/fabro-server/src/petri_runs.rs
View file

@ -14,12 +14,13 @@
//! `RunOptions::run_key`.
use std::collections::HashMap;
use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
use std::sync::{Arc, Mutex};
use fabro_db::DbPool;
use fabro_petri::SqliteRunStore;
use fabro_petri::petri::{Access, OwnerId, RunKey, RunLogs, RunStore as _, StoreError};
use fabro_types::RunId;
use fabro_util::sync;
use tracing::debug;
pub(crate) struct PetriRuns {
@ -66,7 +67,7 @@ impl PetriRuns {
) -> Result<Arc<dyn RunLogs>, StoreError> {
let handle = self.store.open(&Self::key(&run_id), access.clone()).await?;
if let Some(owner) = access.owner() {
lock(&self.handles).insert((run_id, owner.clone()), Arc::clone(&handle));
sync::lock(&self.handles).insert((run_id, owner.clone()), Arc::clone(&handle));
}
Ok(handle)
}
@ -80,7 +81,7 @@ impl PetriRuns {
run_id: RunId,
owner: &OwnerId,
) -> Result<Arc<dyn RunLogs>, StoreError> {
if let Some(handle) = lock(&self.handles).get(&(run_id, owner.clone())) {
if let Some(handle) = sync::lock(&self.handles).get(&(run_id, owner.clone())) {
return Ok(Arc::clone(handle));
}
let holder = self.store.owner(&Self::key(&run_id)).await?;
@ -107,7 +108,7 @@ impl PetriRuns {
/// Drop the handle `owner` holds on the run: the worker's own release.
/// The store ends the lease when this was the owner's last handle.
pub(crate) fn release(&self, run_id: RunId, owner: &OwnerId) {
let handle = lock(&self.handles).remove(&(run_id, owner.clone()));
let handle = sync::lock(&self.handles).remove(&(run_id, owner.clone()));
debug!(
run_id = %run_id,
owner = %owner,
@ -132,7 +133,7 @@ impl PetriRuns {
/// releasing does not keep the lease.
pub(crate) fn worker_exited(&self, run_id: RunId) {
let dropped = {
let mut handles = lock(&self.handles);
let mut handles = sync::lock(&self.handles);
let owners: Vec<_> = handles
.keys()
.filter(|(held, _)| *held == run_id)
@ -154,10 +155,6 @@ impl PetriRuns {
}
}
fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
mutex.lock().unwrap_or_else(PoisonError::into_inner)
}
#[cfg(test)]
mod tests {
use std::collections::BTreeMap;
@ -226,14 +223,14 @@ mod tests {
}
fn launched_mode(&self) -> Option<&'static str> {
*lock(&self.mode)
*sync::lock(&self.mode)
}
}
#[async_trait::async_trait]
impl WorkerRuntime for HeldWorkerRuntime {
async fn start(&self, spec: WorkerLaunchSpec) -> anyhow::Result<StartedWorker> {
*lock(&self.mode) = Some(spec.mode);
*sync::lock(&self.mode) = Some(spec.mode);
self.running.store(true, Ordering::SeqCst);
let exit = Arc::clone(&self.exit);
let stderr: Pin<Box<dyn AsyncRead + Send + 'static>> = Box::pin(tokio::io::empty());

53
lib/components/fabro-petri/src/hooks.rs
View file

@ -73,7 +73,7 @@
use std::collections::{BTreeMap, HashMap, HashSet};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, MutexGuard, OnceLock, PoisonError};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::Duration;
use fabro_checkpoint::author::GitAuthor;
@ -86,6 +86,7 @@ use fabro_types::{
BlobHash, DiffSummary, GitIdentity, GitIdentitySource, RunId, SandboxProviderKind,
};
use fabro_util::error::collect_chain;
use fabro_util::sync;
use fabro_util::workspace_glob::{WorkspaceGlobError, WorkspaceGlobSet};
use petri_execution::{CancelReason, CoordinatorHandle, InvocationId, RunKey, RunStore};
use petri_runtime::driver::lifecycle::{
@ -315,7 +316,7 @@ impl FabroHooks {
/// engine reports the run failed with.
#[must_use]
pub fn checkpoint_failure(&self) -> Option<String> {
lock(&self.failure).clone()
sync::lock(&self.failure).clone()
}
/// The run's workspaces on this host, as the hooks reach them.
@ -327,14 +328,14 @@ impl FabroHooks {
/// The lock that serializes Git work in one workspace.
fn workspace_lock(&self, workspace: &str) -> Arc<AsyncMutex<()>> {
Arc::clone(
lock(&self.workspace_locks)
sync::lock(&self.workspace_locks)
.entry(workspace.to_string())
.or_default(),
)
}
fn fail_run(&self, message: &str) {
let mut failure = lock(&self.failure);
let mut failure = sync::lock(&self.failure);
if failure.is_none() {
*failure = Some(message.to_string());
}
@ -358,7 +359,9 @@ impl FabroHooks {
if self.workspaces.workspace_exists(&isolated).await {
return Ok(isolated);
}
let cached = lock(&self.inherited).get(&context.invocation).cloned();
let cached = sync::lock(&self.inherited)
.get(&context.invocation)
.cloned();
let inherited = if let Some(inherited) = cached {
inherited
} else {
@ -373,7 +376,7 @@ impl FabroHooks {
collect_chain(&error).join(": ")
)
})?;
lock(&self.inherited).insert(context.invocation, inherited.clone());
sync::lock(&self.inherited).insert(context.invocation, inherited.clone());
inherited
};
Ok(inherited.unwrap_or(isolated))
@ -382,7 +385,9 @@ impl FabroHooks {
/// The environment of `scope` in the context's execution, as
/// `scope_acquired` kept it, with the workspace id the executor named.
fn env_of(&self, context: &HookContext, scope: ScopeId) -> Option<AcquiredEnv> {
lock(&self.envs).get(&(context.execution, scope)).cloned()
sync::lock(&self.envs)
.get(&(context.execution, scope))
.cloned()
}
/// The checkpoint commit for one attempt's result. `Ok(Some)` is the
@ -532,7 +537,7 @@ impl FabroHooks {
/// Remember a commit this process made, and record the run branch when
/// this commit created it.
async fn committed(&self, key: CheckpointKey, workspace: &str, snapshot: &Snapshot) {
lock(&self.committed).insert(key, (workspace.to_string(), snapshot.sha.clone()));
sync::lock(&self.committed).insert(key, (workspace.to_string(), snapshot.sha.clone()));
let Some(branched) = &snapshot.branched else {
return;
};
@ -668,7 +673,7 @@ impl FabroHooks {
/// seeded.
async fn restore_host(&self, workspace: &str) -> Result<(), ScopeAcquiredError> {
let targets = self.restore_targets().await?;
let target = lock(targets).remove(workspace);
let target = sync::lock(targets).remove(workspace);
let Some(target) = target else {
return Ok(());
};
@ -700,7 +705,7 @@ impl FabroHooks {
env: &Arc<dyn ExecEnv>,
) -> Result<(), ScopeAcquiredError> {
let targets = self.restore_targets().await?;
let target = lock(targets).remove(workspace);
let target = sync::lock(targets).remove(workspace);
let Some(target) = target else {
return Ok(());
};
@ -735,10 +740,10 @@ impl FabroHooks {
self.recorded_loaded
.get_or_try_init(|| self.load_recorded())
.await?;
if lock(&self.recorded).contains(&key) {
if sync::lock(&self.recorded).contains(&key) {
return Ok(());
}
let committed = lock(&self.committed).get(&key).cloned();
let committed = sync::lock(&self.committed).get(&key).cloned();
let (workspace, sha) = if let Some(committed) = committed {
committed
} else {
@ -806,8 +811,8 @@ impl FabroHooks {
collect_chain(&error).join(": ")
)
})?;
lock(&self.recorded).insert(key);
*lock(&self.last_checkpoint) = Some((workspace, sha));
sync::lock(&self.recorded).insert(key);
*sync::lock(&self.last_checkpoint) = Some((workspace, sha));
Ok(())
}
@ -868,7 +873,7 @@ impl FabroHooks {
collect_chain(&error).join(": ")
)
})?;
let mut recorded = lock(&self.recorded);
let mut recorded = sync::lock(&self.recorded);
let mut last = None;
for record in stored {
let PlatformRecord::Checkpoint(checkpoint) = &record.record else {
@ -888,7 +893,7 @@ impl FabroHooks {
}
}
drop(recorded);
let mut last_checkpoint = lock(&self.last_checkpoint);
let mut last_checkpoint = sync::lock(&self.last_checkpoint);
if last_checkpoint.is_none() {
*last_checkpoint = last;
}
@ -941,7 +946,7 @@ impl FabroHooks {
};
let digest = BlobHash::new(&bytes);
let identity = (path.clone(), digest.to_string());
if lock(&self.collected).contains(&identity) {
if sync::lock(&self.collected).contains(&identity) {
continue;
}
let blob = blobs
@ -974,7 +979,7 @@ impl FabroHooks {
collect_chain(&error).join(": ")
)
})?;
lock(&self.collected).insert(identity);
sync::lock(&self.collected).insert(identity);
total_bytes = total_bytes.saturating_add(size);
collected += 1;
}
@ -994,7 +999,7 @@ impl FabroHooks {
collect_chain(&error).join(": ")
)
})?;
let mut collected = lock(&self.collected);
let mut collected = sync::lock(&self.collected);
for record in stored {
if let PlatformRecord::ArtifactCollected(artifact) = record.record {
collected.insert((artifact.path, artifact.digest));
@ -1022,7 +1027,7 @@ impl FabroHooks {
let Some(base_sha) = branch.base_sha.clone() else {
return Ok(());
};
let last = lock(&self.last_checkpoint).clone();
let last = sync::lock(&self.last_checkpoint).clone();
let Some((workspace, head_sha)) = last else {
debug!(run_id = %self.run_id, "no checkpoint is recorded; no run diff");
return Ok(());
@ -1149,10 +1154,6 @@ fn select_artifacts(mut candidates: Vec<(String, u64)>) -> Vec<(String, u64)> {
selected
}
fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
mutex.lock().unwrap_or_else(PoisonError::into_inner)
}
fn is_checkpoint_failure(status: &Status) -> bool {
matches!(status, Status::Failure(info) if info.class.as_str() == CHECKPOINT_FAILED_CLASS)
}
@ -1297,7 +1298,7 @@ impl ExecutionHooks for FabroHooks {
);
let scope = released.scope;
let notes = self.inner.scope_released(context, released).await;
lock(&self.envs).remove(&(context.execution, scope));
sync::lock(&self.envs).remove(&(context.execution, scope));
notes
}
@ -1308,7 +1309,7 @@ impl ExecutionHooks for FabroHooks {
) -> Result<(), ScopeAcquiredError> {
self.inner.scope_acquired(context, acquired.clone()).await?;
let workspace = acquired.workspace.as_str().to_owned();
lock(&self.envs).insert(
sync::lock(&self.envs).insert(
(context.execution, acquired.scope),
(workspace.clone(), Arc::clone(&acquired.env)),
);

15
lib/components/fabro-petri/src/http_store.rs
View file

@ -62,13 +62,14 @@
use std::collections::HashMap;
use std::future::Future;
use std::sync::{Arc, Mutex, MutexGuard, PoisonError, Weak};
use std::sync::{Arc, Mutex, Weak};
use std::time::Duration;
use std::{fmt, mem, ptr};
use fabro_api::types::{PetriAccess, PetriAppendRequest, PetriOpenRequest, PetriRecord};
use fabro_client::{Client, api_failure_for};
use fabro_types::{BlobHash, RunId};
use fabro_util::sync;
use petri_store::{Access, Digest, LogId, OwnerId, Record, RunKey, RunLogs, RunStore, StoreError};
use serde_json::Value;
use tokio::runtime::Handle;
@ -148,7 +149,7 @@ impl HttpRunStore {
owner: OwnerId,
locator: String,
) -> Arc<HttpRunLogs> {
let mut live = lock(&self.shared.live);
let mut live = sync::lock(&self.shared.live);
let slot = (key.clone(), owner.clone());
if let Some(handle) = live.get(&slot).and_then(Weak::upgrade) {
return handle;
@ -185,7 +186,7 @@ impl Shared {
/// Await every release a dropped handle spawned, so what follows sees
/// the lease as the drops left it.
async fn drain_releases(&self) {
let pending = mem::take(&mut *lock(&self.releases));
let pending = mem::take(&mut *sync::lock(&self.releases));
for release in pending {
// A release task never panics: it reports its own failure.
let _ = release.await;
@ -436,7 +437,7 @@ impl Drop for HttpRunLogs {
return;
};
{
let mut live = lock(&self.shared.live);
let mut live = sync::lock(&self.shared.live);
let slot = (self.key.clone(), owner.clone());
let this: *const Self = self;
if live
@ -454,7 +455,7 @@ impl Drop for HttpRunLogs {
let release = runtime.spawn(async move {
shared.release(&key, run_id, &owner).await;
});
lock(&self.shared.releases).push(release);
sync::lock(&self.shared.releases).push(release);
}
Err(_) => {
warn!(
@ -551,10 +552,6 @@ impl RunLogs for HttpRunLogs {
}
}
fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
mutex.lock().unwrap_or_else(PoisonError::into_inner)
}
#[cfg(test)]
mod tests {
use fabro_client::ApiFailure;

15
lib/components/fabro-petri/src/projector.rs
View file

@ -57,7 +57,7 @@ mod cache;
use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use fabro_db::DbPool;
@ -65,6 +65,7 @@ use fabro_store::platform_records::{PlatformRecordStore, StoredPlatformRecord, n
use fabro_store::{RunProjection, RunSummaryStore};
use fabro_types::{RunId, RunStreamItem, RunStreamItemKind};
use fabro_util::error::collect_chain;
use fabro_util::sync;
use petri_execution::events::{self, EventId, EventSource, RunEvent};
use petri_execution::{Access, CoordinatorEvent, RunKey, RunStore as _, inspect};
use petri_runtime::engine::Event;
@ -270,7 +271,7 @@ impl Projector {
.map_err(ProjectError::Database)?;
}
records.commit().await.map_err(ProjectError::Database)?;
lock(&self.slots).remove(&run_id);
sync::lock(&self.slots).remove(&run_id);
Ok(())
}
@ -290,7 +291,7 @@ impl Projector {
/// more when it ends; any number of signals in between coalesce.
pub fn signal(self: &Arc<Self>, run_id: RunId) {
{
let mut slots = lock(&self.slots);
let mut slots = sync::lock(&self.slots);
let slot = slots.entry(run_id).or_default();
if slot.running {
slot.pending = true;
@ -312,7 +313,7 @@ impl Projector {
false
}
};
let mut slots = lock(&projector.slots);
let mut slots = sync::lock(&projector.slots);
let slot = slots.entry(run_id).or_default();
if again || slot.pending {
slot.pending = false;
@ -329,7 +330,7 @@ impl Projector {
pub async fn settle(&self, run_id: RunId) {
loop {
let idle = {
let slots = lock(&self.slots);
let slots = sync::lock(&self.slots);
slots
.get(&run_id)
.is_none_or(|slot| !slot.running && !slot.pending)
@ -1059,10 +1060,6 @@ fn column(value: u64) -> i64 {
i64::try_from(value).unwrap_or(i64::MAX)
}
fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
mutex.lock().unwrap_or_else(PoisonError::into_inner)
}
/// The run's projection rebuilt from its records alone, with nothing
/// stored: what a fresh projector would commit over the same records. A test
/// compares it with the live view. `records` and `views` are the two pools

13
lib/components/fabro-petri/src/projector/cache.rs
View file

@ -15,10 +15,11 @@
//! rows: the rebuild test in `tests/projection.rs` compares the two.
use std::collections::HashMap;
use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use fabro_types::RunId;
use fabro_util::sync;
use petri_execution::events::{RunEvent, RunReplay};
use tokio::sync::Mutex as AsyncMutex;
@ -86,7 +87,7 @@ impl Caches {
/// The run's pass lock, holding its cache if one is kept; the run counts
/// as used now.
pub(super) fn pass_of(&self, run_id: RunId) -> Arc<AsyncMutex<Option<RunCache>>> {
let mut runs = lock(&self.runs);
let mut runs = sync::lock(&self.runs);
let entry = runs.entry(run_id).or_insert_with(|| Entry {
pass: Arc::default(),
touched: Instant::now(),
@ -99,7 +100,7 @@ impl Caches {
/// with no cache and no pass under way. A run whose pass is running is
/// in use and left alone. How many caches were dropped.
pub(crate) fn sweep(&self, idle: Duration) -> usize {
let mut runs = lock(&self.runs);
let mut runs = sync::lock(&self.runs);
let mut dropped = 0;
runs.retain(|_, entry| {
if entry.touched.elapsed() < idle {
@ -121,12 +122,8 @@ impl Caches {
/// Whether a cache is kept for the run: a test's view of the cache.
pub(crate) fn holds(&self, run_id: RunId) -> bool {
let runs = lock(&self.runs);
let runs = sync::lock(&self.runs);
runs.get(&run_id)
.is_some_and(|entry| entry.pass.try_lock().is_ok_and(|cache| cache.is_some()))
}
}
fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
mutex.lock().unwrap_or_else(PoisonError::into_inner)
}

17
lib/components/fabro-petri/src/run_store.rs
View file

@ -55,13 +55,14 @@
use std::collections::HashMap;
use std::error::Error;
use std::sync::{Arc, Mutex, MutexGuard, PoisonError, Weak};
use std::sync::{Arc, Mutex, Weak};
use std::time::{SystemTime, UNIX_EPOCH};
use std::{fmt, mem, ptr};
use fabro_db::DbPool;
use fabro_store::BlobStore;
use fabro_types::BlobHash;
use fabro_util::sync;
use petri_store::{
Access, Digest, ExecutionId, LogId, OwnerId, Record, RunKey, RunLogs, RunStore, StoreError,
};
@ -150,7 +151,7 @@ impl SqliteRunStore {
if result.rows_affected() == 0 {
return Err(self.shared.not_found(key));
}
lock(&self.shared.live).remove(key);
sync::lock(&self.shared.live).remove(key);
debug!(run_id = %key, "Petri run lease released from outside");
Ok(())
}
@ -174,7 +175,7 @@ impl SqliteRunStore {
/// The writer handle for `owner`, once the lease is taken: the live one
/// when this owner already holds a handle here, else a new one.
fn writer(&self, key: &RunKey, owner: OwnerId) -> Arc<dyn RunLogs> {
let mut live = lock(&self.shared.live);
let mut live = sync::lock(&self.shared.live);
if let Some(handle) = live.get(key).and_then(Weak::upgrade) {
if handle.owner.as_ref() == Some(&owner) {
return handle;
@ -214,7 +215,7 @@ impl Shared {
/// Await every release a dropped handle spawned, so what follows sees
/// the lease as the drops left it.
async fn drain_releases(&self) {
let pending = mem::take(&mut *lock(&self.releases));
let pending = mem::take(&mut *sync::lock(&self.releases));
for release in pending {
// A release task never panics: it reports its own failure.
let _ = release.await;
@ -409,7 +410,7 @@ impl Drop for SqliteRunLogs {
return;
};
{
let mut live = lock(&self.shared.live);
let mut live = sync::lock(&self.shared.live);
let this: *const Self = self;
if live
.get(&self.key)
@ -425,7 +426,7 @@ impl Drop for SqliteRunLogs {
let release = runtime.spawn(async move {
shared.release_owner(&key, &owner).await;
});
lock(&self.shared.releases).push(release);
sync::lock(&self.shared.releases).push(release);
}
Err(_) => {
warn!(
@ -558,10 +559,6 @@ impl RunLogs for SqliteRunLogs {
}
}
fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
mutex.lock().unwrap_or_else(PoisonError::into_inner)
}
/// Milliseconds since the Unix epoch, as SQLite stores them.
fn now_ms() -> i64 {
SystemTime::now()

20
lib/components/fabro-petri/src/test_support.rs
View file

@ -5,7 +5,7 @@
//! dev-dependency turns on.
use std::collections::HashMap;
use std::sync::{Mutex, MutexGuard, PoisonError};
use std::sync::Mutex;
use std::time::Duration;
use async_trait::async_trait;
@ -13,6 +13,7 @@ use bytes::Bytes;
use fabro_store::platform_records::now_ms;
use fabro_store::{PlatformRecord, PlatformRecordKind, StagePosition, StoredPlatformRecord};
use fabro_types::{BlobHash, RunId};
use fabro_util::sync;
pub use petri_testkit::run_store;
use crate::blobs::Blobs;
@ -47,7 +48,7 @@ impl MemoryBlobs {
/// How many blobs the table holds.
#[must_use]
pub fn len(&self) -> usize {
lock(&self.rows).len()
sync::lock(&self.rows).len()
}
#[must_use]
@ -60,12 +61,12 @@ impl MemoryBlobs {
impl Blobs for MemoryBlobs {
async fn write(&self, bytes: &[u8]) -> anyhow::Result<BlobHash> {
let hash = BlobHash::new(bytes);
lock(&self.rows).insert(hash, bytes.to_vec());
sync::lock(&self.rows).insert(hash, bytes.to_vec());
Ok(hash)
}
async fn read(&self, hash: &BlobHash) -> anyhow::Result<Option<Bytes>> {
Ok(lock(&self.rows)
Ok(sync::lock(&self.rows)
.get(hash)
.map(|bytes| Bytes::copy_from_slice(bytes)))
}
@ -86,14 +87,13 @@ impl MemoryPlatformRecords {
/// Every record of the run, in seq order.
#[must_use]
pub fn records(&self, run_id: &RunId) -> Vec<StoredPlatformRecord> {
lock(&self.runs).get(run_id).cloned().unwrap_or_default()
sync::lock(&self.runs)
.get(run_id)
.cloned()
.unwrap_or_default()
}
}
fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
mutex.lock().unwrap_or_else(PoisonError::into_inner)
}
#[async_trait]
impl PlatformRecords for MemoryPlatformRecords {
async fn append(
@ -102,7 +102,7 @@ impl PlatformRecords for MemoryPlatformRecords {
record: &PlatformRecord,
position: Option<StagePosition>,
) -> Result<StoredPlatformRecord, PlatformRecordError> {
let mut runs = lock(&self.runs);
let mut runs = sync::lock(&self.runs);
let records = runs.entry(*run_id).or_default();
let stored = StoredPlatformRecord {
seq: records.len() as u64 + 1,

1
lib/foundation/fabro-util/src/lib.rs
View file

@ -12,6 +12,7 @@ pub mod printer;
pub mod run_log;
pub mod session_secret;
pub mod shell;
pub mod sync;
pub mod terminal;
pub mod text;
pub mod time;

11
lib/foundation/fabro-util/src/sync.rs Normal file
View file

@ -0,0 +1,11 @@
//! Locking helpers shared across crates.
use std::sync::{Mutex, MutexGuard, PoisonError};
/// Lock a mutex, recovering the guard when another holder panicked. The
/// state such a mutex guards is bookkeeping (a cache, a set of ids, a
/// counter) that stays usable after a panic elsewhere, so the poison is
/// cleared rather than propagated.
pub fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
mutex.lock().unwrap_or_else(PoisonError::into_inner)
}