From b2b58da51971e3c1fb9113769a77eb235ed21f2e Mon Sep 17 00:00:00 2001 From: moudyellaz Date: Thu, 30 Jul 2026 06:25:57 +0200 Subject: [PATCH] fix(indexer): advance the read cursor only on a slot boundary --- lez/indexer/core/src/lib.rs | 110 ++++++++++++++++++++++++++++++++---- 1 file changed, 99 insertions(+), 11 deletions(-) diff --git a/lez/indexer/core/src/lib.rs b/lez/indexer/core/src/lib.rs index fe841ebc..ccf8dbe7 100644 --- a/lez/indexer/core/src/lib.rs +++ b/lez/indexer/core/src/lib.rs @@ -29,6 +29,16 @@ pub mod status; /// Consecutive failed apply attempts of the same block before parking. const APPLY_RETRY_LIMIT: u32 = 3; +/// Which slot the ingest loop is currently inside, so the read cursor only ever +/// moves on a slot boundary. +/// +/// One L1 slot can carry several L2 blocks, and the channel stream resumes +/// *after* the stored slot. Advancing the cursor as each block is handled would +/// therefore put a later block in the same slot beyond the cursor whenever a +/// pass ends early, and nothing would ever read it again. +#[derive(Default)] +struct SlotProgress(Option); + #[derive(Clone)] pub struct IndexerCore { pub zone_indexer: Arc>, @@ -42,6 +52,23 @@ pub struct IndexerCore { pub verifier: Option, } +impl SlotProgress { + /// Records that a message from `slot` is being handled, returning the slot + /// that just completed, if this message begins a new one. + fn enter(&mut self, slot: Slot) -> Option { + if self.0 == Some(slot) { + return None; + } + self.0.replace(slot) + } + + /// The slot in progress when the stream drained cleanly, which is therefore + /// complete. Not called when a pass ends early: that slot must be re-read. + const fn drained(self) -> Option { + self.0 + } +} + impl IndexerCore { /// Builds the core, then verifies the stored chain matches the channel's by /// re-reading the channel at the stored tip's position. @@ -264,8 +291,20 @@ impl IndexerCore { let mut announced_syncing = false; let mut had_cycle_error = false; + // The slot being consumed: every message of it seen so far is + // handled, but another may follow, so the cursor may not move + // onto it yet. One L1 slot can carry several L2 blocks, and the + // stream resumes *after* the stored slot, so advancing inside a + // slot would put a later message in it beyond the cursor + // for ever if this pass ends early. + let mut in_progress = SlotProgress::default(); while let Some((msg, slot)) = stream.next().await { + // A message from a later slot means the previous one is complete. + if let Some(done) = in_progress.enter(slot) { + self.advance_cursor(&mut cursor, done); + } + if !announced_syncing { self.set_status(IndexerSyncStatus::syncing()); announced_syncing = true; @@ -286,17 +325,18 @@ impl IndexerCore { break; } // L1 proceeds regardless - self.advance_cursor(&mut cursor, slot); continue; } }; - // Option B: re-derive and verify every cross-zone dispatch - // before applying the block. A forged dispatch halts ingestion - // rather than persisting an invalid state; a replay is accepted - // since the inbox no-ops it on chain. The verified keys are - // marked seen only once the block applies (below), so a block - // that does not apply cannot poison the seen-set. + // Re-derive and verify every cross-zone dispatch the block + // carries before applying it, so the destination never trusts + // a dispatch just because a sequencer signed the block: a + // forged one halts ingestion rather than persisting invalid + // state, while a replay is accepted since the inbox no-ops it + // on chain. The verified keys are marked seen only once the + // block applies (below), so a block that does not apply + // cannot poison the seen-set. let verified_keys = match &self.verifier { Some(verifier) => match verifier.verify_block(&block).await { Ok(keys) => keys, @@ -334,7 +374,6 @@ impl IndexerCore { retry_gate.reset(); info!("Indexed L2 block {}", block.header.block_id); self.set_status(IndexerSyncStatus::syncing()); - self.advance_cursor(&mut cursor, slot); yield Ok(block); } Ok(AcceptOutcome::AlreadyApplied) => { @@ -342,7 +381,6 @@ impl IndexerCore { "Skipping already-applied block {}", block.header.block_id ); - self.advance_cursor(&mut cursor, slot); } Ok(AcceptOutcome::Parked(ingest_err)) => { error!( @@ -351,7 +389,6 @@ impl IndexerCore { ); self.set_status(IndexerSyncStatus::stalled(ingest_err.to_string())); // L1 proceeds regardless - self.advance_cursor(&mut cursor, slot); } Ok(AcceptOutcome::RetryableFailure(ingest_err)) => { let attempts = retry_gate.register_failure(block.header.block_id); @@ -377,7 +414,6 @@ impl IndexerCore { break; } self.set_status(IndexerSyncStatus::stalled(ingest_err.to_string())); - self.advance_cursor(&mut cursor, slot); retry_gate.reset(); } else { error!( @@ -408,10 +444,17 @@ impl IndexerCore { } if had_cycle_error { + // The slot in progress is not finished, so the cursor stays + // below it and the next pass re-reads it whole. tokio::time::sleep(poll_interval).await; continue; } + // The stream drained cleanly, so the slot in progress completed too. + if let Some(done) = in_progress.drained() { + self.advance_cursor(&mut cursor, done); + } + // Stream drained. Stay Stalled if parked; otherwise we are caught up. // A store error here must not be collapsed to "no stall recorded": // that would wrongly flip us to caught-up, so we log and hold state. @@ -438,6 +481,51 @@ mod tests { use super::*; use crate::config::{ChannelId, ClientConfig, IndexerConfig}; + /// The cursor must not move while more of the same slot may still arrive. + /// + /// Two L2 blocks in one L1 slot: the first applies, the second stalls on an + /// unavailable peer and the pass retries. If handling the first had advanced + /// the cursor onto the slot, the retry would resume past it and the second + /// block would never be read again, silently losing whatever it carried. + #[test] + fn a_slot_is_only_left_behind_once_it_is_finished() { + let mut progress = SlotProgress::default(); + let slot = Slot::from(7); + + assert_eq!( + progress.enter(slot), + None, + "nothing precedes the first slot" + ); + assert_eq!( + progress.enter(slot), + None, + "a second message in the same slot must not release it" + ); + + // The pass ends early here, so `drained` is never called and the cursor + // is still below slot 7: the next pass re-reads it whole. + } + + #[test] + fn a_completed_slot_is_released_when_the_next_one_starts() { + let mut progress = SlotProgress::default(); + + assert_eq!(progress.enter(Slot::from(3)), None); + assert_eq!(progress.enter(Slot::from(3)), None); + assert_eq!( + progress.enter(Slot::from(4)), + Some(Slot::from(3)), + "slot 3 is complete once a message from slot 4 arrives" + ); + assert_eq!(progress.drained(), Some(Slot::from(4))); + } + + #[test] + fn draining_an_untouched_stream_releases_nothing() { + assert_eq!(SlotProgress::default().drained(), None); + } + fn unreachable_core(dir: &std::path::Path) -> IndexerCore { let config = IndexerConfig { consensus_info_polling_interval: Duration::from_secs(1),