fix(sequencer): fixes after rebase

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<BP: BlockPublisherTrait + Send + 'static> Actor for ExecutorActor<BP> {
     }
 }
 
+impl<BP: BlockPublisherTrait + Send + 'static> Message<ProduceBlock> for ExecutorActor<BP> {
+    type Reply = Result<()>;
+
+    async fn handle(
+        &mut self,
+        ProduceBlock: ProduceBlock,
+        _ctx: &mut Context<Self, Self::Reply>,
+    ) -> 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<BP: BlockPublisherTrait + Send + 'static> Message<Transaction> for ExecutorActor<BP> {
     type Reply = Result<()>;
 
@@ -263,41 +302,20 @@ impl<BP: BlockPublisherTrait + Send + 'static> Message<GetChannelId> for Executo
     }
 }
 
-impl<BP: BlockPublisherTrait + Send + 'static> Message<ProduceBlock> for ExecutorActor<BP> {
-    type Reply = Result<()>;
+impl<BP: BlockPublisherTrait + Send + 'static> Message<GetCrossZoneDeadLetters>
+    for ExecutorActor<BP>
+{
+    type Reply = Result<GetCrossZoneDeadLettersReply>;
 
     async fn handle(
         &mut self,
-        ProduceBlock: ProduceBlock,
+        GetCrossZoneDeadLetters: GetCrossZoneDeadLetters,
         _ctx: &mut Context<Self, Self::Reply>,
     ) -> 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,
+        })
     }
 }

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<DeadLetterDispatchRecord>,
+}

lez/sequencer/actors/rpc_server/src/actor/service.rs

@@ -211,16 +211,18 @@ impl<BP: BlockPublisherTrait + Send + 'static> sequencer_service_rpc::RpcServer
     async fn get_cross_zone_dead_letters(
         &self,
     ) -> Result<CrossZoneDeadLetterReport, ErrorObjectOwned> {
-        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),

lez/sequencer/core/src/lib.rs

@@ -1152,6 +1152,7 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
         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