diff --git a/lez/indexer/core/src/block_store.rs b/lez/indexer/core/src/block_store.rs index bd47d582..4b26dbe4 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::{BREAKPOINT_INTERVAL, indexer::RocksDBIO}; +use storage::indexer::RocksDBIO; use tokio::sync::RwLock; use crate::{ingest_error::BlockIngestError, stall_reason::StallReason}; @@ -269,13 +269,8 @@ impl IndexerStore { 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(), breakpoint) + .put_block(&stored, [0_u8; 32], l1_slot.into_inner(), &scratch) .context("Failed to persist accepted block")?; // Commit in-memory state (infallible) only after the DB write succeeded. diff --git a/lez/storage/src/indexer/tests.rs b/lez/storage/src/indexer/tests.rs index 840c99fe..d87aaf1c 100644 --- a/lez/storage/src/indexer/tests.rs +++ b/lez/storage/src/indexer/tests.rs @@ -49,10 +49,11 @@ fn initial_state() -> lee::V03State { #[test] fn start_db() { + let initial_state = initial_state(); 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(temdir_path, &initial_state).unwrap(); let last_id = dbio.get_meta_last_block_id_in_db().unwrap(); let first_id = dbio.get_meta_first_block_id_in_db().unwrap(); @@ -79,13 +80,15 @@ fn start_db() { #[test] fn one_block_insertion() { + let initial_state = initial_state(); 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(temdir_path, &initial_state).unwrap(); let genesis_block = genesis_block(); - dbio.put_block(&genesis_block, [0; 32], 0, None).unwrap(); + dbio.put_block(&genesis_block, [0; 32], 0, &initial_state) + .unwrap(); let prev_hash = genesis_block.header.hash; let from = acc1(); @@ -96,7 +99,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, None).unwrap(); + dbio.put_block(&block, [1; 32], 0, &initial_state).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(); @@ -126,55 +129,44 @@ 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 initial_state = initial_state(); let temp_dir = tempdir().unwrap(); - let dbio = RocksDBIO::open_or_create(temp_dir.path(), &initial_state()).unwrap(); + let dbio = RocksDBIO::open_or_create(temp_dir.path(), &initial_state).unwrap(); 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, None) + dbio.put_block(&genesis_block, [0; 32], 1_000, &initial_state) .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, None).unwrap(); + dbio.put_block(&block, [1; 32], 1_005, &initial_state) + .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, None) + dbio.put_block(&genesis_block, [0; 32], 1_010, &initial_state) .unwrap(); assert_eq!(dbio.get_meta_tip_slot_in_db().unwrap(), Some(1_005)); } #[test] fn put_block_stores_breakpoint_in_same_batch() { + let initial_state = initial_state(); let temp_dir = tempdir().unwrap(); - let dbio = RocksDBIO::open_or_create(temp_dir.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(); - // 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. + // Chain blocks 1..=BREAKPOINT_INTERVAL. The snapshot is scheduled internally + // by put_block at the boundary block; every call passes the same recognizable + // marker state (the initial one), proving it's stored verbatim rather than + // recomputed. 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(); @@ -190,14 +182,13 @@ fn put_block_stores_breakpoint_in_same_batch() { ); let block = produce_dummy_block(i.into(), prev_hash, vec![transfer_tx]); - let marker = (i == BREAKPOINT_INTERVAL).then(initial_state); - dbio.put_block(&block, [i; 32], 0, marker.as_ref()).unwrap(); + dbio.put_block(&block, [i; 32], 0, &initial_state).unwrap(); } 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. + // Only the boundary block schedules a write: breakpoint 0 must be the only other one. assert_eq!( dbio.get_breakpoint(0) .unwrap() @@ -209,8 +200,9 @@ fn put_block_stores_breakpoint_in_same_batch() { #[test] fn state_replay_falls_back_over_missing_breakpoints() { + let initial_state = initial_state(); let temp_dir = tempdir().unwrap(); - let dbio = RocksDBIO::open_or_create(temp_dir.path(), &initial_state()).unwrap(); + let dbio = RocksDBIO::open_or_create(temp_dir.path(), &initial_state).unwrap(); let from = acc1(); let to = acc2(); @@ -229,9 +221,11 @@ fn state_replay_falls_back_over_missing_breakpoints() { &sign_key, ); let block = produce_dummy_block(i, prev_hash, vec![transfer_tx]); - dbio.put_block(&block, [0; 32], 0, None).unwrap(); + dbio.put_block(&block, [0; 32], 0, &initial_state).unwrap(); } + // Simulate a store whose boundary snapshot was lost (#605). + dbio.delete_breakpoint(1).unwrap(); assert!(dbio.get_breakpoint_opt(1).unwrap().is_none()); let final_state = dbio.final_state().unwrap(); assert_eq!( @@ -246,10 +240,11 @@ fn state_replay_falls_back_over_missing_breakpoints() { #[test] fn simple_maps() { + let initial_state = initial_state(); 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(temdir_path, &initial_state).unwrap(); let from = acc1(); let to = acc2(); @@ -261,7 +256,7 @@ fn simple_maps() { let control_hash1 = block.header.hash; - dbio.put_block(&block, [1; 32], 0, None).unwrap(); + dbio.put_block(&block, [1; 32], 0, &initial_state).unwrap(); let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap(); let last_block = dbio.get_block(last_id).unwrap().unwrap(); @@ -273,7 +268,7 @@ fn simple_maps() { let control_hash2 = block.header.hash; - dbio.put_block(&block, [2; 32], 0, None).unwrap(); + dbio.put_block(&block, [2; 32], 0, &initial_state).unwrap(); let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap(); let last_block = dbio.get_block(last_id).unwrap().unwrap(); @@ -285,7 +280,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, None).unwrap(); + dbio.put_block(&block, [3; 32], 0, &initial_state).unwrap(); let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap(); let last_block = dbio.get_block(last_id).unwrap().unwrap(); @@ -297,7 +292,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, None).unwrap(); + dbio.put_block(&block, [4; 32], 0, &initial_state).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(); @@ -318,12 +313,13 @@ fn simple_maps() { #[test] fn block_batch() { + let initial_state = initial_state(); let temp_dir = tempdir().unwrap(); let temdir_path = temp_dir.path(); let mut block_res = vec![]; - let dbio = RocksDBIO::open_or_create(temdir_path, &initial_state()).unwrap(); + let dbio = RocksDBIO::open_or_create(temdir_path, &initial_state).unwrap(); let from = acc1(); let to = acc2(); @@ -334,7 +330,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, None).unwrap(); + dbio.put_block(&block, [1; 32], 0, &initial_state).unwrap(); let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap(); let last_block = dbio.get_block(last_id).unwrap().unwrap(); @@ -345,7 +341,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, None).unwrap(); + dbio.put_block(&block, [2; 32], 0, &initial_state).unwrap(); let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap(); let last_block = dbio.get_block(last_id).unwrap().unwrap(); @@ -356,7 +352,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, None).unwrap(); + dbio.put_block(&block, [3; 32], 0, &initial_state).unwrap(); let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap(); let last_block = dbio.get_block(last_id).unwrap().unwrap(); @@ -367,7 +363,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, None).unwrap(); + dbio.put_block(&block, [4; 32], 0, &initial_state).unwrap(); let block_hashes_mem: Vec<[u8; 32]> = block_res.into_iter().map(|bl| bl.header.hash.0).collect(); @@ -406,10 +402,11 @@ fn block_batch() { #[test] fn account_map() { + let initial_state = initial_state(); 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(temdir_path, &initial_state).unwrap(); let from = acc1(); let to = acc2(); @@ -426,7 +423,7 @@ fn account_map() { let block = produce_dummy_block(1, None, vec![transfer_tx1, transfer_tx2]); - dbio.put_block(&block, [1; 32], 0, None).unwrap(); + dbio.put_block(&block, [1; 32], 0, &initial_state).unwrap(); let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap(); let last_block = dbio.get_block(last_id).unwrap().unwrap(); @@ -441,7 +438,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, None).unwrap(); + dbio.put_block(&block, [2; 32], 0, &initial_state).unwrap(); let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap(); let last_block = dbio.get_block(last_id).unwrap().unwrap(); @@ -456,7 +453,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, None).unwrap(); + dbio.put_block(&block, [3; 32], 0, &initial_state).unwrap(); let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap(); let last_block = dbio.get_block(last_id).unwrap().unwrap(); @@ -468,7 +465,7 @@ fn account_map() { let block = produce_dummy_block(4, Some(prev_hash), vec![transfer_tx]); - dbio.put_block(&block, [4; 32], 0, None).unwrap(); + dbio.put_block(&block, [4; 32], 0, &initial_state).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(); @@ -484,11 +481,12 @@ fn account_map() { #[test] fn reopen_preserves_seeded_breakpoint() { + let initial_state = initial_state(); let temp_dir = tempdir().unwrap(); { - let dbio = RocksDBIO::open_or_create(temp_dir.path(), &initial_state()).unwrap(); + let dbio = RocksDBIO::open_or_create(temp_dir.path(), &initial_state).unwrap(); assert!(dbio.get_breakpoint_opt(0).unwrap().is_some()); } // drop releases the RocksDB lock - let dbio = RocksDBIO::open_or_create(temp_dir.path(), &initial_state()).unwrap(); + let dbio = RocksDBIO::open_or_create(temp_dir.path(), &initial_state).unwrap(); assert!(dbio.get_breakpoint_opt(0).unwrap().is_some()); } diff --git a/lez/storage/src/indexer/write_atomic.rs b/lez/storage/src/indexer/write_atomic.rs index 7cd16e10..729bcc02 100644 --- a/lez/storage/src/indexer/write_atomic.rs +++ b/lez/storage/src/indexer/write_atomic.rs @@ -144,13 +144,14 @@ impl RocksDBIO { } /// Put a block atomically (via [`WriteBatch`]) along with its L1 header, `Slot`, - /// and — for interval-boundary blocks — its post-state snapshot. + /// and (at interval-boundary blocks) a snapshot of `post_state`, the block's + /// post-application state. pub fn put_block( &self, block: &Block, l1_lib_header: [u8; 32], l1_slot: u64, - breakpoint: Option<&V03State>, + post_state: &V03State, ) -> DbResult<()> { let cf_block = self.block_column(); let last_curr_block = self.get_meta_last_block_id_in_db()?.unwrap_or(0); @@ -215,22 +216,17 @@ 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(), - )); - } + if block + .header + .block_id + .is_multiple_of(BREAKPOINT_INTERVAL.into()) + { 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_batch(&BreakpointCellRef(post_state), br_id, &mut write_batch)?; } self.db.write(write_batch).map_err(|rerr| { diff --git a/lez/storage/src/indexer/write_non_atomic.rs b/lez/storage/src/indexer/write_non_atomic.rs index 214531bc..a2c178ec 100644 --- a/lez/storage/src/indexer/write_non_atomic.rs +++ b/lez/storage/src/indexer/write_non_atomic.rs @@ -1,4 +1,6 @@ use super::{DbResult, RocksDBIO, V03State}; +#[cfg(test)] +use crate::error::DbError; use crate::{ DBIO as _, cells::shared_cells::{FirstBlockSetCell, LastBlockCell}, @@ -40,4 +42,16 @@ impl RocksDBIO { pub fn put_breakpoint(&self, br_id: u64, breakpoint: &V03State) -> DbResult<()> { self.put(&BreakpointCellRef(breakpoint), br_id) } + + /// Deletes a breakpoint snapshot. Test-only fault injection for simulating + /// stores whose boundary snapshot was lost. + #[cfg(test)] + pub(crate) fn delete_breakpoint(&self, br_id: u64) -> DbResult<()> { + let key = borsh::to_vec(&br_id).map_err(|err| { + DbError::borsh_cast_message(err, Some("Failed to serialize breakpoint id".to_owned())) + })?; + self.db + .delete_cf(&self.breakpoint_column(), key) + .map_err(|rerr| DbError::rocksdb_cast_message(rerr, None)) + } }