stash return stateupdate

Author: ananas-block

src/ingester/error.rs

@@ -15,8 +15,11 @@ pub enum IngesterError {
     EmptyBatchEvent,
     #[error("Invalid event.")]
     InvalidEvent,
-    #[error("Sequence gap detected: {} gaps found", .0.len())]
-    SequenceGapDetected(Vec<SequenceGap>),
+    #[error("Sequence gap detected: {} gaps found", .gaps.len())]
+    SequenceGapDetected { 
+        gaps: Vec<SequenceGap>,
+        parsed_state: crate::ingester::parser::state_update::StateUpdate,
+    },
 }
 
 impl From<sea_orm::error::DbErr> for IngesterError {
@@ -34,8 +37,8 @@ impl From<String> for IngesterError {
 impl From<crate::ingester::parser::state_update::SequenceGapError> for IngesterError {
     fn from(err: crate::ingester::parser::state_update::SequenceGapError) -> Self {
         match err {
-            crate::ingester::parser::state_update::SequenceGapError::GapDetected(gaps) => {
-                IngesterError::SequenceGapDetected(gaps)
+            crate::ingester::parser::state_update::SequenceGapError::GapDetected { gaps, parsed_state } => {
+                IngesterError::SequenceGapDetected { gaps, parsed_state }
             }
         }
     }

src/ingester/mod.rs

@@ -42,7 +42,7 @@ fn derive_block_state_update(
 
     match StateUpdate::merge_updates_with_slot(state_updates, Some(block.metadata.slot)) {
         Ok(merged_update) => Ok(merged_update),
-        Err(SequenceGapError::GapDetected(gaps)) => {
+        Err(SequenceGapError::GapDetected { gaps, parsed_state }) => {
             if let Some(controller) = rewind_controller {
                 let rewind_slot = determine_rewind_slot(&gaps);
                 let reason = format!(
@@ -52,7 +52,7 @@ fn derive_block_state_update(
                 );
                 controller.request_rewind(rewind_slot, reason)?;
             }
-            Err(IngesterError::SequenceGapDetected(gaps))
+            Err(IngesterError::SequenceGapDetected { gaps, parsed_state })
         }
     }
 }
@@ -121,7 +121,7 @@ pub async fn index_block_batch(
         Some(block_batch.last().unwrap().metadata.slot)
     ) {
         Ok(merged) => merged,
-        Err(SequenceGapError::GapDetected(gaps)) => {
+        Err(SequenceGapError::GapDetected { gaps, parsed_state }) => {
             if let Some(controller) = rewind_controller {
                 let rewind_slot = determine_rewind_slot(&gaps);
                 let reason = format!(
@@ -131,7 +131,7 @@ pub async fn index_block_batch(
                 );
                 controller.request_rewind(rewind_slot, reason)?;
             }
-            return Err(IngesterError::SequenceGapDetected(gaps));
+            return Err(IngesterError::SequenceGapDetected { gaps, parsed_state });
         }
     };
 
@@ -151,9 +151,9 @@ pub async fn index_block_batch_with_infinite_retries(
     loop {
         match index_block_batch(db, &block_batch, rewind_controller).await {
             Ok(()) => return,
-            Err(IngesterError::SequenceGapDetected(_)) => {
+            Err(IngesterError::SequenceGapDetected { gaps, parsed_state: _ }) => {
                 // For sequence gaps, we don't retry - we let the rewind mechanism handle it
-                log::error!("Sequence gap detected in batch, stopping processing to allow rewind");
+                log::error!("Sequence gap detected in batch with {} gaps, stopping processing to allow rewind", gaps.len());
                 return;
             }
             Err(e) => {

src/ingester/parser/state_update.rs

@@ -24,7 +24,10 @@ pub struct SequenceGap {
 
 #[derive(Debug, Clone)]
 pub enum SequenceGapError {
-    GapDetected(Vec<SequenceGap>),
+    GapDetected {
+        gaps: Vec<SequenceGap>,
+        parsed_state: StateUpdate,
+    },
 }
 
 #[derive(BorshDeserialize, BorshSerialize, Debug, Clone, PartialEq, Eq)]
@@ -188,12 +191,15 @@ impl StateUpdate {
                 .extend(update.batch_nullify_context);
         }
 
-        // If gaps were detected, return error
+        // If gaps were detected, return error WITH the parsed state
         if !detected_gaps.is_empty() {
-            return Err(SequenceGapError::GapDetected(detected_gaps));
+            return Err(SequenceGapError::GapDetected {
+                gaps: detected_gaps,
+                parsed_state: merged,
+            });
         }
 
-        // Update highest sequence numbers for each tree
+        // Update highest sequence numbers for each tree (only if no gaps)
         if let Some(slot) = slot {
             for ((tree, _leaf_index), value) in &merged.indexed_merkle_tree_updates {
                 if let Err(e) = TreeInfo::update_highest_seq(tree, value.seq, slot) {

src/ingester/parser/tx_event_parser_v2.rs

@@ -147,7 +147,7 @@ pub fn create_state_update_v2(
 
         state_updates.push(state_update_event);
     }
-
+    // TODO: return the correct gap error.
     let merged = StateUpdate::merge_updates(state_updates)?;
     Ok(merged)
 }

tests/integration_tests/utils.rs

@@ -465,7 +465,7 @@ pub async fn index_multiple_transactions(
         let tx_state_update = parse_transaction(&transaction_info, 0).unwrap();
         state_updates.push(tx_state_update);
     }
-    let state_update = StateUpdate::merge_updates(state_updates);
+    let state_update = StateUpdate::merge_updates(state_updates).expect("Test data should not have sequence gaps");
     persist_state_update_using_connection(db_conn.as_ref(), state_update)
         .await
         .unwrap();