add delayed index to forwarder

Author: lukeiannucci

timeboost-sequencer/src/delayed_inbox.rs

@@ -7,30 +7,31 @@ use alloy_rpc_types::{BlockId, BlockNumberOrTag, Filter};
 use alloy_sol_types::{SolEvent, sol};
 use multisig::PublicKey;
 use tokio::time::sleep;
-use tracing::{error, info};
+use tracing::{error, info, warn};
 
 use crate::queue::BundleQueue;
 
-const INBOX_DELAY: Duration = Duration::from_secs(120);
+const INBOX_DELAY: Duration = Duration::from_secs(60);
 
 sol! {
     event InboxMessageDelivered(uint256 indexed messageNum, bytes data);
     event InboxMessageDeliveredFromOrigin(uint256 indexed messageNum);
 }
 
-pub struct DelayedInbox<N: Network> {
+pub struct DelayedInbox<'a, N: Network> {
     node: PublicKey,
     ibox_addr: Address,
     provider: RootProvider<N>,
     queue: BundleQueue,
+    url: &'a str,
 }
 
-impl<N: Network> DelayedInbox<N> {
+impl<'a, N: Network> DelayedInbox<'a, N> {
     pub async fn connect(
         node: PublicKey,
-        url: &str,
+        url: &'a str,
         ibox_addr: Address,
-        l1_chain_id: u64,
+        parent_chain_id: u64,
         queue: BundleQueue,
     ) -> Result<Self, Error> {
         let provider = RootProvider::<N>::connect(url)
@@ -40,22 +41,20 @@ impl<N: Network> DelayedInbox<N> {
             .get_chain_id()
             .await
             .map_err(|e| Error::RpcError(e.to_string()))?;
-        if l1_chain_id != rpc_chain_id {
-            error!(%l1_chain_id, %rpc_chain_id, "mismatching chain id");
-            return Err(Error::MismatchingChainID(l1_chain_id, rpc_chain_id));
+        if parent_chain_id != rpc_chain_id {
+            error!(%parent_chain_id, %rpc_chain_id, "mismatching chain id");
+            return Err(Error::MismatchingChainID(parent_chain_id, rpc_chain_id));
         }
         Ok(Self {
             node,
             ibox_addr,
             provider,
             queue,
+            url,
         })
     }
 
     pub async fn go(self) {
-        let Self {
-            node, ibox_addr, ..
-        } = self;
         let mut last_finalized = 0;
         let mut last_delayed_index = 0;
         let events = vec![
@@ -71,6 +70,7 @@ impl<N: Network> DelayedInbox<N> {
             {
                 let finalized = b.header().number();
                 if finalized == last_finalized {
+                    sleep(INBOX_DELAY).await;
                     continue;
                 }
                 if last_finalized == 0 {
@@ -84,31 +84,33 @@ impl<N: Network> DelayedInbox<N> {
                     .from_block(last_finalized)
                     .to_block(finalized)
                     .events(&events);
+                last_finalized = finalized;
 
                 if let Ok(mut logs) = self.provider.get_logs(&filter).await {
-                    last_finalized = finalized;
                     // Make sure event logs are in order, we need highest block number first
                     logs.sort_by(|a, b| {
                         b.block_number
                             .cmp(&a.block_number)
                             .then(b.log_index.cmp(&a.log_index))
                     });
-                    if let Some((Some(tx_hash), Some(msg_num))) = logs
+                    if let Some((Some(tx_hash), Some(index))) = logs
                         .first()
                         .map(|log| (log.transaction_hash, log.topics().get(1)))
                     {
                         // Update delayed index if newer
-                        let delayed_index = U256::from_be_bytes(msg_num.0)
+                        let delayed_index = U256::from_be_bytes(index.0)
                             .try_into()
                             .expect("valid msg number");
                         if delayed_index != last_delayed_index {
                             debug_assert!(delayed_index > last_delayed_index);
-                            info!(%node, ?ibox_addr, %last_finalized, %delayed_index, %tx_hash, "delayed index updated");
+                            info!(node = %self.node, ibox_addr = %self.ibox_addr, parent_finalized_block = %finalized, %delayed_index, %tx_hash, "delayed index updated");
                             last_delayed_index = delayed_index;
                             self.queue.set_delayed_inbox_index(delayed_index.into());
                         }
                     }
                 }
+            } else {
+                warn!(node = %self.node, ibox_addr = %self.ibox_addr, url = %self.url, "failed to get latest finalized block");
             }
             sleep(INBOX_DELAY).await;
         }

timeboost-sequencer/src/lib.rs

@@ -19,7 +19,7 @@ use sailfish::consensus::{Consensus, ConsensusMetrics};
 use sailfish::rbc::{Rbc, RbcError, RbcMetrics};
 use sailfish::types::{Action, ConsensusTime, Evidence, Round, RoundNumber};
 use sailfish::{Coordinator, Event};
-use timeboost_types::{BundleVariant, Timestamp, Transaction};
+use timeboost_types::{BundleVariant, DelayedInboxIndex, Timestamp, Transaction};
 use timeboost_types::{CandidateList, CandidateListBytes, InclusionList};
 use tokio::select;
 use tokio::sync::mpsc::{self, Receiver, Sender};
@@ -44,6 +44,7 @@ pub enum Output {
         round: RoundNumber,
         timestamp: Timestamp,
         transactions: Vec<Transaction>,
+        delayed_inbox_index: DelayedInboxIndex,
     },
     UseCommittee(Round),
 }
@@ -300,9 +301,10 @@ impl Task {
                     Ok(incl) => {
                         let round = incl.round();
                         let timestamp = incl.timestamp();
+                        let delayed_inbox_index = incl.delayed_inbox_index();
                         let transactions = self.sorter.sort(incl);
                         if !transactions.is_empty() {
-                            let out = Output::Transactions { round, timestamp, transactions };
+                            let out = Output::Transactions { round, timestamp, transactions, delayed_inbox_index };
                             self.output.send(out).await.map_err(|_| TimeboostError::ChannelClosed)?;
                         }
                         if self.decrypter.has_capacity() {

timeboost/src/forwarder/nitro_forwarder.rs

@@ -8,7 +8,7 @@ use multisig::PublicKey;
 use sailfish::types::RoundNumber;
 use timeboost_proto::forward::forward_api_client::ForwardApiClient;
 use timeboost_proto::inclusion::InclusionList;
-use timeboost_types::{Timestamp, Transaction};
+use timeboost_types::{DelayedInboxIndex, Timestamp, Transaction};
 use tokio::sync::mpsc::{Sender, channel};
 use tokio::task::JoinHandle;
 use tonic::transport::Endpoint;
@@ -44,6 +44,7 @@ impl NitroForwarder {
         round: RoundNumber,
         timestamp: Timestamp,
         txns: &[Transaction],
+        index: DelayedInboxIndex,
     ) -> Result<(), Error> {
         let incl = InclusionList {
             round: *round,
@@ -56,7 +57,7 @@ impl NitroForwarder {
                 })
                 .collect(),
             consensus_timestamp: timestamp.into(),
-            delayed_messages_read: 0,
+            delayed_messages_read: index.into(),
         };
         self.incls_tx
             .send(incl)

timeboost/src/lib.rs

@@ -104,15 +104,15 @@ impl Timeboost {
                     }
                 },
                 out = self.sequencer.next() => match out {
-                    Ok(Output::Transactions { round, timestamp, transactions }) => {
+                    Ok(Output::Transactions { round, timestamp, transactions, delayed_inbox_index }) => {
                         info!(
                             node  = %self.label,
                             round = %round,
                             trxs  = %transactions.len(),
                             "sequencer output"
                         );
                         if let Some(ref mut f) = self.nitro_forwarder {
-                            f.enqueue(round, timestamp, &transactions).await?;
+                            f.enqueue(round, timestamp, &transactions, delayed_inbox_index).await?;
                         }
                         else {
                             warn!(node = %self.label, %round, "no forwarder => dropping output")