backport fixes to release

bin/strata-client/src/el_sync.rs

@@ -1,12 +1,16 @@
-use strata_db::traits::{ChainstateProvider, Database, L2DataProvider};
+use std::sync::Arc;
+
+use strata_db::traits::{ChainstateProvider, Database};
 use strata_eectl::{engine::ExecEngineCtl, messages::ExecPayloadData};
+use strata_storage::L2BlockManager;
 use tracing::debug;
 
 /// Sync missing blocks in EL using payloads stored in L2 block database.
 ///
 /// TODO: retry on network errors
 pub fn sync_chainstate_to_el(
     database: &impl Database,
+    l2_block_manager: Arc<L2BlockManager>,
     engine: &impl ExecEngineCtl,
 ) -> anyhow::Result<()> {
     let chain_state_prov = database.chain_state_provider();
@@ -29,7 +33,6 @@
 
     debug!(?sync_from_idx, "last known index in EL");
 
-    let l2_prov = database.l2_provider();
     for idx in sync_from_idx..=latest_idx {
         debug!(?idx, "Syncing chainstate");
         let Some(chain_state) = chain_state_prov.get_toplevel_state(idx)? else {
@@ -38,13 +41,14 @@
 
         let block_id = chain_state.chain_tip_blockid();
 
-        let Some(l2block) = l2_prov.get_block_data(block_id)? else {
+        let Some(l2block) = l2_block_manager.get_block_blocking(&block_id)? else {
             anyhow::bail!(format!("Missing L2 block idx: {}", block_id));
         };
 
         let payload = ExecPayloadData::from_l2_block_bundle(&l2block);
 
         engine.submit_payload(payload)?;
+        engine.update_safe_block(block_id)?;
     }
 
     Ok(())

bin/strata-client/src/main.rs

@@ -19,7 +19,7 @@
     sync_manager::{self, SyncManager},
 };
 use strata_db::{
-    traits::{ChainstateProvider, Database},
+    traits::{ChainstateProvider, Database, L2DataProvider, L2DataStore},
     DbError,
 };
 use strata_eectl::engine::ExecEngineCtl;
@@ -186,13 +186,9 @@
                 ctx.l2_block_manager.clone(),
                 ctx.sync_manager.clone(),
             );
-            // NOTE: this might block for some time during first run with empty db until genesis
-            // block is generated
-            let mut l2_sync_state =
-                strata_sync::block_until_csm_ready_and_init_sync_state(&l2_sync_context)?;
 
             executor.spawn_critical_async("l2-sync-manager", async move {
-                strata_sync::sync_worker(&mut l2_sync_state, &l2_sync_context)
+                strata_sync::simple_sync_worker(&l2_sync_context)
                     .await
                     .map_err(Into::into)
             });
@@ -257,6 +253,7 @@
 
 fn do_startup_checks(
     database: &impl Database,
+    l2_block_manager: Arc<L2BlockManager>,
     engine: &impl ExecEngineCtl,
     bitcoin_client: &impl Reader,
     runtime: &Runtime,
@@ -301,14 +298,36 @@
         Ok(false) => {
             // Current chain tip tip block is not known by the EL.
             warn!("missing expected evm block, block_id = {}", chain_tip);
-            sync_chainstate_to_el(database, engine)?;
+            sync_chainstate_to_el(database, l2_block_manager, engine)?;
         }
         Err(error) => {
             // Likely network issue
             anyhow::bail!("could not connect to exec engine, err = {}", error);
         }
     }
 
+    // remove any extra L2 blocks beyond latest state idx
+    // this will resolve block production issue due to block already being in L2 db
+    let l2_prov = database.l2_provider();
+    let l2_store = database.l2_store();
+    let mut extra_blockids = Vec::new();
+    let mut blockidx = last_state_idx + 1;
+    loop {
+        info!(?blockidx, "check for extra blocks beyond latest state idx");
+        let mut blockids = l2_prov.get_blocks_at_height(blockidx)?;
+        if blockids.is_empty() {
+            break;
+        }
+        info!(?blockidx, ?blockids, "found extra blocks");
+        extra_blockids.append(&mut blockids);
+        blockidx += 1;
+    }
+    info!(count = extra_blockids.len(), "total extra blocks found");
+    for blockid in extra_blockids {
+        info!(?blockid, "removing extra block");
+        l2_store.del_block_data(blockid)?;
+    }
+
     // everything looks ok
     info!("Startup checks passed");
     Ok(())
@@ -341,6 +360,7 @@
     // do startup checks
     do_startup_checks(
         database.as_ref(),
+        l2_block_manager.clone(),
         engine.as_ref(),
         bitcoin_client.as_ref(),
         runtime,
@@ -349,6 +369,7 @@
     // Start the sync manager.
     let sync_manager: Arc<_> = sync_manager::start_sync_tasks(
         executor,
+        runtime,
         database.clone(),
         l2_block_manager.clone(),
         engine.clone(),

crates/btcio/Cargo.toml

@@ -5,6 +5,7 @@ version = "0.1.0"
 
 [dependencies]
 strata-bridge-tx-builder.workspace = true
+strata-common.workspace = true
 strata-db.workspace = true
 strata-primitives.workspace = true
 strata-rpc-types.workspace = true

crates/btcio/src/reader/query.rs

@@ -6,6 +6,7 @@
 
 use anyhow::bail;
 use bitcoin::BlockHash;
+use strata_common::env::parse_env_or;
 use strata_db::traits::Database;
 use strata_primitives::params::Params;
 use strata_state::l1::{
@@ -26,6 +27,8 @@
     status::{apply_status_updates, L1StatusUpdate},
 };
 
+const L1_POLL_INTERVAL_MS_ENVVAR: &str = "L1_POLL_INTERVAL_MS";
+
 // TODO: remove this
 pub async fn bitcoin_data_reader_task<D: Database + 'static>(
     client: Arc<impl Reader>,
@@ -186,7 +189,12 @@
 
     // Now process each block we missed.
     let scan_start_height = state.next_height();
+    let scan_interval_ms = parse_env_or(L1_POLL_INTERVAL_MS_ENVVAR, 1);
+    let mut interval = tokio::time::interval(Duration::from_millis(scan_interval_ms));
+    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
     for fetch_height in scan_start_height..=client_height {
+        // HACK: need to slow down l1 reader
+        interval.tick().await;
         let l1blkid = match fetch_and_process_block(
             fetch_height,
             client,

crates/consensus-logic/src/client_transition.rs

@@ -94,7 +94,7 @@
             let maturable_height = next_exp_height.saturating_sub(safe_depth);
 
             if maturable_height > params.rollup().horizon_l1_height && state.is_chain_active() {
-                let (wrs, acts) = handle_maturable_height(maturable_height, state);
+                let (wrs, acts) = handle_mature_l1_height(maturable_height, state, database);
                 writes.extend(wrs);
                 actions.extend(acts);
             }
@@ -224,15 +224,41 @@
                 block.block().header().blockidx(),
             ));
             actions.push(SyncAction::UpdateTip(*blkid));
+
+            let (wrs, acts) = handle_checkpoint_finalization(state, blkid, params, database);
+            writes.extend(wrs);
+            actions.extend(acts);
         }
     }
 
     Ok(ClientUpdateOutput::new(writes, actions))
 }
 
-fn handle_maturable_height(
+/// Handles the maturation of L1 height by finalizing checkpoints and emitting
+/// sync actions.
+///
+/// This function checks if there are any verified checkpoints at or before the
+/// given `maturable_height`. If such checkpoints exist, it attempts to
+/// finalize them by checking if the corresponding L2 block is available in the
+/// L2 database. If the L2 block is found, it marks the checkpoint as finalized
+/// and emits a sync action to finalize the L2 block. If the L2 block is not
+/// found, it logs a warning and skips the finalization.
+///
+/// # Arguments
+///
+/// * `maturable_height` - The height at which L1 blocks are considered mature.
+/// * `state` - A reference to the current client state.
+/// * `database` - A reference to the database interface.
+///
+/// # Returns
+///
+/// A tuple containing:
+/// * A vector of [`ClientStateWrite`] representing the state changes to be written.
+/// * A vector of [`SyncAction`] representing the actions to be synchronized.
+fn handle_mature_l1_height(
     maturable_height: u64,
     state: &ClientState,
+    database: &impl Database,
 ) -> (Vec<ClientStateWrite>, Vec<SyncAction>) {
     let mut writes = Vec::new();
     let mut actions = Vec::new();
@@ -242,15 +268,48 @@
         .l1_view()
         .has_verified_checkpoint_before(maturable_height)
     {
-        debug!(%maturable_height, "Writing CheckpointFinalized");
-        writes.push(ClientStateWrite::CheckpointFinalized(maturable_height));
-
-        // Emit sync action for finalizing a l2 block
         if let Some(checkpt) = state
             .l1_view()
             .get_last_verified_checkpoint_before(maturable_height)
         {
-            actions.push(SyncAction::FinalizeBlock(checkpt.batch_info.l2_blockid));
+            // FinalizeBlock Should only be applied when l2_block is actually
+            // available in l2_db
+            // If l2 blocks is not in db then finalization will happen when
+            // l2Block is fetched from the network and the corresponding
+            //checkpoint is already finalized.
+            let l2_blockid = checkpt.batch_info.l2_blockid;
+
+            match database.l2_provider().get_block_data(l2_blockid) {
+                Ok(Some(bundle)) => {
+                    let block_height = bundle.header().blockidx();
+                    // FIXME: only checking height and not for potential fork
+                    match database.chain_state_provider().get_last_state_idx() {
+                        Ok(chainstate_height) => {
+                            if block_height > chainstate_height {
+                                warn!(%maturable_height, %l2_blockid, "l2 block not in chainstate yet, skipping finalize");
+                            } else {
+                                debug!(%maturable_height, "Writing CheckpointFinalized");
+                                writes
+                                    .push(ClientStateWrite::CheckpointFinalized(maturable_height));
+                                // Emit sync action for finalizing a l2 block
+                                info!(%maturable_height, %l2_blockid, "l2 block found in db, push FinalizeBlock SyncAction");
+                                actions.push(SyncAction::FinalizeBlock(l2_blockid));
+                            }
+                        }
+                        Err(e) => {
+                            error!(%e, "error while fetching block data from chainstate db");
+                        }
+                    }
+                }
+                Ok(None) => {
+                    warn!(
+                        %maturable_height,%l2_blockid, "l2 block not in db yet, skipping finalize"
+                    );
+                }
+                Err(e) => {
+                    error!(%e, "error while fetching block data from l2_db");
+                }
+            }
         } else {
             warn!(
             %maturable_height,
@@ -261,7 +320,74 @@
     (writes, actions)
 }
 
-/// Filters a list of `BatchCheckpoint`s, returning only those that form a valid sequence
+/// Handles the finalization of a checkpoint by processing the corresponding L2
+/// block ID.
+///
+/// This function checks if the given L2 block ID corresponds to a verified
+/// checkpoint. If it does, it calculates the maturable height based on the
+/// rollup parameters and the current state. If the L1 height associated with
+/// the L2 block ID is less than the maturable height, it calls
+/// [`handle_mature_l1_height`] and returns writes and sync actions.
+///
+/// # Arguments
+///
+/// * `state` - A reference to the current client state.
+/// * `blkid` - A reference to the L2 block ID to be finalized.
+/// * `params` - A reference to the rollup parameters.
+/// * `database` - A reference to the database interface.
+///
+/// # Returns
+///
+/// A tuple containing:
+/// * A vector of [`ClientStateWrite`] representing the state changes to be written.
+/// * A vector of [`SyncAction`] representing the actions to be synchronized.
+fn handle_checkpoint_finalization(
+    state: &ClientState,
+    blkid: &L2BlockId,
+    params: &Params,
+    database: &impl Database,
+) -> (Vec<ClientStateWrite>, Vec<SyncAction>) {
+    let mut writes = Vec::new();
+    let mut actions = Vec::new();
+    let verified_checkpoints: &[L1Checkpoint] = state.l1_view().verified_checkpoints();
+    match find_l1_height_for_l2_blockid(verified_checkpoints, blkid) {
+        Some(l1_height) => {
+            let safe_depth = params.rollup().l1_reorg_safe_depth as u64;
+
+            // Maturable height is the height at which l1 blocks are sufficiently buried
+            // and have negligible chance of reorg.
+            let maturable_height = state
+                .l1_view()
+                .next_expected_block()
+                .saturating_sub(safe_depth);
+
+            // The l1 height should be handled only if it is less than maturable height
+            if l1_height < maturable_height {
+                let (wrs, acts) = handle_mature_l1_height(l1_height, state, database);
+                writes.extend(wrs);
+                actions.extend(acts);
+            }
+        }
+        None => {
+            debug!(%blkid, "L2 block not found in verified checkpoints, possibly not a last block in the checkpoint.");
+        }
+    }
+    (writes, actions)
+}
+
+/// Searches for a given [`L2BlockId`] within a slice of [`L1Checkpoint`] structs
+/// and returns the height of the corresponding L1 block if found.
+fn find_l1_height_for_l2_blockid(
+    checkpoints: &[L1Checkpoint],
+    target_l2_blockid: &L2BlockId,
+) -> Option<u64> {
+    checkpoints
+        .binary_search_by(|checkpoint| checkpoint.batch_info.l2_blockid.cmp(target_l2_blockid))
+        .ok()
+        .map(|index| checkpoints[index].height)
+}
+
+/// Filters a list of [`BatchCheckpoint`]s, returning only those that form a valid sequence
 /// of checkpoints.
 ///
 /// A valid checkpoint is one whose proof passes verification, and its index follows
@@ -270,12 +396,13 @@
 /// # Arguments
 ///
 /// * `state` - The client's current state, which provides the L1 view and pending checkpoints.
-/// * `checkpoints` - A slice of `BatchCheckpoint`s to be filtered.
+/// * `checkpoints` - A slice of [`BatchCheckpoint`]s to be filtered.
 /// * `params` - Parameters required for verifying checkpoint proofs.
 ///
 /// # Returns
 ///
-/// A vector containing the valid sequence of `BatchCheckpoint`s, starting from the first valid one.
+/// A vector containing the valid sequence of [`BatchCheckpoint`]s, starting from the first valid
+/// one.
 pub fn filter_verified_checkpoints(
     state: &ClientState,
     checkpoints: &[BatchCheckpoint],

crates/consensus-logic/src/config.rs

@@ -0,0 +1,29 @@
+use std::time;
+
+const RETRY_BACKOFF_BASE: u32 = 1024;
+
+/// Run-time config for CSM executor.
+///
+/// This is *not* like system params.
+pub struct CsmExecConfig {
+    /// Base retry duration, which is increases exponentially for each retry.
+    pub retry_base_dur: time::Duration,
+
+    /// Maximum retry count.
+    pub retry_cnt_max: u32,
+
+    /// Retry backoff multiplier used to control the exponential backoff.
+    ///
+    /// This is multiplied against the current wait dur and then divided by
+    /// 1024.  A sensible value for this should ensure that we don't sleep more
+    /// than 10x-20x `retry_base_dur` before terminating.
+    pub retry_backoff_mult: u32,
+}
+
+impl CsmExecConfig {
+    /// Computes the next step of retry backoff.  This is effectively a fixp
+    /// multiplication by the `retry_backoff_mult`.
+    pub fn compute_retry_backoff(&self, cur_dur: time::Duration) -> time::Duration {
+        (cur_dur * self.retry_backoff_mult) / RETRY_BACKOFF_BASE
+    }
+}