Skip to content

Replace parking_lot Mutex with tokio Mutex to align with kona-node async primitives #433

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Replace parking_lot Mutex with tokio Mutex to align with kona-node async primitives #433

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 7 additions & 2 deletions crates/rollup-boost/src/client/rpc.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -474,7 +474,7 @@ pub mod tests {
use assert_cmd::Command;
use http::Uri;
use jsonrpsee::core::client::ClientT;
use parking_lot::Mutex;
use std::sync::Mutex;
Copy link
Contributor

@theochap theochap Nov 7, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: We should tokio mutexes here.

Suggested change
use std::sync::Mutex;
use tokio::sync::Mutex;


use crate::payload::PayloadSource;
use alloy_rpc_types_engine::JwtSecret;
Expand All @@ -499,7 +499,12 @@ pub mod tests {
LazyLock::new(|| Mutex::new(HashSet::new()));
loop {
let port: u16 = rand::random_range(1000..20000);
if TcpListener::bind(("127.0.0.1", port)).is_ok() && CLAIMED_PORTS.lock().insert(port) {
if TcpListener::bind(("127.0.0.1", port)).is_ok()
&& CLAIMED_PORTS
.lock()
.expect("CLAIMED_PORTS poisoned")
.insert(port)
{
return port;
}
}
Expand Down
16 changes: 8 additions & 8 deletions crates/rollup-boost/src/debug_api.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,9 @@ use jsonrpsee::core::{RpcResult, async_trait};
use jsonrpsee::http_client::HttpClient;
use jsonrpsee::proc_macros::rpc;
use jsonrpsee::server::Server;
use parking_lot::Mutex;
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use tokio::sync::Mutex;

use crate::update_execution_mode_gauge;

Expand Down Expand Up @@ -81,7 +81,7 @@ impl DebugServer {
let server = Server::builder().build(debug_addr).await?;

// Register the initial execution mode metric
let current_mode = self.execution_mode();
let current_mode = self.execution_mode().await;
update_execution_mode_gauge(current_mode);

let handle = server.start(self.into_rpc());
Expand All @@ -95,12 +95,12 @@ impl DebugServer {
Ok(())
}

pub fn execution_mode(&self) -> ExecutionMode {
*self.execution_mode.lock()
pub async fn execution_mode(&self) -> ExecutionMode {
*self.execution_mode.lock().await
}

pub fn set_execution_mode(&self, mode: ExecutionMode) {
*self.execution_mode.lock() = mode;
pub async fn set_execution_mode(&self, mode: ExecutionMode) {
*self.execution_mode.lock().await = mode;
update_execution_mode_gauge(mode);
}
}
Expand All @@ -111,7 +111,7 @@ impl DebugApiServer for DebugServer {
&self,
request: SetExecutionModeRequest,
) -> RpcResult<SetExecutionModeResponse> {
self.set_execution_mode(request.execution_mode);
self.set_execution_mode(request.execution_mode).await;

tracing::info!("Set execution mode to {:?}", request.execution_mode);

Expand All @@ -122,7 +122,7 @@ impl DebugApiServer for DebugServer {

async fn get_execution_mode(&self) -> RpcResult<GetExecutionModeResponse> {
Ok(GetExecutionModeResponse {
execution_mode: self.execution_mode(),
execution_mode: self.execution_mode().await,
})
}
}
Expand Down
16 changes: 14 additions & 2 deletions crates/rollup-boost/src/flashblocks/service.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -422,7 +422,13 @@ mod tests {
.await?;

let get_payload_requests_builder = builder_mock.get_payload_requests.clone();
assert_eq!(get_payload_requests_builder.lock().len(), 1);
assert_eq!(
get_payload_requests_builder
.lock()
.expect("get_payload_requests mutex poisoned")
.len(),
1
);

Ok(())
}
Expand Down Expand Up @@ -455,7 +461,13 @@ mod tests {
.await?;

let get_payload_requests_builder = builder_mock.get_payload_requests.clone();
assert_eq!(get_payload_requests_builder.lock().len(), 1);
assert_eq!(
get_payload_requests_builder
.lock()
.expect("get_payload_requests mutex poisoned")
.len(),
1
);

Ok(())
}
Expand Down
34 changes: 17 additions & 17 deletions crates/rollup-boost/src/health.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ use std::{
};

use alloy_rpc_types_eth::BlockNumberOrTag;
use parking_lot::Mutex;
use tokio::sync::Mutex;
use tokio::{
task::JoinHandle,
time::{Instant, sleep_until},
Expand Down Expand Up @@ -63,26 +63,26 @@ impl HealthHandle {
.gt(&self.max_unsafe_interval)
{
warn!(target: "rollup_boost::health", curr_unix = %t, unsafe_unix = %block.header.timestamp, "L2 client - unsafe block timestamp is too old, updating health status to ServiceUnavailable");
self.probes.set_health(Health::ServiceUnavailable);
self.probes.set_health(Health::ServiceUnavailable).await;
sleep_until(Instant::now() + self.health_check_interval).await;
continue;
} else if self.execution_mode.lock().is_disabled()
|| self.execution_mode.lock().is_dry_run()
} else if self.execution_mode.lock().await.is_disabled()
|| self.execution_mode.lock().await.is_dry_run()
{
self.probes.set_health(Health::Healthy);
self.probes.set_health(Health::Healthy).await;
sleep_until(Instant::now() + self.health_check_interval).await;
continue;
}
}
Err(e) => {
warn!(target: "rollup_boost::health", "L2 client - Failed to get unsafe block {} - updating health status", e);
self.probes.set_health(Health::ServiceUnavailable);
self.probes.set_health(Health::ServiceUnavailable).await;
sleep_until(Instant::now() + self.health_check_interval).await;
continue;
}
};

if self.execution_mode.lock().is_enabled() {
if self.execution_mode.lock().await.is_enabled() {
// Only check builder client health if execution mode is enabled
// If its unhealthy, set the health status to PartialContent
match self
Expand All @@ -95,14 +95,14 @@ impl HealthHandle {
.gt(&self.max_unsafe_interval)
{
warn!(target: "rollup_boost::health", curr_unix = %t, unsafe_unix = %block.header.timestamp, "Builder client - unsafe block timestamp is too old updating health status");
self.probes.set_health(Health::PartialContent);
self.probes.set_health(Health::PartialContent).await;
} else {
self.probes.set_health(Health::Healthy);
self.probes.set_health(Health::Healthy).await;
}
}
Err(e) => {
warn!(target: "rollup_boost::health", "Builder client - Failed to get unsafe block {} - updating health status", e);
self.probes.set_health(Health::PartialContent);
self.probes.set_health(Health::PartialContent).await;
}
};
}
Expand Down Expand Up @@ -327,7 +327,7 @@ mod tests {

health_handle.spawn();
tokio::time::sleep(Duration::from_secs(2)).await;
assert!(matches!(probes.health(), Health::Healthy));
assert!(matches!(probes.health().await, Health::Healthy));
Ok(())
}

Expand Down Expand Up @@ -369,7 +369,7 @@ mod tests {

health_handle.spawn();
tokio::time::sleep(Duration::from_secs(2)).await;
assert!(matches!(probes.health(), Health::PartialContent));
assert!(matches!(probes.health().await, Health::PartialContent));
Ok(())
}

Expand Down Expand Up @@ -411,7 +411,7 @@ mod tests {

health_handle.spawn();
tokio::time::sleep(Duration::from_secs(2)).await;
assert!(matches!(probes.health(), Health::ServiceUnavailable));
assert!(matches!(probes.health().await, Health::ServiceUnavailable));
Ok(())
}

Expand Down Expand Up @@ -452,7 +452,7 @@ mod tests {

health_handle.spawn();
tokio::time::sleep(Duration::from_secs(2)).await;
assert!(matches!(probes.health(), Health::Healthy));
assert!(matches!(probes.health().await, Health::Healthy));
Ok(())
}

Expand Down Expand Up @@ -493,7 +493,7 @@ mod tests {

health_handle.spawn();
tokio::time::sleep(Duration::from_secs(2)).await;
assert!(matches!(probes.health(), Health::Healthy));
assert!(matches!(probes.health().await, Health::Healthy));
Ok(())
}

Expand Down Expand Up @@ -533,7 +533,7 @@ mod tests {

health_handle.spawn();
tokio::time::sleep(Duration::from_secs(2)).await;
assert!(matches!(probes.health(), Health::PartialContent));
assert!(matches!(probes.health().await, Health::PartialContent));
Ok(())
}

Expand Down Expand Up @@ -574,7 +574,7 @@ mod tests {

health_handle.spawn();
tokio::time::sleep(Duration::from_secs(2)).await;
assert!(matches!(probes.health(), Health::ServiceUnavailable));
assert!(matches!(probes.health().await, Health::ServiceUnavailable));
Ok(())
}

Expand Down
12 changes: 6 additions & 6 deletions crates/rollup-boost/src/probe.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ use jsonrpsee::{
http_client::{HttpRequest, HttpResponse},
server::HttpBody,
};
use parking_lot::Mutex;
use tokio::sync::Mutex;
use tower::{Layer, Service};
use tracing::info;

Expand Down Expand Up @@ -45,13 +45,13 @@ pub struct Probes {
}

impl Probes {
pub fn set_health(&self, value: Health) {
pub async fn set_health(&self, value: Health) {
info!(target: "rollup_boost::probe", "Updating health probe to to {:?}", value);
*self.health.lock() = value;
*self.health.lock().await = value;
}

pub fn health(&self) -> Health {
*self.health.lock()
pub async fn health(&self) -> Health {
*self.health.lock().await
}
}

Expand Down Expand Up @@ -115,7 +115,7 @@ where
async move {
match request.uri().path() {
// Return health status
"/healthz" => Ok(service.probes.health().into()),
"/healthz" => Ok(service.probes.health().await.into()),
// Service is responding, and therefor ready
"/readyz" => Ok(ok()),
// Service is responding, and therefor live
Expand Down
Loading
Loading