fix(indexer): retry apply failures before parking

lez/indexer/core/src/block_store.rs

@@ -237,8 +237,9 @@ impl IndexerStore {
     }
 
     /// Validates `block` against the tip and, if it chains, applies it atomically
-    /// (scratch clone, commit only on full success) and advances the tip. On any
-    /// failure records the stall and returns `Parked` without touching state.
+    /// (scratch clone, commit only on full success) and advances the tip.
+    /// Retryable apply failures return `ApplyFailed` without recording a stall
+    /// or touching state; other failures record the stall and return `Parked`.
     pub async fn accept_block(&self, block: &Block, l1_slot: Slot) -> Result<AcceptOutcome> {
         let tip = self.validated_tip()?;
 

lez/indexer/core/src/lib.rs

@@ -10,6 +10,7 @@ use log::{error, info, warn};
 use logos_blockchain_zone_sdk::{
     CommonHttpClient, Slot, ZoneMessage, adapter::NodeHttpClient, indexer::ZoneIndexer,
 };
+use retry::ApplyRetryGate;
 pub use stall_reason::StallReason;
 
 use crate::{
@@ -22,9 +23,13 @@ pub mod block_store;
 pub mod chain_consistency;
 pub mod config;
 pub mod ingest_error;
+mod retry;
 pub mod stall_reason;
 pub mod status;
 
+/// Consecutive failed apply attempts of the same block before parking.
+const APPLY_RETRY_LIMIT: u32 = 3;
+
 #[derive(Clone)]
 pub struct IndexerCore {
     pub zone_indexer: Arc<ZoneIndexer<NodeHttpClient>>,
@@ -167,6 +172,7 @@ impl IndexerCore {
 
         async_stream::stream! {
             let mut cursor = initial_cursor;
+            let mut retry_gate = ApplyRetryGate::new();
 
             if cursor.is_some() {
                 info!("Resuming indexer from cursor {cursor:?}");
@@ -215,6 +221,7 @@ impl IndexerCore {
 
                     match self.store.accept_block(&block, slot).await {
                         Ok(AcceptOutcome::Applied) => {
+                            retry_gate.reset();
                             info!("Indexed L2 block {}", block.header.block_id);
                             self.set_status(IndexerSyncStatus::syncing());
                             self.advance_cursor(&mut cursor, slot);
@@ -237,13 +244,33 @@ impl IndexerCore {
                             self.advance_cursor(&mut cursor, slot);
                         }
                         Ok(AcceptOutcome::ApplyFailed(ingest_err)) => {
-                            error!(
-                                "Apply failed at block {}: {ingest_err}",
-                                block.header.block_id
-                            );
-                            self.set_status(IndexerSyncStatus::error(ingest_err.to_string()));
-                            had_cycle_error = true;
-                            break;
+                            let attempts = retry_gate.register_failure(block.header.block_id);
+                            if attempts >= APPLY_RETRY_LIMIT {
+                                error!(
+                                    "Parked at block {} after {attempts} failed apply attempts: {ingest_err}",
+                                    block.header.block_id
+                                );
+                                if let Err(err) = self.store.record_stall(
+                                    Some(&block.header),
+                                    slot,
+                                    ingest_err.clone(),
+                                ) {
+                                    warn!("Failed to record stall reason: {err:#}");
+                                }
+                                self.set_status(IndexerSyncStatus::stalled(ingest_err.to_string()));
+                                self.advance_cursor(&mut cursor, slot);
+                                retry_gate.reset();
+                            } else {
+                                error!(
+                                    "Failed to apply block {} (attempt {attempts}/{APPLY_RETRY_LIMIT}), will retry: {ingest_err}",
+                                    block.header.block_id
+                                );
+                                self.set_status(IndexerSyncStatus::error(format!(
+                                    "apply failed, retrying: {ingest_err}"
+                                )));
+                                had_cycle_error = true;
+                                break;
+                            }
                         }
                         Err(err) => {
                             // Infrastructure error (DB read/write), not a bad block.

lez/indexer/core/src/retry.rs

@@ -0,0 +1,64 @@
+//! Retry gate for possibly-transient block-apply failures.
+
+/// Counts consecutive apply failures per block id so the ingest loop can
+/// retry before parking.
+///
+/// A failure streak is keyed to one block id: a failure of a different block
+/// starts a fresh streak. Reset only on a genuinely applied block — the
+/// `AlreadyApplied` replay of the prefix after a retry cycle must not clear
+/// the failing block's streak.
+pub struct ApplyRetryGate {
+    failing: Option<(u64, u32)>,
+}
+
+impl ApplyRetryGate {
+    #[must_use]
+    pub const fn new() -> Self {
+        Self { failing: None }
+    }
+
+    /// Registers a failed apply of `block_id`; returns its consecutive
+    /// attempt count.
+    pub const fn register_failure(&mut self, block_id: u64) -> u32 {
+        let attempts = match self.failing {
+            Some((id, attempts)) if id == block_id => attempts.saturating_add(1),
+            _ => 1,
+        };
+        self.failing = Some((block_id, attempts));
+        attempts
+    }
+
+    /// Clears the streak; call when a block actually applies.
+    pub const fn reset(&mut self) {
+        self.failing = None;
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn counts_consecutive_failures_of_same_block() {
+        let mut gate = ApplyRetryGate::new();
+        assert_eq!(gate.register_failure(7), 1);
+        assert_eq!(gate.register_failure(7), 2);
+        assert_eq!(gate.register_failure(7), 3);
+    }
+
+    #[test]
+    fn different_block_starts_fresh_streak() {
+        let mut gate = ApplyRetryGate::new();
+        gate.register_failure(7);
+        gate.register_failure(7);
+        assert_eq!(gate.register_failure(8), 1);
+    }
+
+    #[test]
+    fn reset_clears_streak() {
+        let mut gate = ApplyRetryGate::new();
+        gate.register_failure(7);
+        gate.reset();
+        assert_eq!(gate.register_failure(7), 1);
+    }
+}