From 37aaa3100a38e231a5f3861fcd2a4b0beaab9e53 Mon Sep 17 00:00:00 2001 From: erhant Date: Thu, 9 Jul 2026 14:19:12 +0300 Subject: [PATCH] fix(indexer): retry apply failures before parking --- lez/indexer/core/src/block_store.rs | 5 ++- lez/indexer/core/src/lib.rs | 41 ++++++++++++++---- lez/indexer/core/src/retry.rs | 64 +++++++++++++++++++++++++++++ 3 files changed, 101 insertions(+), 9 deletions(-) create mode 100644 lez/indexer/core/src/retry.rs diff --git a/lez/indexer/core/src/block_store.rs b/lez/indexer/core/src/block_store.rs index db3bed79..5298b96d 100644 --- a/lez/indexer/core/src/block_store.rs +++ b/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 { let tip = self.validated_tip()?; diff --git a/lez/indexer/core/src/lib.rs b/lez/indexer/core/src/lib.rs index f748bf75..6e7ef6f8 100644 --- a/lez/indexer/core/src/lib.rs +++ b/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>, @@ -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. diff --git a/lez/indexer/core/src/retry.rs b/lez/indexer/core/src/retry.rs new file mode 100644 index 00000000..d7c5f5a2 --- /dev/null +++ b/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); + } +}