diff --git a/lez/sequencer/actors/executor/src/actor.rs b/lez/sequencer/actors/executor/src/actor.rs index 0178d7340..c3ba0292b 100644 --- a/lez/sequencer/actors/executor/src/actor.rs +++ b/lez/sequencer/actors/executor/src/actor.rs @@ -24,8 +24,8 @@ use crate::{ error::Error, protocol::{ GetAccount, GetAccountBalance, GetAccountNonces, GetAccountReply, GetBlock, GetBlockRange, - GetChannelId, GetChannelIdReply, GetLastBlockId, GetProofsAndRoot, GetTransaction, - ProduceBlock, Transaction, + GetChannelId, GetChannelIdReply, GetCrossZoneDeadLetters, GetCrossZoneDeadLettersReply, + GetLastBlockId, GetProofsAndRoot, GetTransaction, ProduceBlock, Transaction, }, }; @@ -107,6 +107,45 @@ impl Actor for ExecutorActor { } } +impl Message for ExecutorActor { + type Reply = Result<()>; + + async fn handle( + &mut self, + ProduceBlock: ProduceBlock, + _ctx: &mut Context, + ) -> Self::Reply { + // Only produce on our turn. + if !self.sequencer.is_our_turn() { + info!("Not our turn to produce a block, skipping"); + return Ok(()); + } + + // Never inscribe a second block at a height we already published: the + // channel would carry two chains from there and nothing resolves that. + // The head rewinds under us when the sdk orphans our own unfinalized + // blocks, and recovers once they finalize, so this is a wait. + if let Some(high_water) = self.sequencer.rewound_below_published() { + warn!( + "Skipping turn: head rewound to {} but block {high_water} is already inscribed; \ + waiting for the channel to restore it", + self.sequencer.next_block_height().saturating_sub(1), + ); + return Ok(()); + } + + info!("Our turn: collecting transactions from mempool, creating block"); + let id = self + .sequencer + .produce_new_block() + .await + .map_err(Error::BlockProductionFailed)?; + + info!("Block with id {id} created"); + Ok(()) + } +} + impl Message for ExecutorActor { type Reply = Result<()>; @@ -263,41 +302,20 @@ impl Message for Executo } } -impl Message for ExecutorActor { - type Reply = Result<()>; +impl Message + for ExecutorActor +{ + type Reply = Result; async fn handle( &mut self, - ProduceBlock: ProduceBlock, + GetCrossZoneDeadLetters: GetCrossZoneDeadLetters, _ctx: &mut Context, ) -> Self::Reply { - // Only produce on our turn. - if !self.sequencer.is_our_turn() { - info!("Not our turn to produce a block, skipping"); - return Ok(()); - } - - // Never inscribe a second block at a height we already published: the - // channel would carry two chains from there and nothing resolves that. - // The head rewinds under us when the sdk orphans our own unfinalized - // blocks, and recovers once they finalize, so this is a wait. - if let Some(high_water) = self.sequencer.rewound_below_published() { - warn!( - "Skipping turn: head rewound to {} but block {high_water} is already inscribed; \ - waiting for the channel to restore it", - self.sequencer.next_block_height().saturating_sub(1), - ); - return Ok(()); - } - - info!("Our turn: collecting transactions from mempool, creating block"); - let id = self - .sequencer - .produce_new_block() - .await - .map_err(Error::BlockProductionFailed)?; - - info!("Block with id {id} created"); - Ok(()) + let (total_retired, retained) = self.sequencer.cross_zone_dead_letters()?; + Ok(GetCrossZoneDeadLettersReply { + total_retired, + retained, + }) } } diff --git a/lez/sequencer/actors/executor/src/protocol.rs b/lez/sequencer/actors/executor/src/protocol.rs index 9fb5be123..e4dc73ee5 100644 --- a/lez/sequencer/actors/executor/src/protocol.rs +++ b/lez/sequencer/actors/executor/src/protocol.rs @@ -6,6 +6,10 @@ use lee_core::{ BlockId, Commitment, account::{Account, AccountId}, }; +use sequencer_core::DeadLetterDispatchRecord; + +#[derive(Copy, Clone)] +pub struct ProduceBlock; pub struct Transaction { pub transaction: LeeTransaction, @@ -53,5 +57,10 @@ pub struct GetChannelIdReply { pub channel_id: [u8; 32], } -#[derive(Copy, Clone)] -pub struct ProduceBlock; +pub struct GetCrossZoneDeadLetters; + +#[derive(Reply)] +pub struct GetCrossZoneDeadLettersReply { + pub total_retired: u64, + pub retained: Vec, +} diff --git a/lez/sequencer/actors/rpc_server/src/actor/service.rs b/lez/sequencer/actors/rpc_server/src/actor/service.rs index eb511a4c8..71f463242 100644 --- a/lez/sequencer/actors/rpc_server/src/actor/service.rs +++ b/lez/sequencer/actors/rpc_server/src/actor/service.rs @@ -211,16 +211,18 @@ impl sequencer_service_rpc::RpcServer async fn get_cross_zone_dead_letters( &self, ) -> Result { - let (total_retired, records) = self - .sequencer - .lock() + let sequencer_executor_actor::protocol::GetCrossZoneDeadLettersReply { + total_retired, + retained, + } = self + .executor_ref + .ask(sequencer_executor_actor::protocol::GetCrossZoneDeadLetters) .await - .cross_zone_dead_letters() - .map_err(|err| internal_error(&err))?; + .map_err(internal_error)?; Ok(CrossZoneDeadLetterReport { total_retired, - retained: records + retained: retained .into_iter() .map(|record| CrossZoneDeadLetter { message_key: HashType(record.message_key), diff --git a/lez/sequencer/core/src/lib.rs b/lez/sequencer/core/src/lib.rs index a3652b2e7..1f433c487 100644 --- a/lez/sequencer/core/src/lib.rs +++ b/lez/sequencer/core/src/lib.rs @@ -1152,6 +1152,7 @@ impl SequencerCore { let total = dbio.get_dead_letter_cross_zone_dispatch_count()?; Ok((total, retained)) } + /// Every background task that holds this sequencer's store handle. /// /// Taken before the core is shared, so a shutdown path can wait for them