fix(indexer): harden stall durability and breakpoint guards w.r.t. Copilot review

Cargo.lock

@@ -9460,6 +9460,7 @@ dependencies = [
  "borsh",
  "common",
  "lee",
+ "log",
  "programs",
  "rocksdb",
  "system_accounts",

lez/indexer/core/src/lib.rs

@@ -250,12 +250,21 @@ impl IndexerCore {
                                     "Parked at block {} after {attempts} failed apply attempts: {ingest_err}",
                                     block.header.block_id
                                 );
+                                // The stall must be durable before the cursor moves.
                                 if let Err(err) = self.store.record_stall(
                                     Some(&block.header),
                                     slot,
                                     ingest_err.clone(),
                                 ) {
-                                    warn!("Failed to record stall reason: {err:#}");
+                                    error!(
+                                        "Failed to record stall reason for block {}: {err:#}",
+                                        block.header.block_id
+                                    );
+                                    self.set_status(IndexerSyncStatus::error(format!(
+                                        "store error: {err:#}"
+                                    )));
+                                    had_cycle_error = true;
+                                    break;
                                 }
                                 self.set_status(IndexerSyncStatus::stalled(ingest_err.to_string()));
                                 self.advance_cursor(&mut cursor, slot);

lez/indexer/service/src/service.rs

@@ -197,7 +197,7 @@ impl SubscriptionService {
             );
 
             // Respawn the subscription service loop if it has finished (either with error or panic)
-            if guard.handle.is_finished() {
+            if guard.handle.is_finished() && !self.shutdown.is_cancelled() {
                 drop(guard);
                 let new_parts = Self::spawn_respond_subscribers_loop(
                     self.indexer.clone(),

lez/storage/Cargo.toml

@@ -13,6 +13,7 @@ lee.workspace = true
 
 thiserror.workspace = true
 borsh.workspace = true
+log.workspace = true
 rocksdb.workspace = true
 tempfile.workspace = true
 zstd.workspace = true

lez/storage/src/indexer/mod.rs

@@ -5,6 +5,7 @@ use common::{
     transaction::{LeeTransaction, clock_invocation},
 };
 use lee::{GENESIS_BLOCK_ID, V03State};
+use log::warn;
 use rocksdb::{
     BoundColumnFamily, ColumnFamilyDescriptor, DBWithThreadMode, MultiThreaded, Options,
 };
@@ -159,25 +160,31 @@ impl RocksDBIO {
         }
 
         // walk down to the nearest snapshot that exists
-        let mut breakpoint_id = closest_breakpoint_id(block_id);
+        let target = closest_breakpoint_id(block_id);
+        let mut br_id = target;
         let mut state = loop {
-            match self.get_breakpoint_opt(breakpoint_id)? {
+            match self.get_breakpoint_opt(br_id)? {
                 Some(state) => break state,
-                None if breakpoint_id == 0 => {
+                None if br_id == 0 => {
                     return Err(DbError::db_interaction_error(
                         "Breakpoint 0 is missing".to_owned(),
                     ));
                 }
                 None => {
-                    breakpoint_id = breakpoint_id
+                    br_id = br_id
                         .checked_sub(1)
                         .expect("breakpoint_id > 0 checked above");
                 }
             }
         };
+        if br_id < target {
+            warn!(
+                "Breakpoint {target} missing; replaying from breakpoint {br_id} for block {block_id}"
+            );
+        }
 
         let start = u64::from(BREAKPOINT_INTERVAL)
-            .checked_mul(breakpoint_id)
+            .checked_mul(br_id)
             .expect("Reached maximum breakpoint id");
 
         for block in self.get_block_batch_seq(

lez/storage/src/indexer/tests.rs

@@ -130,6 +130,21 @@ fn one_block_insertion() {
     );
 }
 
+#[test]
+fn put_block_rejects_breakpoint_on_non_boundary_block() {
+    let temp_dir = tempdir().unwrap();
+    let temdir_path = temp_dir.path();
+
+    let dbio = RocksDBIO::open_or_create(temdir_path, &initial_state()).unwrap();
+
+    let block = produce_dummy_block(1, None, vec![]);
+
+    assert!(
+        dbio.put_block(&block, [0; 32], 0, Some(&initial_state()))
+            .is_err()
+    );
+}
+
 #[test]
 fn put_block_records_tip_inscription_slot() {
     let temp_dir = tempdir().unwrap();

lez/storage/src/indexer/write_atomic.rs

@@ -224,13 +224,15 @@ impl RocksDBIO {
         }
 
         if let Some(state) = breakpoint {
-            debug_assert!(
+            if !block
-                block
+                .header
-                    .header
+                .block_id
-                    .block_id
+                .is_multiple_of(BREAKPOINT_INTERVAL.into())
-                    .is_multiple_of(BREAKPOINT_INTERVAL.into()),
+            {
-                "breakpoint snapshot must accompany an interval-boundary block"
+                return Err(DbError::db_interaction_error(
-            );
+                    "Breakpoint snapshot must accompany an interval-boundary block".to_owned(),
+                ));
+            }
             let br_id = block
                 .header
                 .block_id