diff --git a/Cargo.lock b/Cargo.lock index 1c21dc68..47ffe7a3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9463,6 +9463,7 @@ dependencies = [ "borsh", "common", "lee", + "log", "programs", "rocksdb", "system_accounts", diff --git a/lez/indexer/core/src/lib.rs b/lez/indexer/core/src/lib.rs index 6e7ef6f8..fb9f7c31 100644 --- a/lez/indexer/core/src/lib.rs +++ b/lez/indexer/core/src/lib.rs @@ -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); diff --git a/lez/indexer/service/src/service.rs b/lez/indexer/service/src/service.rs index fedef895..3ef6a598 100644 --- a/lez/indexer/service/src/service.rs +++ b/lez/indexer/service/src/service.rs @@ -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(), 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 eae1e511..28aec986 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, }; @@ -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( diff --git a/lez/storage/src/indexer/tests.rs b/lez/storage/src/indexer/tests.rs index 45361d4d..7c8f055f 100644 --- a/lez/storage/src/indexer/tests.rs +++ b/lez/storage/src/indexer/tests.rs @@ -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(); diff --git a/lez/storage/src/indexer/write_atomic.rs b/lez/storage/src/indexer/write_atomic.rs index 98abaec5..9cab8755 100644 --- a/lez/storage/src/indexer/write_atomic.rs +++ b/lez/storage/src/indexer/write_atomic.rs @@ -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