logos-blockchain/logos-execution-zone
1
0
Fork 0
mirror of https://github.com/logos-blockchain/logos-execution-zone.git synced 2026-07-09 23:39:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(indexer): make put_block own the breakpoint schedule

Browse Source 
`RISC0_DEV_MODE=1 RISC0_SKIP_BUILD=1 cargo test -p storage -p indexer_core`
This commit is contained in:
erhant 2026-07-09 22:48:03 +03:00
parent 313ca7959e
commit 15512a4c97
4 changed files with 72 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
lez/indexer/core/src/block_store.rs
View File

@ -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.

96
lez/storage/src/indexer/tests.rs
View File

@ -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());
}

22
lez/storage/src/indexer/write_atomic.rs
View File

@ -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| {

14
lez/storage/src/indexer/write_non_atomic.rs
View File

@ -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))
}
}