refactor(l1): spawned p2p

Cargo.toml

@@ -74,7 +74,7 @@ crc32fast = "1.4.2"
 lazy_static = "1.5.0"
 sha2 = "0.10.8"
 sha3 = "0.10.8"
-tokio-util = { version = "0.7.12", features = ["rt"] }
+tokio-util = { version = "0.7.15", features = ["rt"] }
 jsonwebtoken = "9.3.0"
 rand = "0.8.5"
 cfg-if = "1.0.0"
@@ -95,6 +95,10 @@ eyre = "0.6.12"
 kzg-rs = "0.2.6"
 libsql = "0.9.10"
 futures = "0.3.31"
+# Changing the tag for spawned will break the TDX image build
+# When updating it try to build the TDX image and update service.nix with the new hash
+spawned-concurrency = {git = "https://github.com/lambdaclass/spawned.git", tag = "v0.1.2-alpha"}
+spawned-rt = {git = "https://github.com/lambdaclass/spawned.git", tag = "v0.1.2-alpha"}
 lambdaworks-crypto = "0.11.0"
 
 [patch.crates-io]

crates/l2/Cargo.toml

@@ -34,13 +34,11 @@ keccak-hash.workspace = true
 envy = "0.4.2"
 rand.workspace = true
 thiserror.workspace = true
+spawned-rt.workspace = true
+spawned-concurrency.workspace = true
 directories = "5.0.1"
 bincode = "1.3.3"
 serde_with = "3.11.0"
-# Changing the tag for spawned will break the TDX image build
-# When updating it try to build the TDX image and update service.nix with the new hash
-spawned-concurrency = { git = "https://github.com/lambdaclass/spawned.git", tag = "v0.1.0-alpha" }
-spawned-rt = { git = "https://github.com/lambdaclass/spawned.git", tag = "v0.1.0-alpha" }
 lazy_static.workspace = true
 aligned-sdk = { git = "https://github.com/yetanotherco/aligned_layer", tag = "v0.16.1" }
 ethers = "2.0"

crates/l2/based/block_fetcher.rs

@@ -17,7 +17,11 @@ use ethrex_rpc::{EthClient, types::receipt::RpcLog};
 use ethrex_storage::Store;
 use ethrex_storage_rollup::{RollupStoreError, StoreRollup};
 use keccak_hash::keccak;
-use spawned_concurrency::{CallResponse, CastResponse, GenServer, send_after};
+use spawned_concurrency::{
+    error::GenServerError,
+    messages::Unused,
+    tasks::{CastResponse, GenServer, GenServerHandle, send_after},
+};
 use tracing::{debug, error, info};
 
 use crate::{
@@ -52,12 +56,14 @@ pub enum BlockFetcherError {
     EvmError(#[from] ethrex_vm::EvmError),
     #[error("Failed to produce the blob bundle")]
     BlobBundleError,
-    #[error("Failed to compute privileged transactions hash: {0}")]
+    #[error("Failed to compute deposit logs hash: {0}")]
     PrivilegedTransactionError(
         #[from] ethrex_l2_common::privileged_transactions::PrivilegedTransactionError,
     ),
+    // TODO: Avoid propagating GenServerErrors outside GenServer modules
+    // See https://github.com/lambdaclass/ethrex/issues/3376
     #[error("Spawned GenServer Error")]
-    GenServerError(spawned_concurrency::GenServerError),
+    GenServerError(GenServerError),
 }
 
 #[derive(Clone)]
@@ -131,7 +137,8 @@ impl BlockFetcher {
 }
 
 impl GenServer for BlockFetcher {
-    type InMsg = InMessage;
+    type CallMsg = Unused;
+    type CastMsg = InMessage;
     type OutMsg = OutMessage;
     type State = BlockFetcherState;
     type Error = BlockFetcherError;
@@ -140,32 +147,23 @@ impl GenServer for BlockFetcher {
         Self {}
     }
 
-    async fn handle_call(
-        &mut self,
-        _message: Self::InMsg,
-        _tx: &spawned_rt::mpsc::Sender<spawned_concurrency::GenServerInMsg<Self>>,
-        _state: &mut Self::State,
-    ) -> spawned_concurrency::CallResponse<Self::OutMsg> {
-        CallResponse::Reply(OutMessage::Done)
-    }
-
     async fn handle_cast(
         &mut self,
-        _message: Self::InMsg,
-        _tx: &spawned_rt::mpsc::Sender<spawned_concurrency::GenServerInMsg<Self>>,
-        state: &mut Self::State,
-    ) -> spawned_concurrency::CastResponse {
+        _message: Self::CastMsg,
+        handle: &GenServerHandle<Self>,
+        mut state: Self::State,
+    ) -> CastResponse<Self> {
         if let SequencerStatus::Following = state.sequencer_state.status().await {
-            let _ = fetch(state).await.inspect_err(|err| {
+            let _ = fetch(&mut state).await.inspect_err(|err| {
                 error!("Block Fetcher Error: {err}");
             });
         }
         send_after(
             Duration::from_millis(state.fetch_interval_ms),
-            _tx.clone(),
-            Self::InMsg::Fetch,
+            handle.clone(),
+            Self::CastMsg::Fetch,
         );
-        CastResponse::NoReply
+        CastResponse::NoReply(state)
     }
 }
 

crates/l2/based/state_updater.rs

@@ -6,7 +6,11 @@ use ethrex_l2_sdk::calldata::encode_calldata;
 use ethrex_rpc::{EthClient, clients::Overrides};
 use ethrex_storage::Store;
 use ethrex_storage_rollup::{RollupStoreError, StoreRollup};
-use spawned_concurrency::{CallResponse, CastResponse, GenServer, GenServerError, send_after};
+use spawned_concurrency::{
+    error::GenServerError,
+    messages::Unused,
+    tasks::{CastResponse, GenServer, GenServerHandle, send_after},
+};
 use tracing::{debug, error, info, warn};
 
 use crate::{
@@ -32,6 +36,8 @@ pub enum StateUpdaterError {
     InvalidForkChoice(#[from] ethrex_blockchain::error::InvalidForkChoice),
     #[error("Internal Error: {0}")]
     InternalError(String),
+    // TODO: Avoid propagating GenServerErrors outside GenServer modules
+    // See https://github.com/lambdaclass/ethrex/issues/3376
     #[error("Spawned GenServer Error")]
     GenServerError(GenServerError),
 }
@@ -99,7 +105,8 @@ impl StateUpdater {
 }
 
 impl GenServer for StateUpdater {
-    type InMsg = InMessage;
+    type CallMsg = Unused;
+    type CastMsg = InMessage;
     type OutMsg = OutMessage;
     type State = StateUpdaterState;
     type Error = StateUpdaterError;
@@ -108,30 +115,21 @@ impl GenServer for StateUpdater {
         Self {}
     }
 
-    async fn handle_call(
-        &mut self,
-        _message: Self::InMsg,
-        _tx: &spawned_rt::mpsc::Sender<spawned_concurrency::GenServerInMsg<Self>>,
-        _state: &mut Self::State,
-    ) -> spawned_concurrency::CallResponse<Self::OutMsg> {
-        CallResponse::Reply(OutMessage::Done)
-    }
-
     async fn handle_cast(
         &mut self,
-        _message: Self::InMsg,
-        tx: &spawned_rt::mpsc::Sender<spawned_concurrency::GenServerInMsg<Self>>,
-        state: &mut Self::State,
-    ) -> spawned_concurrency::CastResponse {
-        let _ = update_state(state)
+        _message: Self::CastMsg,
+        handle: &GenServerHandle<Self>,
+        mut state: Self::State,
+    ) -> CastResponse<Self> {
+        let _ = update_state(&mut state)
             .await
             .inspect_err(|err| error!("State Updater Error: {err}"));
         send_after(
             Duration::from_millis(state.check_interval_ms),
-            tx.clone(),
-            Self::InMsg::UpdateState,
+            handle.clone(),
+            Self::CastMsg::UpdateState,
         );
-        CastResponse::NoReply
+        CastResponse::NoReply(state)
     }
 }
 

crates/l2/sequencer/block_producer.rs

@@ -16,8 +16,10 @@ use ethrex_storage_rollup::StoreRollup;
 use ethrex_vm::BlockExecutionResult;
 use keccak_hash::H256;
 use payload_builder::build_payload;
-use spawned_concurrency::{CallResponse, CastResponse, GenServer, GenServerInMsg, send_after};
-use spawned_rt::mpsc::Sender;
+use spawned_concurrency::{
+    messages::Unused,
+    tasks::{CastResponse, GenServer, GenServerHandle, send_after},
+};
 use tracing::{debug, error, info};
 
 use crate::{
@@ -104,7 +106,8 @@ impl BlockProducer {
 }
 
 impl GenServer for BlockProducer {
-    type InMsg = InMessage;
+    type CallMsg = Unused;
+    type CastMsg = InMessage;
     type OutMsg = OutMessage;
     type State = BlockProducerState;
 
@@ -114,33 +117,24 @@ impl GenServer for BlockProducer {
         Self {}
     }
 
-    async fn handle_call(
-        &mut self,
-        _message: Self::InMsg,
-        _tx: &Sender<GenServerInMsg<Self>>,
-        _state: &mut Self::State,
-    ) -> CallResponse<Self::OutMsg> {
-        CallResponse::Reply(OutMessage::Done)
-    }
-
     async fn handle_cast(
         &mut self,
-        _message: Self::InMsg,
-        tx: &Sender<GenServerInMsg<Self>>,
-        state: &mut Self::State,
-    ) -> CastResponse {
+        _message: Self::CastMsg,
+        handle: &GenServerHandle<Self>,
+        state: Self::State,
+    ) -> CastResponse<Self> {
         // Right now we only have the Produce message, so we ignore the message
         if let SequencerStatus::Sequencing = state.sequencer_state.status().await {
-            let _ = produce_block(state)
+            let _ = produce_block(&state)
                 .await
                 .inspect_err(|e| error!("Block Producer Error: {e}"));
         }
         send_after(
             Duration::from_millis(state.block_time_ms),
-            tx.clone(),
-            Self::InMsg::Produce,
+            handle.clone(),
+            Self::CastMsg::Produce,
         );
-        CastResponse::NoReply
+        CastResponse::NoReply(state)
     }
 }
 

crates/l2/sequencer/errors.rs

@@ -12,7 +12,7 @@ use ethrex_rpc::clients::eth::errors::{CalldataEncodeError, EthClientError};
 use ethrex_storage::error::StoreError;
 use ethrex_storage_rollup::RollupStoreError;
 use ethrex_vm::{EvmError, ProverDBError};
-use spawned_concurrency::GenServerError;
+use spawned_concurrency::error::GenServerError;
 use tokio::task::JoinError;
 
 #[derive(Debug, thiserror::Error)]
@@ -61,6 +61,8 @@ pub enum L1WatcherError {
     FailedAccessingRollUpStore(#[from] RollupStoreError),
     #[error("{0}")]
     Custom(String),
+    // TODO: Avoid propagating GenServerErrors outside GenServer modules
+    // See https://github.com/lambdaclass/ethrex/issues/3376
     #[error("Spawned GenServer Error")]
     GenServerError(GenServerError),
 }
@@ -123,6 +125,8 @@ pub enum ProofSenderError {
     InternalError(String),
     #[error("Failed to parse OnChainProposer response: {0}")]
     FailedToParseOnChainProposerResponse(String),
+    // TODO: Avoid propagating GenServerErrors outside GenServer modules
+    // See https://github.com/lambdaclass/ethrex/issues/3376
     #[error("Spawned GenServer Error")]
     GenServerError(GenServerError),
     #[error("Proof Sender failed because of a rollup store error: {0}")]
@@ -189,6 +193,8 @@ pub enum BlockProducerError {
     FailedToEncodeAccountStateDiff(#[from] StateDiffError),
     #[error("Failed to get data from: {0}")]
     FailedToGetDataFrom(String),
+    // TODO: Avoid propagating GenServerErrors outside GenServer modules
+    // See https://github.com/lambdaclass/ethrex/issues/3376
     #[error("Spawned GenServer Error")]
     GenServerError(GenServerError),
 }
@@ -235,6 +241,8 @@ pub enum CommitterError {
     FailedToGetWithdrawals(#[from] UtilsError),
     #[error("Privileged Transaction error: {0}")]
     PrivilegedTransactionError(#[from] PrivilegedTransactionError),
+    // TODO: Avoid propagating GenServerErrors outside GenServer modules
+    // See https://github.com/lambdaclass/ethrex/issues/3376
     #[error("Spawned GenServer Error")]
     GenServerError(GenServerError),
 }
@@ -261,6 +269,8 @@ pub enum MetricsGathererError {
     EthClientError(#[from] EthClientError),
     #[error("MetricsGatherer: {0}")]
     TryInto(String),
+    // TODO: Avoid propagating GenServerErrors outside GenServer modules
+    // See https://github.com/lambdaclass/ethrex/issues/3376
     #[error("Spawned GenServer Error")]
     GenServerError(GenServerError),
 }
@@ -275,6 +285,8 @@ pub enum ExecutionCacheError {
 
 #[derive(Debug, thiserror::Error)]
 pub enum ConnectionHandlerError {
+    // TODO: Avoid propagating GenServerErrors outside GenServer modules
+    // See https://github.com/lambdaclass/ethrex/issues/3376
     #[error("Spawned GenServer Error")]
     GenServerError(GenServerError),
 }

crates/l2/sequencer/l1_committer.rs

@@ -37,8 +37,10 @@ use std::{collections::HashMap, sync::Arc};
 use tracing::{debug, error, info, warn};
 
 use super::{errors::BlobEstimationError, utils::random_duration};
-use spawned_concurrency::{CallResponse, CastResponse, GenServer, GenServerInMsg, send_after};
-use spawned_rt::mpsc::Sender;
+use spawned_concurrency::{
+    messages::Unused,
+    tasks::{CastResponse, GenServer, GenServerHandle, send_after},
+};
 
 const COMMIT_FUNCTION_SIGNATURE_BASED: &str =
     "commitBatch(uint256,bytes32,bytes32,bytes32,bytes32,bytes[])";
@@ -136,7 +138,8 @@ impl L1Committer {
 }
 
 impl GenServer for L1Committer {
-    type InMsg = InMessage;
+    type CallMsg = Unused;
+    type CastMsg = InMessage;
     type OutMsg = OutMessage;
     type State = CommitterState;
 
@@ -146,30 +149,21 @@ impl GenServer for L1Committer {
         Self {}
     }
 
-    async fn handle_call(
-        &mut self,
-        _message: Self::InMsg,
-        _tx: &Sender<GenServerInMsg<Self>>,
-        _state: &mut Self::State,
-    ) -> CallResponse<Self::OutMsg> {
-        CallResponse::Reply(OutMessage::Done)
-    }
-
     async fn handle_cast(
         &mut self,
-        _message: Self::InMsg,
-        tx: &Sender<GenServerInMsg<Self>>,
-        state: &mut Self::State,
-    ) -> CastResponse {
+        _message: Self::CastMsg,
+        handle: &GenServerHandle<Self>,
+        mut state: Self::State,
+    ) -> CastResponse<Self> {
         // Right now we only have the Commit message, so we ignore the message
         if let SequencerStatus::Sequencing = state.sequencer_state.status().await {
-            let _ = commit_next_batch_to_l1(state)
+            let _ = commit_next_batch_to_l1(&mut state)
                 .await
                 .inspect_err(|err| error!("L1 Committer Error: {err}"));
         }
         let check_interval = random_duration(state.commit_time_ms);
-        send_after(check_interval, tx.clone(), Self::InMsg::Commit);
-        CastResponse::NoReply
+        send_after(check_interval, handle.clone(), Self::CastMsg::Commit);
+        CastResponse::NoReply(state)
     }
 }