diff --git a/Cargo.lock b/Cargo.lock index 93ac4ee1..45ba1549 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3566,6 +3566,7 @@ dependencies = [ "metrics-exporter-prometheus", "nexum-runtime", "nexum-tasks", + "parking_lot", "redb", "serde", "serde_json", @@ -3592,6 +3593,7 @@ dependencies = [ "alloy-rpc-types-eth", "alloy-sol-types", "http", + "parking_lot", "proptest", "serde_json", "shepherd-sdk-test", diff --git a/Cargo.toml b/Cargo.toml index 516a45ac..6ece3443 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -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"] } diff --git a/crates/nexum-runtime/Cargo.toml b/crates/nexum-runtime/Cargo.toml index a4ddab3e..3ab15ce7 100644 --- a/crates/nexum-runtime/Cargo.toml +++ b/crates/nexum-runtime/Cargo.toml @@ -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 diff --git a/crates/nexum-runtime/src/host/logs/mod.rs b/crates/nexum-runtime/src/host/logs/mod.rs index 74e0760f..e8e47d2a 100644 --- a/crates/nexum-runtime/src/host/logs/mod.rs +++ b/crates/nexum-runtime/src/host/logs/mod.rs @@ -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 { 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); diff --git a/crates/nexum-runtime/src/host/logs/stdio.rs b/crates/nexum-runtime/src/host/logs/stdio.rs index 251e6753..0fcaad29 100644 --- a/crates/nexum-runtime/src/host/logs/stdio.rs +++ b/crates/nexum-runtime/src/host/logs/stdio.rs @@ -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 { 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); } diff --git a/crates/nexum-runtime/src/host/logs/store.rs b/crates/nexum-runtime/src/host/logs/store.rs index e953d01e..6bcd9697 100644 --- a/crates/nexum-runtime/src/host/logs/store.rs +++ b/crates/nexum-runtime/src/host/logs/store.rs @@ -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 { - 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) diff --git a/crates/nexum-runtime/src/test_utils/chain.rs b/crates/nexum-runtime/src/test_utils/chain.rs index a1deeeae..bc9f26d4 100644 --- a/crates/nexum-runtime/src/test_utils/chain.rs +++ b/crates/nexum-runtime/src/test_utils/chain.rs @@ -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> + 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::()), }; @@ -236,7 +237,7 @@ impl ChainProvider for MockChainProvider { _chain: Chain, ) -> impl Future> + 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` 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::, 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::, 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()) diff --git a/crates/nexum-runtime/src/test_utils/clock.rs b/crates/nexum-runtime/src/test_utils/clock.rs index fcda4919..a50eac9e 100644 --- a/crates/nexum-runtime/src/test_utils/clock.rs +++ b/crates/nexum-runtime/src/test_utils/clock.rs @@ -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() } } diff --git a/crates/nexum-runtime/src/test_utils/store.rs b/crates/nexum-runtime/src/test_utils/store.rs index 7d8be449..e2e91592 100644 --- a/crates/nexum-runtime/src/test_utils/store.rs +++ b/crates/nexum-runtime/src/test_utils/store.rs @@ -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() } } diff --git a/crates/nexum-sdk/Cargo.toml b/crates/nexum-sdk/Cargo.toml index 8410babb..4fbe9f8e 100644 --- a/crates/nexum-sdk/Cargo.toml +++ b/crates/nexum-sdk/Cargo.toml @@ -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 diff --git a/crates/nexum-sdk/src/tracing.rs b/crates/nexum-sdk/src/tracing.rs index 81d1bb43..5f68284f 100644 --- a/crates/nexum-sdk/src/tracing.rs +++ b/crates/nexum-sdk/src/tracing.rs @@ -178,7 +178,7 @@ impl Visit for LineVisitor { #[cfg(test)] mod tests { - use std::sync::Mutex; + use parking_lot::Mutex; use super::*; @@ -190,7 +190,7 @@ mod tests { impl LogSink for Arc { fn log(&self, level: Level, message: &str) { - self.lines.lock().unwrap().push((level, message.to_owned())); + self.lines.lock().push((level, message.to_owned())); } } @@ -198,7 +198,7 @@ mod tests { 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]