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

fix(indexer): initial prep for breakpoint fix

Browse Source
This commit is contained in:
erhant 2026-07-09 11:04:21 +03:00
parent 9b2b4a3d8c
commit 55a7ab8587
4 changed files with 69 additions and 93 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -261,7 +261,7 @@ impl IndexerStore {
let mut stored = block.clone();
stored.bedrock_status = BedrockStatus::Finalized;
self.dbio
.put_block(&stored, [0_u8; 32], l1_slot.into_inner())
.put_block(&stored, [0_u8; 32], l1_slot.into_inner(), None)
.context("Failed to persist accepted block")?;
// Commit in-memory state (infallible) only after the DB write succeeded.

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

@ -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();
@ -138,30 +138,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 +178,22 @@ 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!(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);
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!(
prev_breakpoint.get_account_by_id(acc1()).balance
- final_state.get_account_by_id(acc1()).balance,
101
);
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
dbio.get_breakpoint(0)
.unwrap()
.get_account_by_id(acc1())
.balance,
10000
);
}
@ -231,7 +214,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 +226,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 +238,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 +250,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 +287,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 +298,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 +309,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 +320,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 +379,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 +394,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 +409,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 +421,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();

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

@ -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,27 @@ impl RocksDBIO {
self.put_account_transactions_dependant(acc_id, &tx_hashes, &mut write_batch)?;
}
if let Some(state) = breakpoint {
debug_assert!(
block
.header
.block_id
.is_multiple_of(BREAKPOINT_INTERVAL.into()),
"breakpoint snapshot must accompany an interval-boundary block"
);
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(())
}
}

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

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