fix(indexer): harden stall durability and breakpoint guards w.r.t. Copilot review

This commit is contained in:
erhant 2026-07-09 15:33:29 +03:00
parent dd75aebb9b
commit 4c79ce8d9c
7 changed files with 49 additions and 14 deletions

1
Cargo.lock generated
View File

@ -9463,6 +9463,7 @@ dependencies = [
"borsh",
"common",
"lee",
"log",
"programs",
"rocksdb",
"system_accounts",

View File

@ -250,12 +250,21 @@ impl IndexerCore {
"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(),
) {
warn!("Failed to record stall reason: {err:#}");
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);

View File

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

View File

@ -13,6 +13,7 @@ lee.workspace = true
thiserror.workspace = true
borsh.workspace = true
log.workspace = true
rocksdb.workspace = true
tempfile.workspace = true

View File

@ -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,
};
@ -159,25 +160,31 @@ impl RocksDBIO {
}
// walk down to the nearest snapshot that exists
let mut breakpoint_id = closest_breakpoint_id(block_id);
let target = closest_breakpoint_id(block_id);
let mut br_id = target;
let mut state = loop {
match self.get_breakpoint_opt(breakpoint_id)? {
match self.get_breakpoint_opt(br_id)? {
Some(state) => break state,
None if breakpoint_id == 0 => {
None if br_id == 0 => {
return Err(DbError::db_interaction_error(
"Breakpoint 0 is missing".to_owned(),
));
}
None => {
breakpoint_id = breakpoint_id
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(breakpoint_id)
.checked_mul(br_id)
.expect("Reached maximum breakpoint id");
for block in self.get_block_batch_seq(

View File

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

View File

@ -224,13 +224,15 @@ impl RocksDBIO {
}
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"
);
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