Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions bin/ethlambda/src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -642,14 +642,17 @@ async fn fetch_initial_state(
// Checkpoint sync path

// Prefer resuming from a fresh on-disk state to avoid re-downloading what we already have.
if let Some(store) = Store::from_db_state(backend.clone(), genesis.genesis_time) {
if let Ok(Some(store)) = Store::from_db_state(backend.clone(), genesis.genesis_time) {
let now_ms = SystemTime::UNIX_EPOCH
.elapsed()
.expect("already past the unix epoch")
.as_millis() as u64;
let current_slot =
now_ms.saturating_sub(genesis.genesis_time * 1000) / MILLISECONDS_PER_SLOT;
let finalized_slot = store.latest_finalized().slot;
let finalized_slot = store
.latest_finalized()
.expect("latest finalized checkpoint exists")
.slot;
let gap = current_slot.saturating_sub(finalized_slot);
if gap <= MAX_RESUMABLE_DB_STATE_AGE {
info!(
Expand Down
2 changes: 1 addition & 1 deletion crates/blockchain/src/coverage.rs
Original file line number Diff line number Diff line change
Expand Up @@ -130,7 +130,7 @@ pub(crate) fn emit_post_block_coverage(
// the head is normally the block proposed at `reporting_slot + 1`, which
// carries this round's votes; filter by `data.slot` so we count the same
// cohort even if the head is at a different slot.
if let Some(block) = store.get_block(&store.head()) {
if let Ok(Some(block)) = store.get_block(&store.head().expect("head exists")) {
for att in block.body.attestations.iter() {
if att.data.slot == reporting_slot {
cov_add(&mut block_v, &mut block_s, &att.aggregation_bits);
Expand Down
69 changes: 53 additions & 16 deletions crates/blockchain/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -113,7 +113,10 @@ impl BlockChain {
) -> BlockChain {
metrics::set_is_aggregator(aggregator.is_enabled());
metrics::set_node_sync_status(metrics::SyncStatus::Idle);
let genesis_time = store.config().genesis_time;
let genesis_time = store
.config()
.expect("failed to load config: config missing or database error")
.genesis_time;
let mut key_manager = key_manager::KeyManager::new(validator_keys);

// Catch XMSS keys up to the current slot before the first tick
Expand Down Expand Up @@ -213,7 +216,7 @@ pub struct BlockChainServer {

impl BlockChainServer {
async fn on_tick(&mut self, timestamp_ms: u64, ctx: &Context<Self>) {
let genesis_time_ms = self.store.config().genesis_time * 1000;
let genesis_time_ms = self.store.config().expect("config exists").genesis_time * 1000;

// Calculate current slot and interval from milliseconds
let time_since_genesis_ms = timestamp_ms.saturating_sub(genesis_time_ms);
Expand All @@ -227,7 +230,7 @@ impl BlockChainServer {
// inside VMs, so a tick scheduled for the next interval boundary can fire
// while the wall clock still reads the previous interval.
let tick_interval = time_since_genesis_ms / MILLISECONDS_PER_INTERVAL;
let store_time = self.store.time();
let store_time = self.store.time().expect("store time exists");

if store_time > 0 && tick_interval <= store_time {
debug!(
Expand Down Expand Up @@ -514,7 +517,7 @@ impl BlockChainServer {
async fn propose_block(&mut self, slot: u64, validator_id: u64) {
info!(%slot, %validator_id, "We are the proposer for this slot");

let genesis_time_ms = self.store.config().genesis_time * 1000;
let genesis_time_ms = self.store.config().expect("config exists").genesis_time * 1000;
let slot_start_ms = genesis_time_ms + slot * MILLISECONDS_PER_SLOT;

// Build the block. `produce_block_with_signatures` advances the store to
Expand Down Expand Up @@ -702,8 +705,18 @@ impl BlockChainServer {
fn process_block(&mut self, signed_block: SignedBlock) -> Result<(), StoreError> {
store::on_block(&mut self.store, signed_block)?;
metrics::update_head_slot(self.store.head_slot());
metrics::update_latest_justified_slot(self.store.latest_justified().slot);
metrics::update_latest_finalized_slot(self.store.latest_finalized().slot);
let latest_justified_slot = self
.store
.latest_justified()
.expect("Error: Latest justified checkpoint does not exist")
.slot;
metrics::update_latest_justified_slot(latest_justified_slot);
let latest_finalized_slot = self
.store
.latest_finalized()
.expect("Error: Latest finalized checkpoint does not exist")
.slot;
metrics::update_latest_finalized_slot(latest_finalized_slot);
metrics::update_validators_count(self.key_manager.validator_ids().len() as u64);

for table in ALL_TABLES {
Expand All @@ -727,7 +740,9 @@ impl BlockChainServer {
// Prune old states and blocks AFTER the entire cascade completes.
// Running this mid-cascade would delete states that pending children
// still need, causing re-processing loops when fallback pruning is active.
self.store.prune_old_data();
self.store
.prune_old_data()
.expect("DB pruning should succeed");
}

/// Try to process a single block. If its parent state is missing, store it
Expand All @@ -747,7 +762,12 @@ impl BlockChainServer {
// already part of the canonical chain and cannot affect fork choice.
// Discard any pending children: since we won't process this block,
// children referencing it as parent would remain stuck indefinitely.
if slot <= self.store.latest_finalized().slot {
let latest_finalized_slot = self
.store
.latest_finalized()
.expect("Error: Latest finalized checkpoint does not exist")
.slot;
if slot <= latest_finalized_slot {
self.discard_pending_subtree(block_root);
return;
}
Expand All @@ -759,7 +779,7 @@ impl BlockChainServer {
// Catching this early also avoids persisting bogus future blocks to
// RocksDB and triggering BlocksByRoot fan-out for fabricated parents.
let block_start_interval = slot.saturating_mul(INTERVALS_PER_SLOT);
let store_time = self.store.time();
let store_time = self.store.time().expect("store time exists");
if block_start_interval > store_time + GOSSIP_DISPARITY_INTERVALS {
warn!(
%slot,
Expand All @@ -774,7 +794,11 @@ impl BlockChainServer {
}

// Check if parent state exists before attempting to process
if !self.store.has_state(&parent_root) {
if !self
.store
.has_state(&parent_root)
.expect("DB read should succeed")
{
info!(%slot, %parent_root, %block_root, "Block parent missing, storing as pending");

// Resolve the actual missing ancestor by walking the chain. A stale entry
Expand Down Expand Up @@ -802,14 +826,23 @@ impl BlockChainServer {
// session, the actual missing block is further up the chain.
// Note: this loop always terminates — blocks reference parents by hash,
// so a cycle would require a hash collision.
while let Some(header) = self.store.get_block_header(&missing_root) {
if self.store.has_state(&header.parent_root) {
while let Some(header) = self
.store
.get_block_header(&missing_root)
.expect("DB read should succeed")
{
if self
.store
.has_state(&header.parent_root)
.expect("DB read should succeed")
{
// Parent state available — enqueue for processing, cascade
// handles the rest via the outer loop.
let block = self
.store
.get_signed_block(&missing_root)
.expect("header and parent state exist, so the full signed block must too");
.expect("header and parent state exist, so the full signed block must too")
.unwrap();
queue.push_back(block);
return;
}
Expand Down Expand Up @@ -921,7 +954,7 @@ impl BlockChainServer {
self.pending_block_parents.remove(&block_root);

// Load block data from DB
let Some(child_block) = self.store.get_signed_block(&block_root) else {
let Ok(Some(child_block)) = self.store.get_signed_block(&block_root) else {
warn!(
block_root = %ShortRoot(&block_root.0),
"Pending block missing from DB, skipping"
Expand Down Expand Up @@ -966,7 +999,11 @@ impl BlockChainServer {

fn update_sync_status(&mut self, current_slot: u64) {
let head_slot = self.store.head_slot();
let max_seen_slot = self.store.max_live_chain_slot().unwrap_or(head_slot);
let max_seen_slot = self
.store
.max_live_chain_slot()
.expect("max live chain slot exists")
.unwrap_or(head_slot);
let status = self
.sync_status
.update(current_slot, head_slot, max_seen_slot);
Expand All @@ -990,7 +1027,7 @@ impl BlockChainServer {
let now_ms = unix_now_ms();
self.on_tick(now_ms, ctx).await;

let genesis_time_ms = self.store.config().genesis_time * 1000;
let genesis_time_ms = self.store.config().expect("Config exists").genesis_time * 1000;
let remaining_at_entry = ms_until_next_interval(now_ms, genesis_time_ms);
let now_after_tick = unix_now_ms();
let elapsed = now_after_tick.saturating_sub(now_ms);
Expand Down
9 changes: 6 additions & 3 deletions crates/blockchain/src/reaggregate.rs
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,7 @@ pub fn reaggregate_from_block(
// The multi-message aggregate proof was built against the parent state's
// validator set. Without it we cannot resolve the pubkey layout the SNARK
// was bound to.
let Some(parent_state) = store.get_state(&block.parent_root) else {
let Ok(Some(parent_state)) = store.get_state(&block.parent_root) else {
debug!(
block_root = %ethlambda_types::ShortRoot(&block.hash_tree_root().0),
"Skipping reaggregation: parent state missing"
Expand Down Expand Up @@ -239,7 +239,10 @@ struct Candidate {
/// capped at [`MAX_REAGGREGATIONS_PER_BLOCK`] so an attacker-shaped block
/// cannot blow past the slot budget.
fn select_candidates(store: &Store, attestations: &[AggregatedAttestation]) -> Vec<Candidate> {
let justified_slot = store.latest_justified().slot;
let justified_slot = store
.latest_justified()
.expect("latest justified checkpoint exists")
.slot;
let mut candidates: Vec<Candidate> = Vec::new();
for (idx, att) in attestations.iter().enumerate() {
if att.data.target.slot <= justified_slot {
Expand Down Expand Up @@ -315,7 +318,7 @@ mod tests {
// Justified at slot 5; an attestation with target.slot = 5 must be skipped.
store
.update_checkpoints(ethlambda_storage::ForkCheckpoints::new(
store.head(),
store.head().expect("head exists"),
Some(Checkpoint {
root: H256::ZERO,
slot: 5,
Expand Down
Loading