refactor: clear cache and sequence state on gap detection rewinds

refactor: add a dedicated error variant for gap detection

Author: sergeytimoshin

src/ingester/error.rs

@@ -16,6 +16,8 @@ pub enum IngesterError {
     InvalidEvent,
     #[error("Custom error: {0}")]
     CustomError(String),
+    #[error("Gap detected, triggering rewind")]
+    GapDetectedRewind,
 }
 
 impl From<sea_orm::error::DbErr> for IngesterError {

src/ingester/fetchers/poller.rs

@@ -72,11 +72,13 @@ pub fn get_block_poller_stream(
                                 log::error!("Rewinding block stream to {}: {}", to_slot, reason);
                                 // Clear cached blocks
                                 block_cache.clear();
+                                // Clear sequence state to re-learn from rewound point
+                                crate::ingester::gap::clear_sequence_state();
                                 // Reset positions
                                 last_indexed_slot = to_slot - 1;
                                 current_start_slot = to_slot;
                                 rewind_occurred = true;
-                                log::info!("Cleared cache, restarting from slot {}", current_start_slot);
+                                log::info!("Cleared cache and sequence state, restarting from slot {}", current_start_slot);
                                 break;
                             }
                         }

src/ingester/indexer/mod.rs

@@ -112,9 +112,11 @@ pub async fn index_block_stream(
                 }
             }
             Err(e) => {
-                if e.to_string().contains("Gap detection triggered rewind") {
+                if matches!(e, crate::ingester::error::IngesterError::GapDetectedRewind) {
                     // Gap detected, rewind triggered - the slot stream should handle repositioning
                     log::info!("Gap detection triggered rewind");
+                    // Clear sequence state to re-learn from the rewound point
+                    crate::ingester::gap::clear_sequence_state();
                     continue;
                 } else {
                     log::error!("Unexpected error in block processing: {}", e);

src/ingester/mod.rs

@@ -72,9 +72,7 @@ fn derive_block_state_update(
                 ));
             }
             // Return early after requesting rewind - don't continue processing
-            return Err(IngesterError::CustomError(
-                "Gap detection triggered rewind".to_string(),
-            ));
+            return Err(IngesterError::GapDetectedRewind);
         }
     }
 
@@ -203,7 +201,7 @@ pub async fn index_block_batch_with_infinite_retries(
             Ok(()) => return Ok(()),
             Err(e) => {
                 // Check if this is a gap-triggered rewind error
-                if e.to_string().contains("Gap detection triggered rewind") {
+                if matches!(e, IngesterError::GapDetectedRewind) {
                     // Don't retry, propagate the rewind error up
                     return Err(e);
                 }