Merge 3bfa210eac8942a39393d79574b7e2e587443a56 into 9b2b4a3d8ca714fa089df634a230d1e0764dcdd7

Cargo.lock

@@ -9463,6 +9463,7 @@ dependencies = [
  "borsh",
  "common",
  "lee",
+ "log",
  "programs",
  "rocksdb",
  "system_accounts",
@@ -9760,6 +9761,7 @@ dependencies = [
  "tempfile",
  "testcontainers",
  "tokio",
+ "tokio-util",
  "url",
  "wallet",
 ]

lez/indexer/core/src/block_store.rs

@@ -11,7 +11,7 @@ use lee_core::BlockId;
 use log::warn;
 use logos_blockchain_core::header::HeaderId;
 use logos_blockchain_zone_sdk::Slot;
-use storage::indexer::RocksDBIO;
+use storage::{BREAKPOINT_INTERVAL, indexer::RocksDBIO};
 use tokio::sync::RwLock;
 
 use crate::{ingest_error::BlockIngestError, stall_reason::StallReason};
@@ -29,6 +29,11 @@ pub enum AcceptOutcome {
     AlreadyApplied,
     /// Did not chain or failed to apply; tip stays frozen, stall recorded.
     Parked(BlockIngestError),
+    /// Chained but failed to apply, possibly transiently
+    /// ([`BlockIngestError::is_retryable`]); nothing recorded, tip and state
+    /// untouched. The caller retries and parks via
+    /// [`IndexerStore::record_stall`] once it gives up.
+    ApplyFailed(BlockIngestError),
 }
 
 #[derive(Clone)]
@@ -232,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()?;
 
@@ -254,14 +260,22 @@ impl IndexerStore {
         // TODO: we use scratch state to be atomic, but need to revisit how expensive a clone is
         let mut scratch = self.current_state.read().await.clone();
         if let Err(err) = apply_block_to_scratch(block, &mut scratch) {
+            if err.is_retryable() {
+                return Ok(AcceptOutcome::ApplyFailed(err));
+            }
             self.record_stall(Some(&block.header), l1_slot, err.clone())?;
             return Ok(AcceptOutcome::Parked(err));
         }
 
         let mut stored = block.clone();
         stored.bedrock_status = BedrockStatus::Finalized;
+        let breakpoint = block
+            .header
+            .block_id
+            .is_multiple_of(BREAKPOINT_INTERVAL.into())
+            .then_some(&scratch);
         self.dbio
-            .put_block(&stored, [0_u8; 32], l1_slot.into_inner())
+            .put_block(&stored, [0_u8; 32], l1_slot.into_inner(), breakpoint)
             .context("Failed to persist accepted block")?;
 
         // Commit in-memory state (infallible) only after the DB write succeeded.
@@ -824,4 +838,91 @@ mod accept_tests {
             "a benign re-delivery must not park the indexer"
         );
     }
+
+    #[tokio::test]
+    async fn accept_block_snapshots_state_at_breakpoint_interval() {
+        use testnet_initial_state::initial_pub_accounts_private_keys;
+
+        let dir = tempfile::tempdir().expect("tempdir");
+        let store = IndexerStore::open_db(dir.path()).expect("open store");
+
+        let accounts = initial_pub_accounts_private_keys();
+        let from = accounts[0].account_id;
+        let to = accounts[1].account_id;
+        let sign_key = accounts[0].pub_sign_key.clone();
+
+        let genesis = produce_dummy_block(1, None, vec![]);
+        assert!(matches!(
+            store.accept_block(&genesis, Slot::from(0)).await.unwrap(),
+            AcceptOutcome::Applied
+        ));
+        let mut prev_hash = genesis.header.hash;
+
+        // Blocks 2..=101: one transfer of 1 each; block 100 crosses the interval.
+        for i in 0..100_u64 {
+            let tx = common::test_utils::create_transaction_native_token_transfer(
+                from,
+                i.into(),
+                to,
+                1,
+                &sign_key,
+            );
+            let block = produce_dummy_block(i + 2, Some(prev_hash), vec![tx]);
+            prev_hash = block.header.hash;
+            assert!(matches!(
+                store.accept_block(&block, Slot::from(0)).await.unwrap(),
+                AcceptOutcome::Applied
+            ));
+        }
+
+        // Snapshot at block 100 = genesis + 99 transfers, written with the block.
+        let bp1 = store.dbio.get_breakpoint(1).expect("breakpoint 1 present");
+        assert_eq!(bp1.get_account_by_id(from).balance, 10000 - 99);
+        assert_eq!(store.dbio.get_meta_last_breakpoint_id().unwrap(), Some(1));
+
+        // The #605 restart: reopening past the boundary must work.
+        drop(store);
+        let reopened = IndexerStore::open_db(dir.path()).expect("reopen");
+        assert_eq!(reopened.last_block().unwrap(), Some(101));
+    }
+
+    #[tokio::test]
+    async fn transient_apply_failure_returns_apply_failed_without_stall() {
+        use testnet_initial_state::initial_pub_accounts_private_keys;
+
+        let dir = tempfile::tempdir().expect("tempdir");
+        let store = IndexerStore::open_db(dir.path()).expect("open store");
+
+        let accounts = initial_pub_accounts_private_keys();
+        let from = accounts[0].account_id;
+        let to = accounts[1].account_id;
+        let sign_key = accounts[0].pub_sign_key.clone();
+
+        let genesis = produce_dummy_block(1, None, vec![]);
+        store
+            .accept_block(&genesis, Slot::from(0))
+            .await
+            .expect("accept genesis");
+
+        // Overdraft: rejected during execution → StateTransition → retryable.
+        let tx = common::test_utils::create_transaction_native_token_transfer(
+            from,
+            0,
+            to,
+            1_000_000_000,
+            &sign_key,
+        );
+        let block = produce_dummy_block(2, Some(genesis.header.hash), vec![tx]);
+        let outcome = store.accept_block(&block, Slot::from(0)).await.unwrap();
+
+        assert!(matches!(
+            outcome,
+            AcceptOutcome::ApplyFailed(BlockIngestError::StateTransition { .. })
+        ));
+        assert!(
+            store.get_stall_reason().unwrap().is_none(),
+            "retryable failure must not persist a stall"
+        );
+        assert_eq!(store.get_last_block_id().unwrap(), Some(1), "tip frozen");
+    }
 }

lez/indexer/core/src/ingest_error.rs

@@ -40,6 +40,19 @@ pub enum BlockIngestError {
     },
 }
 
+impl BlockIngestError {
+    /// Whether the failure may be transient rather than a property of the block.
+    ///
+    /// FIXME: `StateTransition` is too coarse — its `reason` string mixes genuine
+    /// state-transition rejections with infra failures (risc0 executor teardown,
+    /// storage errors). Once it carries a structured cause, narrow this so only
+    /// infra failures retry.
+    #[must_use]
+    pub const fn is_retryable(&self) -> bool {
+        matches!(self, Self::StateTransition { .. })
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

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);
@@ -236,6 +243,44 @@ impl IndexerCore {
                             // L1 proceeds regardless
                             self.advance_cursor(&mut cursor, slot);
                         }
+                        Ok(AcceptOutcome::ApplyFailed(ingest_err)) => {
+                            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
+                                );
+                                // The stall must be durable before the cursor moves.
+                                if let Err(err) = self.store.record_stall(
+                                    Some(&block.header),
+                                    slot,
+                                    ingest_err.clone(),
+                                ) {
+                                    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);
+                                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.
                             // will re-poll from the same cursor next cycle.

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);
+    }
+}

lez/indexer/service/src/lib.rs

@@ -5,6 +5,7 @@ pub use indexer_core::config::*;
 use indexer_service_rpc::RpcServer as _;
 use jsonrpsee::server::{Server, ServerHandle};
 use log::{error, info};
+use tokio_util::sync::CancellationToken;
 
 pub mod service;
 
@@ -69,9 +70,10 @@ pub async fn run_server(
     config: IndexerConfig,
     storage_dir: &Path,
     port: u16,
+    shutdown: CancellationToken,
 ) -> Result<IndexerHandle> {
     #[cfg(feature = "mock-responses")]
-    let _ = (config, storage_dir);
+    let _ = (config, storage_dir, shutdown);
 
     let server = Server::builder()
         .build(SocketAddr::from(([0, 0, 0, 0], port)))
@@ -86,7 +88,7 @@ pub async fn run_server(
 
     #[cfg(not(feature = "mock-responses"))]
     let handle = {
-        let service = service::IndexerService::new(config, storage_dir)
+        let service = service::IndexerService::new(config, storage_dir, shutdown)
             .await
             .context("Failed to initialize indexer service")?;
         server.start(service.into_rpc())

lez/indexer/service/src/main.rs

@@ -34,7 +34,9 @@ async fn main() -> Result<()> {
     let cancellation_token = listen_for_shutdown_signal();
 
     let config = indexer_service::IndexerConfig::from_path(&config_path)?;
-    let indexer_handle = indexer_service::run_server(config, data_dir.as_path(), port).await?;
+    let indexer_handle =
+        indexer_service::run_server(config, data_dir.as_path(), port, cancellation_token.clone())
+            .await?;
 
     tokio::select! {
         () = cancellation_token.cancelled() => {

lez/indexer/service/src/service.rs

@@ -2,7 +2,7 @@ use std::{path::Path, pin::pin, sync::Arc};
 
 use anyhow::{Context as _, Result, bail};
 use arc_swap::ArcSwap;
-use futures::{StreamExt as _, never::Never};
+use futures::StreamExt as _;
 use indexer_core::{IndexerCore, config::IndexerConfig};
 use indexer_service_protocol::{
     Account, AccountId, Block, BlockId, HashType, IndexerStatus, Transaction,
@@ -14,6 +14,7 @@ use jsonrpsee::{
 };
 use log::{debug, error, info, warn};
 use tokio::sync::mpsc::UnboundedSender;
+use tokio_util::sync::CancellationToken;
 
 pub struct IndexerService {
     subscription_service: SubscriptionService,
@@ -21,9 +22,13 @@ pub struct IndexerService {
 }
 
 impl IndexerService {
-    pub async fn new(config: IndexerConfig, storage_dir: &Path) -> Result<Self> {
+    pub async fn new(
+        config: IndexerConfig,
+        storage_dir: &Path,
+        shutdown: CancellationToken,
+    ) -> Result<Self> {
         let indexer = IndexerCore::new(config, storage_dir).await?;
-        let subscription_service = SubscriptionService::spawn_new(indexer.clone());
+        let subscription_service = SubscriptionService::spawn_new(indexer.clone(), shutdown);
 
         Ok(Self {
             subscription_service,
@@ -170,15 +175,17 @@ impl indexer_service_rpc::RpcServer for IndexerService {
 struct SubscriptionService {
     parts: ArcSwap<SubscriptionLoopParts>,
     indexer: IndexerCore,
+    shutdown: CancellationToken,
 }
 
 impl SubscriptionService {
-    pub fn spawn_new(indexer: IndexerCore) -> Self {
-        let parts = Self::spawn_respond_subscribers_loop(indexer.clone());
+    pub fn spawn_new(indexer: IndexerCore, shutdown: CancellationToken) -> Self {
+        let parts = Self::spawn_respond_subscribers_loop(indexer.clone(), shutdown.clone());
 
         Self {
             parts: ArcSwap::new(Arc::new(parts)),
             indexer,
+            shutdown,
         }
     }
 
@@ -190,14 +197,18 @@ 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());
+                let new_parts = Self::spawn_respond_subscribers_loop(
+                    self.indexer.clone(),
+                    self.shutdown.clone(),
+                );
                 let old_handle_and_sender = self.parts.swap(Arc::new(new_parts));
                 let old_parts = Arc::into_inner(old_handle_and_sender)
                     .expect("There should be no other references to the old handle and sender");
 
                 match old_parts.handle.await {
+                    Ok(Ok(())) => {}
                     Ok(Err(err)) => {
                         error!(
                             "Subscription service loop has unexpectedly finished with error: {err:#}"
@@ -215,7 +226,10 @@ impl SubscriptionService {
         Ok(())
     }
 
-    fn spawn_respond_subscribers_loop(indexer: IndexerCore) -> SubscriptionLoopParts {
+    fn spawn_respond_subscribers_loop(
+        indexer: IndexerCore,
+        shutdown: CancellationToken,
+    ) -> SubscriptionLoopParts {
         let (new_subscription_sender, mut sub_receiver) =
             tokio::sync::mpsc::unbounded_channel::<Subscription<BlockId>>();
 
@@ -231,6 +245,10 @@ impl SubscriptionService {
                 )]
                 loop {
                     tokio::select! {
+                        () = shutdown.cancelled() => {
+                            info!("Shutdown requested; stopping block ingestion");
+                            return Ok(());
+                        }
                         sub = sub_receiver.recv() => {
                             let Some(subscription) = sub else {
                                 bail!("Subscription receiver closed unexpectedly");
@@ -259,10 +277,11 @@ impl SubscriptionService {
                     }
                 }
             };
-            let res: anyhow::Result<futures::never::Never> = run_loop.await;
-            let Err(err) = res;
-            error!("Subscription service loop has unexpectedly finished with error: {err:#?}");
-            Err(err)
+            let res: anyhow::Result<()> = run_loop.await;
+            if let Err(err) = &res {
+                error!("Subscription service loop has unexpectedly finished with error: {err:#?}");
+            }
+            res
         });
         SubscriptionLoopParts {
             handle,
@@ -278,7 +297,7 @@ impl Drop for SubscriptionService {
 }
 
 struct SubscriptionLoopParts {
-    handle: tokio::task::JoinHandle<Result<Never>>,
+    handle: tokio::task::JoinHandle<Result<()>>,
     new_subscription_sender: UnboundedSender<Subscription<BlockId>>,
 }
 

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
 

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,
 };
@@ -88,9 +89,11 @@ impl RocksDBIO {
 
         let dbio = Self { db };
 
-        // First breakpoint setup
-        dbio.put_breakpoint(0, initial_state)?;
-        dbio.put_meta_last_breakpoint_id(0)?;
+        // Seed the genesis snapshot once; reopening must not clobber breakpoint meta.
+        if dbio.get_meta_last_breakpoint_id()?.is_none() {
+            dbio.put_breakpoint(0, initial_state)?;
+            dbio.put_meta_last_breakpoint_id(0)?;
+        }
 
         Ok(dbio)
     }
@@ -156,8 +159,33 @@ impl RocksDBIO {
             ));
         }
 
-        let br_id = closest_breakpoint_id(block_id);
-        let mut state = self.get_breakpoint(br_id)?;
+        // walk down to the nearest snapshot that exists
+        //
+        // NOTE: we do not use `get_meta_last_breakpoint_id` here because
+        // it may be stale if the last breakpoint was deleted, and simply
+        // computing it from `block_id` is enough
+        let target = closest_breakpoint_id(block_id);
+        let mut br_id = target;
+        let mut state = loop {
+            match self.get_breakpoint_opt(br_id)? {
+                Some(state) => break state,
+                None if br_id == 0 => {
+                    return Err(DbError::db_interaction_error(
+                        "Breakpoint 0 is missing".to_owned(),
+                    ));
+                }
+                None => {
+                    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(br_id)

lez/storage/src/indexer/read_once.rs

@@ -55,6 +55,11 @@ impl RocksDBIO {
         self.get::<BreakpointCellOwned>(br_id).map(|cell| cell.0)
     }
 
+    pub fn get_breakpoint_opt(&self, br_id: u64) -> DbResult<Option<V03State>> {
+        self.get_opt::<BreakpointCellOwned>(br_id)
+            .map(|opt| opt.map(|cell| cell.0))
+    }
+
     // Mappings
 
     pub fn get_block_id_by_hash(&self, hash: [u8; 32]) -> DbResult<Option<u64>> {

lez/storage/src/indexer/tests.rs

@@ -87,7 +87,7 @@ fn one_block_insertion() {
     let dbio = RocksDBIO::open_or_create(temdir_path, &initial_state()).unwrap();
 
     let genesis_block = genesis_block();
-    dbio.put_block(&genesis_block, [0; 32], 0).unwrap();
+    dbio.put_block(&genesis_block, [0; 32], 0, None).unwrap();
 
     let prev_hash = genesis_block.header.hash;
     let from = acc1();
@@ -98,7 +98,7 @@ fn one_block_insertion() {
         common::test_utils::create_transaction_native_token_transfer(from, 0, to, 1, &sign_key);
     let block = produce_dummy_block(2, Some(prev_hash), vec![transfer_tx]);
 
-    dbio.put_block(&block, [1; 32], 0).unwrap();
+    dbio.put_block(&block, [1; 32], 0, None).unwrap();
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
     let first_id = dbio.get_meta_first_block_id_in_db().unwrap();
@@ -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();
@@ -138,30 +153,33 @@ fn put_block_records_tip_inscription_slot() {
     assert_eq!(dbio.get_meta_tip_slot_in_db().unwrap(), None);
 
     let genesis_block = genesis_block();
-    dbio.put_block(&genesis_block, [0; 32], 1_000).unwrap();
+    dbio.put_block(&genesis_block, [0; 32], 1_000, None)
+        .unwrap();
     assert_eq!(dbio.get_meta_tip_slot_in_db().unwrap(), Some(1_000));
 
     let block = produce_dummy_block(2, Some(genesis_block.header.hash), vec![]);
-    dbio.put_block(&block, [1; 32], 1_005).unwrap();
+    dbio.put_block(&block, [1; 32], 1_005, None).unwrap();
     assert_eq!(dbio.get_meta_tip_slot_in_db().unwrap(), Some(1_005));
 
     // Re-inserting a block at/below the tip must not move the tip slot.
-    dbio.put_block(&genesis_block, [0; 32], 1_010).unwrap();
+    dbio.put_block(&genesis_block, [0; 32], 1_010, None)
+        .unwrap();
     assert_eq!(dbio.get_meta_tip_slot_in_db().unwrap(), Some(1_005));
 }
 
 #[test]
-fn new_breakpoint() {
+fn put_block_stores_breakpoint_in_same_batch() {
     let temp_dir = tempdir().unwrap();
-    let temdir_path = temp_dir.path();
-
-    let dbio = RocksDBIO::open_or_create(temdir_path, &initial_state()).unwrap();
+    let dbio = RocksDBIO::open_or_create(temp_dir.path(), &initial_state()).unwrap();
 
     let from = acc1();
     let to = acc2();
     let sign_key = acc1_sign_key();
 
-    for i in 1..=BREAKPOINT_INTERVAL + 1 {
+    // Chain blocks 1..=BREAKPOINT_INTERVAL; only the boundary block carries a
+    // snapshot. A recognizable marker state (the initial one) proves put_block
+    // stores the caller's snapshot verbatim rather than recomputing it.
+    for i in 1..=BREAKPOINT_INTERVAL {
         let prev_hash = dbio.get_meta_last_block_id_in_db().unwrap().map(|last_id| {
             let last_block = dbio.get_block(last_id).unwrap().unwrap();
             last_block.header.hash
@@ -175,42 +193,59 @@ fn new_breakpoint() {
             &sign_key,
         );
         let block = produce_dummy_block(i.into(), prev_hash, vec![transfer_tx]);
-        dbio.put_block(&block, [i; 32], 0).unwrap();
+
+        let marker = (i == BREAKPOINT_INTERVAL).then(initial_state);
+        dbio.put_block(&block, [i; 32], 0, marker.as_ref()).unwrap();
     }
 
-    let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
-    let first_id = dbio.get_meta_first_block_id_in_db().unwrap();
-    let is_first_set = dbio.get_meta_is_first_block_set().unwrap();
-    let last_br_id = dbio.get_meta_last_breakpoint_id().unwrap();
-    let last_block = dbio.get_block(last_id).unwrap().unwrap();
-    let prev_breakpoint = dbio.get_breakpoint(0).unwrap();
-    let breakpoint = dbio.get_breakpoint(1).unwrap();
-    let final_state = dbio.final_state().unwrap();
+    assert_eq!(dbio.get_meta_last_breakpoint_id().unwrap(), Some(1));
+    let bp1 = dbio.get_breakpoint(1).unwrap();
+    assert_eq!(bp1.get_account_by_id(acc1()).balance, 10000);
+    assert_eq!(bp1.get_account_by_id(acc2()).balance, 20000);
+    // Non-boundary blocks passed None: breakpoint 0 must be the only other one.
+    assert_eq!(
+        dbio.get_breakpoint(0)
+            .unwrap()
+            .get_account_by_id(acc1())
+            .balance,
+        10000
+    );
+}
 
-    assert_eq!(last_id, 101);
-    assert_eq!(first_id, Some(1));
-    assert!(is_first_set);
-    assert_eq!(last_br_id, Some(1));
-    assert_ne!(last_block.header.hash, genesis_block().header.hash);
+#[test]
+fn state_replay_falls_back_over_missing_breakpoints() {
+    let temp_dir = tempdir().unwrap();
+    let dbio = RocksDBIO::open_or_create(temp_dir.path(), &initial_state()).unwrap();
+
+    let from = acc1();
+    let to = acc2();
+    let sign_key = acc1_sign_key();
+
+    for i in 1..=u64::from(BREAKPOINT_INTERVAL) + 1 {
+        let prev_hash = dbio.get_meta_last_block_id_in_db().unwrap().map(|last_id| {
+            let last_block = dbio.get_block(last_id).unwrap().unwrap();
+            last_block.header.hash
+        });
+        let transfer_tx = common::test_utils::create_transaction_native_token_transfer(
+            from,
+            (i - 1).into(),
+            to,
+            1,
+            &sign_key,
+        );
+        let block = produce_dummy_block(i, prev_hash, vec![transfer_tx]);
+        dbio.put_block(&block, [0; 32], 0, None).unwrap();
+    }
+
+    assert!(dbio.get_breakpoint_opt(1).unwrap().is_none());
+    let final_state = dbio.final_state().unwrap();
     assert_eq!(
-        prev_breakpoint.get_account_by_id(acc1()).balance
-            - final_state.get_account_by_id(acc1()).balance,
-        101
+        10000 - final_state.get_account_by_id(acc1()).balance,
+        u128::from(BREAKPOINT_INTERVAL) + 1
     );
     assert_eq!(
-        final_state.get_account_by_id(acc2()).balance
-            - prev_breakpoint.get_account_by_id(acc2()).balance,
-        101
-    );
-    assert_eq!(
-        breakpoint.get_account_by_id(acc1()).balance
-            - final_state.get_account_by_id(acc1()).balance,
-        1
-    );
-    assert_eq!(
-        final_state.get_account_by_id(acc2()).balance
-            - breakpoint.get_account_by_id(acc2()).balance,
-        1
+        final_state.get_account_by_id(acc2()).balance - 20000,
+        u128::from(BREAKPOINT_INTERVAL) + 1
     );
 }
 
@@ -231,7 +266,7 @@ fn simple_maps() {
 
     let control_hash1 = block.header.hash;
 
-    dbio.put_block(&block, [1; 32], 0).unwrap();
+    dbio.put_block(&block, [1; 32], 0, None).unwrap();
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
     let last_block = dbio.get_block(last_id).unwrap().unwrap();
@@ -243,7 +278,7 @@ fn simple_maps() {
 
     let control_hash2 = block.header.hash;
 
-    dbio.put_block(&block, [2; 32], 0).unwrap();
+    dbio.put_block(&block, [2; 32], 0, None).unwrap();
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
     let last_block = dbio.get_block(last_id).unwrap().unwrap();
@@ -255,7 +290,7 @@ fn simple_maps() {
     let control_tx_hash1 = transfer_tx.hash();
 
     let block = produce_dummy_block(3, Some(prev_hash), vec![transfer_tx]);
-    dbio.put_block(&block, [3; 32], 0).unwrap();
+    dbio.put_block(&block, [3; 32], 0, None).unwrap();
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
     let last_block = dbio.get_block(last_id).unwrap().unwrap();
@@ -267,7 +302,7 @@ fn simple_maps() {
     let control_tx_hash2 = transfer_tx.hash();
 
     let block = produce_dummy_block(4, Some(prev_hash), vec![transfer_tx]);
-    dbio.put_block(&block, [4; 32], 0).unwrap();
+    dbio.put_block(&block, [4; 32], 0, None).unwrap();
 
     let control_block_id1 = dbio.get_block_id_by_hash(control_hash1.0).unwrap().unwrap();
     let control_block_id2 = dbio.get_block_id_by_hash(control_hash2.0).unwrap().unwrap();
@@ -304,7 +339,7 @@ fn block_batch() {
     let block = produce_dummy_block(1, None, vec![transfer_tx]);
 
     block_res.push(block.clone());
-    dbio.put_block(&block, [1; 32], 0).unwrap();
+    dbio.put_block(&block, [1; 32], 0, None).unwrap();
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
     let last_block = dbio.get_block(last_id).unwrap().unwrap();
@@ -315,7 +350,7 @@ fn block_batch() {
     let block = produce_dummy_block(2, Some(prev_hash), vec![transfer_tx]);
 
     block_res.push(block.clone());
-    dbio.put_block(&block, [2; 32], 0).unwrap();
+    dbio.put_block(&block, [2; 32], 0, None).unwrap();
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
     let last_block = dbio.get_block(last_id).unwrap().unwrap();
@@ -326,7 +361,7 @@ fn block_batch() {
 
     let block = produce_dummy_block(3, Some(prev_hash), vec![transfer_tx]);
     block_res.push(block.clone());
-    dbio.put_block(&block, [3; 32], 0).unwrap();
+    dbio.put_block(&block, [3; 32], 0, None).unwrap();
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
     let last_block = dbio.get_block(last_id).unwrap().unwrap();
@@ -337,7 +372,7 @@ fn block_batch() {
 
     let block = produce_dummy_block(4, Some(prev_hash), vec![transfer_tx]);
     block_res.push(block.clone());
-    dbio.put_block(&block, [4; 32], 0).unwrap();
+    dbio.put_block(&block, [4; 32], 0, None).unwrap();
 
     let block_hashes_mem: Vec<[u8; 32]> =
         block_res.into_iter().map(|bl| bl.header.hash.0).collect();
@@ -396,7 +431,7 @@ fn account_map() {
 
     let block = produce_dummy_block(1, None, vec![transfer_tx1, transfer_tx2]);
 
-    dbio.put_block(&block, [1; 32], 0).unwrap();
+    dbio.put_block(&block, [1; 32], 0, None).unwrap();
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
     let last_block = dbio.get_block(last_id).unwrap().unwrap();
@@ -411,7 +446,7 @@ fn account_map() {
 
     let block = produce_dummy_block(2, Some(prev_hash), vec![transfer_tx1, transfer_tx2]);
 
-    dbio.put_block(&block, [2; 32], 0).unwrap();
+    dbio.put_block(&block, [2; 32], 0, None).unwrap();
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
     let last_block = dbio.get_block(last_id).unwrap().unwrap();
@@ -426,7 +461,7 @@ fn account_map() {
 
     let block = produce_dummy_block(3, Some(prev_hash), vec![transfer_tx1, transfer_tx2]);
 
-    dbio.put_block(&block, [3; 32], 0).unwrap();
+    dbio.put_block(&block, [3; 32], 0, None).unwrap();
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
     let last_block = dbio.get_block(last_id).unwrap().unwrap();
@@ -438,7 +473,7 @@ fn account_map() {
 
     let block = produce_dummy_block(4, Some(prev_hash), vec![transfer_tx]);
 
-    dbio.put_block(&block, [4; 32], 0).unwrap();
+    dbio.put_block(&block, [4; 32], 0, None).unwrap();
 
     let acc1_tx = dbio.get_acc_transactions(*acc1().value(), 0, 7).unwrap();
     let acc1_tx_hashes: Vec<[u8; 32]> = acc1_tx.into_iter().map(|tx| tx.hash().0).collect();
@@ -451,3 +486,14 @@ fn account_map() {
 
     assert_eq!(acc1_tx_limited_hashes.as_slice(), &tx_hash_res[1..5]);
 }
+
+#[test]
+fn reopen_preserves_breakpoint_meta() {
+    let temp_dir = tempdir().unwrap();
+    {
+        let dbio = RocksDBIO::open_or_create(temp_dir.path(), &initial_state()).unwrap();
+        dbio.put_meta_last_breakpoint_id(5).unwrap();
+    } // drop releases the RocksDB lock
+    let dbio = RocksDBIO::open_or_create(temp_dir.path(), &initial_state()).unwrap();
+    assert_eq!(dbio.get_meta_last_breakpoint_id().unwrap(), Some(5));
+}

lez/storage/src/indexer/write_atomic.rs

@@ -2,13 +2,13 @@ use std::collections::HashMap;
 
 use rocksdb::WriteBatch;
 
-use super::{BREAKPOINT_INTERVAL, Block, DbError, DbResult, RocksDBIO};
+use super::{BREAKPOINT_INTERVAL, Block, DbError, DbResult, RocksDBIO, V03State};
 use crate::{
     DBIO as _,
     cells::shared_cells::{FirstBlockCell, FirstBlockSetCell, LastBlockCell},
     indexer::indexer_cells::{
-        AccNumTxCell, BlockHashToBlockIdMapCell, LastBreakpointIdCell, LastObservedL1LibHeaderCell,
-        TipSlotCell, TxHashToBlockIdMapCell,
+        AccNumTxCell, BlockHashToBlockIdMapCell, BreakpointCellRef, LastBreakpointIdCell,
+        LastObservedL1LibHeaderCell, TipSlotCell, TxHashToBlockIdMapCell,
     },
 };
 
@@ -151,8 +151,15 @@ impl RocksDBIO {
         self.put_batch(&FirstBlockSetCell(true), (), write_batch)
     }
 
-    /// Put a block atomically (via [`WriteBatch`]) along with its L1 header and `Slot`.
-    pub fn put_block(&self, block: &Block, l1_lib_header: [u8; 32], l1_slot: u64) -> DbResult<()> {
+    /// Put a block atomically (via [`WriteBatch`]) along with its L1 header, `Slot`,
+    /// and — for interval-boundary blocks — its post-state snapshot.
+    pub fn put_block(
+        &self,
+        block: &Block,
+        l1_lib_header: [u8; 32],
+        l1_slot: u64,
+        breakpoint: Option<&V03State>,
+    ) -> DbResult<()> {
         let cf_block = self.block_column();
         let last_curr_block = self.get_meta_last_block_id_in_db()?.unwrap_or(0);
         let mut write_batch = WriteBatch::default();
@@ -216,18 +223,29 @@ impl RocksDBIO {
             self.put_account_transactions_dependant(acc_id, &tx_hashes, &mut write_batch)?;
         }
 
+        if let Some(state) = breakpoint {
+            if !block
+                .header
+                .block_id
+                .is_multiple_of(BREAKPOINT_INTERVAL.into())
+            {
+                return Err(DbError::db_interaction_error(
+                    "Breakpoint snapshot must accompany an interval-boundary block".to_owned(),
+                ));
+            }
+            let br_id = block
+                .header
+                .block_id
+                .checked_div(BREAKPOINT_INTERVAL.into())
+                .expect("Breakpoint interval is not zero");
+            self.put_batch(&BreakpointCellRef(state), br_id, &mut write_batch)?;
+            self.put_meta_last_breakpoint_id_batch(br_id, &mut write_batch)?;
+        }
+
         self.db.write(write_batch).map_err(|rerr| {
             DbError::rocksdb_cast_message(rerr, Some("Failed to write batch".to_owned()))
         })?;
 
-        if block
-            .header
-            .block_id
-            .is_multiple_of(BREAKPOINT_INTERVAL.into())
-        {
-            self.put_next_breakpoint()?;
-        }
-
         Ok(())
     }
 }

lez/storage/src/indexer/write_non_atomic.rs

@@ -1,4 +1,4 @@
-use super::{BREAKPOINT_INTERVAL, DbError, DbResult, RocksDBIO, V03State};
+use super::{DbResult, RocksDBIO, V03State};
 use crate::{
     DBIO as _,
     cells::shared_cells::{FirstBlockSetCell, LastBlockCell},
@@ -44,27 +44,4 @@ impl RocksDBIO {
     pub fn put_breakpoint(&self, br_id: u64, breakpoint: &V03State) -> DbResult<()> {
         self.put(&BreakpointCellRef(breakpoint), br_id)
     }
-
-    pub fn put_next_breakpoint(&self) -> DbResult<()> {
-        let last_block = self.get_meta_last_block_id_in_db()?.unwrap_or(0);
-        let next_breakpoint_id = self
-            .get_meta_last_breakpoint_id()?
-            .unwrap_or(0)
-            .checked_add(1)
-            .expect("Breakpoint Id will be lesser than u64::MAX");
-        let block_to_break_id = next_breakpoint_id
-            .checked_mul(u64::from(BREAKPOINT_INTERVAL))
-            .expect("Reached maximum breakpoint id");
-
-        if block_to_break_id <= last_block {
-            let next_breakpoint = self.calculate_state_for_id(block_to_break_id)?;
-
-            self.put_breakpoint(next_breakpoint_id, &next_breakpoint)?;
-            self.put_meta_last_breakpoint_id(next_breakpoint_id)
-        } else {
-            Err(DbError::db_interaction_error(
-                "Breakpoint not yet achieved".to_owned(),
-            ))
-        }
-    }
 }

test_fixtures/Cargo.toml

@@ -31,4 +31,5 @@ serde_json.workspace = true
 tempfile.workspace = true
 testcontainers = { version = "0.27.3", features = ["docker-compose"] }
 tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
+tokio-util.workspace = true
 url.workspace = true

test_fixtures/src/setup.rs

@@ -101,10 +101,15 @@ pub async fn setup_indexer(bedrock_addr: SocketAddr) -> Result<(IndexerHandle, T
     let indexer_config =
         config::indexer_config(bedrock_addr).context("Failed to create Indexer config")?;
 
-    indexer_service::run_server(indexer_config, temp_indexer_dir.path(), 0)
-        .await
-        .context("Failed to run Indexer Service")
-        .map(|handle| (handle, temp_indexer_dir))
+    indexer_service::run_server(
+        indexer_config,
+        temp_indexer_dir.path(),
+        0,
+        tokio_util::sync::CancellationToken::new(),
+    )
+    .await
+    .context("Failed to run Indexer Service")
+    .map(|handle| (handle, temp_indexer_dir))
 }
 
 pub async fn setup_sequencer(