diff --git a/Cargo.lock b/Cargo.lock index 0bd194bb..47ffe7a3 100644 --- a/Cargo.lock +++ b/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", ] diff --git a/lez/indexer/core/src/block_store.rs b/lez/indexer/core/src/block_store.rs index 2a2bfbce..5298b96d 100644 --- a/lez/indexer/core/src/block_store.rs +++ b/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 { 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"); + } } diff --git a/lez/indexer/core/src/ingest_error.rs b/lez/indexer/core/src/ingest_error.rs index bda53497..299013d7 100644 --- a/lez/indexer/core/src/ingest_error.rs +++ b/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::*; diff --git a/lez/indexer/core/src/lib.rs b/lez/indexer/core/src/lib.rs index 77839296..fb9f7c31 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); @@ -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. 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); + } +} diff --git a/lez/indexer/service/src/lib.rs b/lez/indexer/service/src/lib.rs index 355d7801..945c75be 100644 --- a/lez/indexer/service/src/lib.rs +++ b/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 { #[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()) diff --git a/lez/indexer/service/src/main.rs b/lez/indexer/service/src/main.rs index 3e36967d..52f195e9 100644 --- a/lez/indexer/service/src/main.rs +++ b/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() => { diff --git a/lez/indexer/service/src/service.rs b/lez/indexer/service/src/service.rs index 767560b1..3ef6a598 100644 --- a/lez/indexer/service/src/service.rs +++ b/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 { + pub async fn new( + config: IndexerConfig, + storage_dir: &Path, + shutdown: CancellationToken, + ) -> Result { 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, 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::>(); @@ -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 = 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>, + handle: tokio::task::JoinHandle>, new_subscription_sender: UnboundedSender>, } diff --git a/lez/storage/Cargo.toml b/lez/storage/Cargo.toml index 8767b525..50eaed45 100644 --- a/lez/storage/Cargo.toml +++ b/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 diff --git a/lez/storage/src/indexer/mod.rs b/lez/storage/src/indexer/mod.rs index 0226bd48..5a265d47 100644 --- a/lez/storage/src/indexer/mod.rs +++ b/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) diff --git a/lez/storage/src/indexer/read_once.rs b/lez/storage/src/indexer/read_once.rs index 83f58d58..043c0311 100644 --- a/lez/storage/src/indexer/read_once.rs +++ b/lez/storage/src/indexer/read_once.rs @@ -55,6 +55,11 @@ impl RocksDBIO { self.get::(br_id).map(|cell| cell.0) } + pub fn get_breakpoint_opt(&self, br_id: u64) -> DbResult> { + self.get_opt::(br_id) + .map(|opt| opt.map(|cell| cell.0)) + } + // Mappings pub fn get_block_id_by_hash(&self, hash: [u8; 32]) -> DbResult> { diff --git a/lez/storage/src/indexer/tests.rs b/lez/storage/src/indexer/tests.rs index 5cd0607b..7c8f055f 100644 --- a/lez/storage/src/indexer/tests.rs +++ b/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)); +} diff --git a/lez/storage/src/indexer/write_atomic.rs b/lez/storage/src/indexer/write_atomic.rs index bed1c53b..9cab8755 100644 --- a/lez/storage/src/indexer/write_atomic.rs +++ b/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(()) } } diff --git a/lez/storage/src/indexer/write_non_atomic.rs b/lez/storage/src/indexer/write_non_atomic.rs index 215250b7..382e9d83 100644 --- a/lez/storage/src/indexer/write_non_atomic.rs +++ b/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(), - )) - } - } } diff --git a/test_fixtures/Cargo.toml b/test_fixtures/Cargo.toml index 71f887b6..c83bdff9 100644 --- a/test_fixtures/Cargo.toml +++ b/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 diff --git a/test_fixtures/src/setup.rs b/test_fixtures/src/setup.rs index 6559192f..af12a381 100644 --- a/test_fixtures/src/setup.rs +++ b/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(