Fix sequencer crash tests

.github/workflows/functional.yml

@@ -126,13 +126,32 @@ jobs:
 
       # Run again just to be sure as some tests are flaky
       - name: Run functional tests (2)
+        id: funcTestsRun2
         if: steps.funcTestsRun1.outcome == 'failure'
+        continue-on-error: true
         run: |
           NEWPATH="$(realpath target/debug/)"
           export PATH="${NEWPATH}:${PATH}"
           which strata-client
           cd functional-tests && poetry run python entry.py
 
+      - name: Upload logs as build artifact on failure
+        if: steps.funcTestsRun1.outcome == 'failure' || steps.funcTestsRun2.outcome == 'failure'
+        id: upload_logs
+        uses: actions/upload-artifact@v4
+        with:
+          name: fntest_dd
+          path: functional-tests/_dd/
+          retention-days: 30
+          if-no-files-found: error
+
+      - name: Fail job if functional tests fail
+        if: steps.funcTestsRun2.outcome == 'failure'
+        run: |
+          echo "Functional tests failed"
+          exit 1
+
+
   functional-tests-success:
     name: Check that all checks pass
     runs-on: ubuntu-latest

bin/strata-client/src/el_sync.rs

@@ -0,0 +1,118 @@
+use strata_db::DbError;
+use strata_eectl::{engine::ExecEngineCtl, errors::EngineError, messages::ExecPayloadData};
+use strata_state::id::L2BlockId;
+use strata_storage::NodeStorage;
+use thiserror::Error;
+use tracing::{debug, info};
+
+#[derive(Debug, Error)]
+pub enum Error {
+    #[error("missing chainstate for slot {0}")]
+    MissingChainstate(u64),
+    #[error("missing l2block {0}")]
+    MissingL2Block(L2BlockId),
+    #[error("db: {0}")]
+    Db(#[from] DbError),
+    #[error("engine: {0}")]
+    Engine(#[from] EngineError),
+}
+
+/// Sync missing blocks in EL using payloads stored in L2 block database.
+///
+/// TODO: retry on network errors
+pub fn sync_chainstate_to_el(
+    storage: &NodeStorage,
+    engine: &impl ExecEngineCtl,
+) -> Result<(), Error> {
+    let chainstate_manager = storage.chainstate();
+    let l2_block_manager = storage.l2();
+    let earliest_idx = chainstate_manager.get_earliest_write_idx_blocking()?;
+    let latest_idx = chainstate_manager.get_last_write_idx_blocking()?;
+
+    info!(%earliest_idx, %latest_idx, "search for last known idx");
+
+    // last idx of chainstate whose corresponding block is present in el
+    let sync_from_idx = find_last_match((earliest_idx, latest_idx), |idx| {
+        let Some(chain_state) = chainstate_manager.get_toplevel_chainstate_blocking(idx)? else {
+            return Err(Error::MissingChainstate(idx));
+        };
+
+        let block_id = chain_state.chain_tip_blkid();
+
+        Ok(engine.check_block_exists(*block_id)?)
+    })?
+    .map(|idx| idx + 1) // sync from next index
+    .unwrap_or(0); // sync from genesis
+
+    info!(%sync_from_idx, "last known index in EL");
+
+    for idx in sync_from_idx..=latest_idx {
+        debug!(?idx, "Syncing chainstate");
+        let Some(chain_state) = chainstate_manager.get_toplevel_chainstate_blocking(idx)? else {
+            return Err(Error::MissingChainstate(idx));
+        };
+
+        let block_id = chain_state.chain_tip_blkid();
+
+        let Some(l2block) = l2_block_manager.get_block_data_blocking(block_id)? else {
+            return Err(Error::MissingL2Block(*block_id));
+        };
+
+        let payload = ExecPayloadData::from_l2_block_bundle(&l2block);
+
+        engine.submit_payload(payload)?;
+        engine.update_head_block(*block_id)?;
+    }
+
+    Ok(())
+}
+
+fn find_last_match(
+    range: (u64, u64),
+    predicate: impl Fn(u64) -> Result<bool, Error>,
+) -> Result<Option<u64>, Error> {
+    let (mut left, mut right) = range;
+
+    // Check the leftmost value first
+    if !predicate(left)? {
+        return Ok(None); // If the leftmost value is false, no values can be true
+    }
+
+    let mut best_match = None;
+
+    // Proceed with binary search
+    while left <= right {
+        let mid = left + (right - left) / 2;
+
+        if predicate(mid)? {
+            best_match = Some(mid); // Update best match
+            left = mid + 1; // Continue searching in the right half
+        } else {
+            right = mid - 1; // Search in the left half
+        }
+    }
+
+    Ok(best_match)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_find_last_match() {
+        // find match
+        assert!(matches!(
+            find_last_match((0, 5), |idx| Ok(idx < 3)),
+            Ok(Some(2))
+        ));
+        // found no match
+        assert!(matches!(find_last_match((0, 5), |_| Ok(false)), Ok(None)));
+        // got error
+        let error_message = "intentional error for test";
+        assert!(matches!(
+            find_last_match((0, 5), |_| Err(EngineError::Other(error_message.into()))?),
+            Err(err) if err.to_string().contains(error_message)
+        ));
+    }
+}

bin/strata-client/src/main.rs

@@ -1,6 +1,7 @@
 use std::{sync::Arc, time::Duration};
 
 use bitcoin::{hashes::Hash, BlockHash};
+use el_sync::sync_chainstate_to_el;
 use jsonrpsee::Methods;
 use parking_lot::lock_api::RwLock;
 use rpc_client::sync_client;
@@ -45,6 +46,7 @@
 use crate::{args::Args, helpers::*};
 
 mod args;
+mod el_sync;
 mod errors;
 mod extractor;
 mod helpers;
@@ -284,8 +286,8 @@
         }
         Ok(false) => {
             // Current chain tip tip block is not known by the EL.
-            // TODO: Try to sync EL using existing block payloads from DB.
-            anyhow::bail!("missing expected evm block, block_id = {}", chain_tip);
+            warn!(%chain_tip, "missing expected EVM block");
+            sync_chainstate_to_el(storage, engine)?;
         }
         Err(error) => {
             // Likely network issue

bin/strata-sequencer-client/src/duty_executor.rs

@@ -1,16 +1,15 @@
 use std::{collections::HashSet, sync::Arc};
 
-use strata_primitives::buf::Buf32;
 use strata_rpc_api::StrataSequencerApiClient;
 use strata_rpc_types::HexBytes64;
 use strata_sequencer::{
     block_template::{BlockCompletionData, BlockGenerationConfig},
-    duty::types::{BatchCheckpointDuty, BlockSigningDuty, Duty, IdentityData},
+    duty::types::{BatchCheckpointDuty, BlockSigningDuty, Duty, DutyId, IdentityData},
     utils::now_millis,
 };
 use thiserror::Error;
 use tokio::{runtime::Handle, select, sync::mpsc};
-use tracing::{debug, error};
+use tracing::{debug, error, info, warn};
 
 use crate::helpers::{sign_checkpoint, sign_header};
 
@@ -36,28 +35,29 @@
     // Keep track of seen duties to avoid processing the same duty multiple times.
     // Does not need to be persisted, as new duties are generated based on current chain state.
     let mut seen_duties = HashSet::new();
-    let (failed_duties_tx, mut failed_duties_rx) = mpsc::channel::<Buf32>(8);
+    let (failed_duties_tx, mut failed_duties_rx) = mpsc::channel::<DutyId>(8);
 
     loop {
         select! {
             duty = duty_rx.recv() => {
                 if let Some(duty) = duty {
                     let duty_id = duty.id();
                     if seen_duties.contains(&duty_id) {
-                        debug!("skipping already seen duty: {:?}", duty);
+                        debug!(%duty_id, "skipping already seen duty");
                         continue;
                     }
-                    seen_duties.insert(duty.id());
+                    seen_duties.insert(duty_id);
                     handle.spawn(handle_duty(rpc.clone(), duty, idata.clone(), failed_duties_tx.clone()));
                 } else {
                     // tx is closed, we are done
                     return Ok(());
                 }
             }
             failed_duty = failed_duties_rx.recv() => {
-                if let Some(failed_duty_id) = failed_duty {
+                if let Some(duty_id) = failed_duty {
                     // remove from seen duties, so we can retry if the duty is seen again
-                    seen_duties.remove(&failed_duty_id);
+                    warn!(%duty_id, "removing failed duty");
+                    seen_duties.remove(&duty_id);
                 }
             }
         }
@@ -68,33 +68,37 @@
     rpc: Arc<R>,
     duty: Duty,
     idata: IdentityData,
-    failed_duties_tx: mpsc::Sender<Buf32>,
+    failed_duties_tx: mpsc::Sender<DutyId>,
 ) where
     R: StrataSequencerApiClient + Send + Sync,
 {
-    debug!("handle_duty: {:?}", duty);
+    let duty_id = duty.id();
+    debug!(%duty_id, ?duty, "handle_duty");
     let duty_result = match duty.clone() {
-        Duty::SignBlock(duty) => handle_sign_block_duty(rpc, duty, idata).await,
-        Duty::CommitBatch(duty) => handle_commit_batch_duty(rpc, duty, idata).await,
+        Duty::SignBlock(duty) => handle_sign_block_duty(rpc, duty, duty_id, idata).await,
+        Duty::CommitBatch(duty) => handle_commit_batch_duty(rpc, duty, duty_id, idata).await,
     };
 
-    if let Err(e) = duty_result {
-        error!(?duty, "duty failed: {}", e);
+    if let Err(error) = duty_result {
+        error!(%duty_id, %error, "duty failed");
         let _ = failed_duties_tx.send(duty.id()).await;
     }
 }
 
 async fn handle_sign_block_duty<R>(
     rpc: Arc<R>,
     duty: BlockSigningDuty,
+    duty_id: DutyId,
     idata: IdentityData,
 ) -> Result<(), DutyExecError>
 where
     R: StrataSequencerApiClient + Send + Sync,
 {
-    if now_millis() < duty.target_ts() {
+    let now = now_millis();
+    if now < duty.target_ts() {
         // wait until target time
         // TODO: ensure duration is within some bounds
+        warn!(%duty_id, %now, target = duty.target_ts(), "got duty too early; sleeping till target time");
         tokio::time::sleep(tokio::time::Duration::from_millis(
             duty.target_ts() - now_millis(),
         ))
@@ -107,26 +111,35 @@
         .await
         .map_err(DutyExecError::GenerateTemplate)?;
 
+    let id = template.template_id();
+
+    info!(%duty_id, block_id = %id, "got block template");
+
     let signature = sign_header(template.header(), &idata.key);
     let completion = BlockCompletionData::from_signature(signature);
 
     rpc.complete_block_template(template.template_id(), completion)
         .await
         .map_err(DutyExecError::CompleteTemplate)?;
 
+    info!(%duty_id, block_id = %id, "block signing complete");
+
     Ok(())
 }
 
 async fn handle_commit_batch_duty<R>(
     rpc: Arc<R>,
     duty: BatchCheckpointDuty,
+    duty_id: DutyId,
     idata: IdentityData,
 ) -> Result<(), DutyExecError>
 where
     R: StrataSequencerApiClient + Send + Sync,
 {
     let sig = sign_checkpoint(duty.inner(), &idata.key);
 
+    debug!(%duty_id, %sig, "checkpoint signature");
+
     rpc.complete_checkpoint_signature(duty.inner().batch_info().epoch(), HexBytes64(sig.0))
         .await
         .map_err(DutyExecError::CompleteCheckpoint)?;

bin/strata-sequencer-client/src/duty_fetcher.rs

@@ -3,7 +3,7 @@
 use strata_rpc_api::StrataSequencerApiClient;
 use strata_sequencer::duty::types::Duty;
 use tokio::sync::mpsc;
-use tracing::{debug, error, warn};
+use tracing::{error, info, warn};
 
 pub(crate) async fn duty_fetcher_worker<R>(
     rpc: Arc<R>,
@@ -25,7 +25,7 @@
             }
         };
 
-        debug!("got {} duties", duties.len());
+        info!(count = %duties.len(), "got new duties");
 
         for duty in duties {
             if duty_tx.send(duty).await.is_err() {

crates/common/src/bail_manager.rs

@@ -5,6 +5,7 @@ use tokio::sync::watch;
 pub static BAIL_DUTY_SIGN_BLOCK: &str = "duty_sign_block";
 pub static BAIL_ADVANCE_CONSENSUS_STATE: &str = "advance_consensus_state";
 pub static BAIL_SYNC_EVENT: &str = "sync_event";
+pub static BAIL_SYNC_EVENT_NEW_TIP: &str = "sync_event_new_tip";
 
 struct BailWatch {
     sender: watch::Sender<Option<String>>,

crates/consensus-logic/src/csm/worker.rs

@@ -5,7 +5,7 @@
 use std::{sync::Arc, thread};
 
 #[cfg(feature = "debug-utils")]
-use strata_common::bail_manager::{check_bail_trigger, BAIL_SYNC_EVENT};
+use strata_common::bail_manager::{check_bail_trigger, BAIL_SYNC_EVENT, BAIL_SYNC_EVENT_NEW_TIP};
 use strata_db::{
     traits::*,
     types::{CheckpointConfStatus, CheckpointEntry, CheckpointProvingStatus},
@@ -143,7 +143,12 @@ impl WorkerState {
         debug!(?ev, "processing sync event");
 
         #[cfg(feature = "debug-utils")]
-        check_bail_trigger(BAIL_SYNC_EVENT);
+        {
+            check_bail_trigger(BAIL_SYNC_EVENT);
+            if matches!(ev, strata_state::sync_event::SyncEvent::NewTipBlock(_)) {
+                check_bail_trigger(BAIL_SYNC_EVENT_NEW_TIP);
+            }
+        }
 
         // Compute the state transition.
         let context = client_transition::StorageEventContext::new(&self.storage);