nullislabs · lgahdl · Jul 16, 2026
@@ -49,6 +49,12 @@ thiserror = "2"
 tokio = { version = "1", features = ["full"] }
 futures = "0.3"
 
+# Faster non-async Mutex/RwLock (no poisoning, one word instead of a
+# boxed OS primitive) for short critical sections that never hold the
+# guard across an `.await`. Already in the tree transitively; pinned
+# here so call sites can depend on it directly.
+parking_lot = "0.12"
+
 # Serde + config.
 serde = { version = "1", features = ["derive"] }
 serde_json = { version = "1", default-features = false, features = ["alloc"] }

@@ -35,6 +35,7 @@ tokio.workspace = true
 # Task lifecycle and graceful shutdown; the sole crate that raw-spawns
 # tokio tasks. Every engine task routes through its executor.
 nexum-tasks = { path = "../nexum-tasks" }
+parking_lot.workspace = true
 
 # Manifest parsing.
 serde.workspace = true

@@ -212,7 +212,7 @@ impl LogPipeline {
 
 #[cfg(test)]
 mod tests {
-    use std::sync::Mutex;
+    use parking_lot::Mutex;
 
     use super::*;
 
@@ -224,7 +224,7 @@ mod tests {
 
     impl RunLogStore for CountingStore {
         fn append(&self, record: LogRecord) {
-            self.appended.lock().unwrap().push(record);
+            self.appended.lock().push(record);
         }
         fn list_runs(&self, _module: &str) -> Vec<RunMeta> {
             Vec::new()
@@ -246,7 +246,7 @@ mod tests {
             Level::INFO,
             "hello".to_owned(),
         ));
-        let appended = store.appended.lock().unwrap();
+        let appended = store.appended.lock();
         assert_eq!(appended.len(), 1, "retention consumer saw the record");
         assert_eq!(appended[0].message, "hello");
         assert_eq!(appended[0].source, LogSource::HostInterface);

@@ -156,8 +156,7 @@ impl Drop for LineWriter {
 
 #[cfg(test)]
 mod tests {
-    use std::sync::Mutex;
-
+    use parking_lot::Mutex;
     use tokio::io::AsyncWriteExt;
 
     use super::*;
@@ -172,7 +171,7 @@ mod tests {
 
     impl RunLogStore for CaptureStore {
         fn append(&self, record: LogRecord) {
-            self.records.lock().unwrap().push(record);
+            self.records.lock().push(record);
         }
         fn list_runs(&self, _module: &str) -> Vec<crate::host::logs::RunMeta> {
             Vec::new()
@@ -198,7 +197,6 @@ mod tests {
         store
             .records
             .lock()
-            .unwrap()
             .iter()
             .map(|r| r.message.clone())
             .collect()
@@ -268,7 +266,7 @@ mod tests {
     async fn stderr_lines_carry_the_warn_level() {
         let (mut w, store) = setup(LogSource::Stderr);
         w.write_all(b"oops\n").await.unwrap();
-        let records = store.records.lock().unwrap();
+        let records = store.records.lock();
         assert_eq!(records[0].source, LogSource::Stderr);
         assert_eq!(records[0].level, Level::WARN);
     }

@@ -5,9 +5,10 @@
 use std::collections::HashMap;
 use std::collections::VecDeque;
 use std::sync::Arc;
-use std::sync::Mutex;
 use std::time::SystemTime;
 
+use parking_lot::Mutex;
+
 use super::{LogRecord, RunId};
 use crate::engine_config::LogRetentionLimits;
 
@@ -158,7 +159,7 @@ impl InMemoryRunLogStore {
 
 impl RunLogStore for InMemoryRunLogStore {
     fn append(&self, record: LogRecord) {
-        let mut inner = self.inner.lock().expect("log store mutex poisoned");
+        let mut inner = self.inner.lock();
         let limits = inner.limits;
         let module = record.run.module.clone();
         let entry = inner.modules.entry(module).or_insert_with(ModuleRuns::new);
@@ -183,7 +184,7 @@ impl RunLogStore for InMemoryRunLogStore {
     }
 
     fn list_runs(&self, module: &str) -> Vec<RunMeta> {
-        let inner = self.inner.lock().expect("log store mutex poisoned");
+        let inner = self.inner.lock();
         let Some(entry) = inner.modules.get(module) else {
             return Vec::new();
         };
@@ -195,7 +196,7 @@ impl RunLogStore for InMemoryRunLogStore {
     }
 
     fn read(&self, run: &RunId, cursor: u64) -> LogPage {
-        let inner = self.inner.lock().expect("log store mutex poisoned");
+        let inner = self.inner.lock();
         inner
             .modules
             .get(&*run.module)

@@ -3,13 +3,14 @@
 
 use std::collections::HashMap;
 use std::future::Future;
-use std::sync::{Arc, Mutex};
+use std::sync::Arc;
 use std::time::Duration;
 
 use alloy_chains::Chain;
 use alloy_rpc_types_eth::{Filter, Header, Log};
 use futures::StreamExt as _;
 use futures::channel::mpsc::{self, UnboundedSender};
+use parking_lot::Mutex;
 
 use crate::host::component::{ChainMethod, ChainProvider};
 use crate::host::provider_pool::{BlockStream, CanonicalLogStream, ProviderError};
@@ -211,8 +212,8 @@ impl MockChainProvider {
         self.lock().recorded.clone()
     }
 
-    fn lock(&self) -> std::sync::MutexGuard<'_, Inner> {
-        self.inner.lock().expect("mock chain mutex poisoned")
+    fn lock(&self) -> parking_lot::MutexGuard<'_, Inner> {
+        self.inner.lock()
     }
 }
 
@@ -223,7 +224,7 @@ impl ChainProvider for MockChainProvider {
     ) -> impl Future<Output = Result<BlockStream, ProviderError>> + Send {
         let inner = self.inner.clone();
         async move {
-            let stream: BlockStream = match inner.lock().expect("mock chain mutex").blocks.take() {
+            let stream: BlockStream = match inner.lock().blocks.take() {
                 Some(rx) => Box::pin(rx),
                 None => Box::pin(futures::stream::pending::<BlockItem>()),
             };
@@ -236,7 +237,7 @@ impl ChainProvider for MockChainProvider {
         _chain: Chain,
     ) -> impl Future<Output = Result<u64, ProviderError>> + Send {
         let inner = self.inner.clone();
-        async move { Ok(inner.lock().expect("mock chain mutex").head_block) }
+        async move { Ok(inner.lock().head_block) }
     }
 
     fn watch_chain_logs(
@@ -249,11 +250,10 @@ impl ChainProvider for MockChainProvider {
         // each into a single-log canonical batch so the poller-shaped
         // stream contract (`Vec<Log>` per block) is satisfied without
         // reworking every test that pushes logs one at a time.
-        let stream: CanonicalLogStream =
-            match self.inner.lock().expect("mock chain mutex").logs.take() {
-                Some(rx) => Box::pin(rx.map(|item| item.map(|log| vec![log]))),
-                None => Box::pin(futures::stream::pending::<Result<Vec<Log>, ProviderError>>()),
-            };
+        let stream: CanonicalLogStream = match self.inner.lock().logs.take() {
+            Some(rx) => Box::pin(rx.map(|item| item.map(|log| vec![log]))),
+            None => Box::pin(futures::stream::pending::<Result<Vec<Log>, ProviderError>>()),
+        };
         Ok(stream)
     }
 
@@ -266,11 +266,12 @@ impl ChainProvider for MockChainProvider {
         let inner = self.inner.clone();
         async move {
             // Record the call and take any one-shot park delay, then drop
-            // the guard before awaiting: a std `Mutex` must not be held
-            // across an await, and taking the delay here (not after the
-            // sleep) is what makes it survive a dropped future.
+            // the guard before awaiting: a `parking_lot::Mutex` guard is
+            // not `Send` and must not be held across an await, and taking
+            // the delay here (not after the sleep) is what makes it
+            // survive a dropped future.
             let delay = {
-                let mut guard = inner.lock().expect("mock chain mutex");
+                let mut guard = inner.lock();
                 guard.recorded.push(RecordedRequest {
                     chain,
                     method,
@@ -281,7 +282,7 @@ impl ChainProvider for MockChainProvider {
             if let Some(delay) = delay {
                 tokio::time::sleep(delay).await;
             }
-            let guard = inner.lock().expect("mock chain mutex");
+            let guard = inner.lock();
             let name = method.as_str();
             if let Some(body) = guard.exact.get(&(name, params_json)) {
                 Ok(body.clone())

@@ -1,8 +1,9 @@
 //! A manually-driven WASI clock for deterministic guest time in tests.
 
-use std::sync::{Arc, Mutex};
+use std::sync::Arc;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
+use parking_lot::Mutex;
 use wasmtime_wasi::{HostMonotonicClock, HostWallClock};
 
 use crate::supervisor::WasiClockOverride;
@@ -71,8 +72,8 @@ impl ManualClock {
         WasiClockOverride::new(Arc::new(self.clone()), Arc::new(self.clone()))
     }
 
-    fn locked(&self) -> std::sync::MutexGuard<'_, Instant> {
-        self.inner.lock().unwrap_or_else(|e| e.into_inner())
+    fn locked(&self) -> parking_lot::MutexGuard<'_, Instant> {
+        self.inner.lock()
     }
 }
 

@@ -4,7 +4,9 @@
 #![allow(clippy::result_large_err)]
 
 use std::collections::HashMap;
-use std::sync::{Arc, Mutex};
+use std::sync::Arc;
+
+use parking_lot::Mutex;
 
 use crate::host::component::{StateHandle, StateStore};
 use crate::host::local_store_redb::StorageError;
@@ -54,8 +56,8 @@ impl StateStore for MockStateStore {
 }
 
 impl MockStateHandle {
-    fn lock(&self) -> std::sync::MutexGuard<'_, Namespaces> {
-        self.namespaces.lock().expect("mock store mutex poisoned")
+    fn lock(&self) -> parking_lot::MutexGuard<'_, Namespaces> {
+        self.namespaces.lock()
     }
 }
 

@@ -39,6 +39,7 @@ tracing-core.workspace = true
 
 [dev-dependencies]
 proptest.workspace = true
+parking_lot.workspace = true
 # Dev-dependencies are excluded from Cargo's dependency-cycle check, so
 # the nexum-sdk -> shepherd-sdk -> shepherd-sdk-test dev-dep chain is a
 # normal, supported arrangement: the keeper stores acceptance-test

@@ -178,7 +178,7 @@ impl Visit for LineVisitor {
 
 #[cfg(test)]
 mod tests {
-    use std::sync::Mutex;
+    use parking_lot::Mutex;
 
     use super::*;
 
@@ -190,15 +190,15 @@ mod tests {
 
     impl LogSink for Arc<Captured> {
         fn log(&self, level: Level, message: &str) {
-            self.lines.lock().unwrap().push((level, message.to_owned()));
+            self.lines.lock().push((level, message.to_owned()));
         }
     }
 
     fn capture(f: impl FnOnce()) -> Vec<(Level, String)> {
         let sink = Arc::new(Captured::default());
         let subscriber = subscriber(Arc::clone(&sink));
         tracing::subscriber::with_default(subscriber, f);
-        sink.lines.lock().unwrap().clone()
+        sink.lines.lock().clone()
     }
 
     #[test]