fix(indexer): advance the read cursor only on a slot boundary

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<Slot>);
+
 #[derive(Clone)]
 pub struct IndexerCore {
     pub zone_indexer: Arc<ZoneIndexer<NodeHttpClient>>,
@@ -42,6 +52,23 @@ pub struct IndexerCore {
     pub verifier: Option<CrossZoneVerifier>,
 }
 
+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<Slot> {
+        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<Slot> {
+        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),