From 4f3ea948a73927d3b8055ec747173f97968ab1a0 Mon Sep 17 00:00:00 2001 From: vladimir-ea Date: Tue, 25 Aug 2026 16:16:17 +0100 Subject: [PATCH 1/3] a write io error blocks all writes --- crates/storage/src/store/io.rs | 56 +++++++++++++--------------------- 1 file changed, 22 insertions(+), 34 deletions(-) diff --git a/crates/storage/src/store/io.rs b/crates/storage/src/store/io.rs index eef89e33..7b3c0c80 100644 --- a/crates/storage/src/store/io.rs +++ b/crates/storage/src/store/io.rs @@ -59,13 +59,13 @@ impl Store { { let mut writes = 0; while writes < MAX_WRITES_PER_LOOP && - let Some(pending) = self.write_queue.front() + let Some(pending) = self.write_queue.pop_front() { writes += 1; tracing::debug!(?pending, "process pending write"); match pending { PendingWrite::Index { block_root, slot } => { - let dir = self.finalized_slot_dir(Payload::Block, *slot); + let dir = self.finalized_slot_dir(Payload::Block, slot); std::fs::create_dir_all(&dir)?; let path = dir.join("block_index.bin"); let mut file = open_file_write(path, true)?; @@ -73,51 +73,46 @@ impl Store { // partial append would misalign every subsequent // fixed-width record on retry. let mut record = [0u8; 40]; - record[..32].copy_from_slice(block_root); + record[..32].copy_from_slice(&block_root); record[32..].copy_from_slice(&slot.to_le_bytes()); file.write_all(&record)?; - self.write_queue.pop_front(); } PendingWrite::Column { slot, column, ssz } => { - let dir = self.finalized_slot_dir(Payload::Column, *slot); + let dir = self.finalized_slot_dir(Payload::Column, slot); std::fs::create_dir_all(&dir)?; let path = dir.join(format!("{slot}_{column}.ssz")); let (buffer, _) = ssz.buffer().map_err(Error::other)?; open_file_write(path, false)?.write_all(buffer)?; StorageCounters::BackfillColumnsWritten.inc(); - self.write_queue.pop_front(); } PendingWrite::WriteUnfinalized { slot, key, ssz } => { let path = - self.unfinalized_dir(key.payload()).join(key.unfinalized_name(*slot)); + self.unfinalized_dir(key.payload()).join(key.unfinalized_name(slot)); let (buffer, _) = ssz.buffer().map_err(Error::other)?; open_file_write(&path, false)?.write_all(buffer)?; key.payload().record_written(); - self.write_queue.pop_front(); } PendingWrite::Promote { slot, key } => { let payload = key.payload(); - let dir = self.finalized_slot_dir(payload, *slot); + let dir = self.finalized_slot_dir(payload, slot); std::fs::create_dir_all(&dir)?; - let from = self.unfinalized_dir(payload).join(key.unfinalized_name(*slot)); - rename_tolerant(&from, &dir.join(key.finalized_name(*slot)))?; + let from = self.unfinalized_dir(payload).join(key.unfinalized_name(slot)); + rename_tolerant(&from, &dir.join(key.finalized_name(slot)))?; // Data before index: a block's index record is appended only // after the rename, so a crash never indexes an unmoved block. if let PayloadKey::Block { block_root, .. } = key { let mut record = [0u8; 40]; - record[..32].copy_from_slice(block_root); + record[..32].copy_from_slice(&block_root); record[32..].copy_from_slice(&slot.to_le_bytes()); open_file_write(dir.join("block_index.bin"), true)?.write_all(&record)?; } payload.record_promoted(); - self.write_queue.pop_front(); } PendingWrite::Prune { slot, key } => { let path = - self.unfinalized_dir(key.payload()).join(key.unfinalized_name(*slot)); + self.unfinalized_dir(key.payload()).join(key.unfinalized_name(slot)); remove_tolerant(&path)?; key.payload().record_pruned(); - self.write_queue.pop_front(); } PendingWrite::TruncateHistory { payload, finalized_slot } => { let epoch = finalized_slot / SLOTS_PER_EPOCH; @@ -127,14 +122,13 @@ impl Store { PathBuf::new().join(&self.store_dir).join(payload.finalized_dir_name()); remove_subdirs(dir, earliest_slot)?; self.history.note_truncation(earliest_slot); - self.write_queue.pop_front(); } PendingWrite::StartBlockBackfill { finalized_slot, finalized_root } => { let epoch = finalized_slot / SLOTS_PER_EPOCH; let to_retain = Payload::Block.slots_retained(&self.spec, epoch); let start_slot = finalized_slot.saturating_sub(to_retain).max(1); tracing::info!( - finalized_slot = *finalized_slot, + finalized_slot, to_retain, start_slot, "block backfill armed" @@ -144,15 +138,15 @@ impl Store { .join(Payload::Block.finalized_dir_name()); let range = match earliest_block(dir)? { Some((slot, parent_root)) if slot > start_slot => { - Some((start_slot..slot.min(*finalized_slot), parent_root)) + Some((start_slot..slot.min(finalized_slot), parent_root)) } // No blocks on disk: backfill `[start_slot, finalized_slot]` // anchored at the finalized block. Skip when nothing is // finalized yet — the zero `finalized_root` is not a real // block root, so the chain can never link and backfill // would respin on slot 0 (genesis is already the anchor). - None if *finalized_root != [0u8; 32] => { - Some((start_slot..*finalized_slot + 1, *finalized_root)) + None if finalized_root != [0u8; 32] => { + Some((start_slot..finalized_slot + 1, finalized_root)) } _ => None, }; @@ -165,57 +159,51 @@ impl Store { None => self.history.no_block_gap(), } - self.write_queue.pop_front(); } PendingWrite::StartBackfill { finalized_slot, finalized_root } => { - self.history.start(*finalized_slot, *finalized_root, &self.spec); - self.write_queue.pop_front(); + self.history.start(finalized_slot, finalized_root, &self.spec); } PendingWrite::BackfillBlock { block_root, slot, ssz } => { - let dir = self.finalized_slot_dir(Payload::Block, *slot); + let dir = self.finalized_slot_dir(Payload::Block, slot); std::fs::create_dir_all(&dir)?; let path = dir.join(format!("{slot}_block.ssz")); let (buffer, _) = ssz.buffer().map_err(Error::other)?; open_file_write(path, false)?.write_all(buffer)?; - if !self.root_index.contains_key(block_root) { + if !self.root_index.contains_key(&block_root) { let mut record = [0u8; 40]; - record[..32].copy_from_slice(block_root); + record[..32].copy_from_slice(&block_root); record[32..].copy_from_slice(&slot.to_le_bytes()); open_file_write(dir.join("block_index.bin"), true)?.write_all(&record)?; - self.root_index.insert(*block_root, *slot); + self.root_index.insert(block_root, slot); } // Set 2: a block fetched by block backfill that falls in the // column window needs its columns too (the pre-block disk // scan couldn't see it — it wasn't on disk yet). Feed the // still-live column backfill. Just-written ⇒ no columns on // disk yet, so the full custody set is missing. - let is_gloas = self.spec.is_gloas_at_slot(*slot); + let is_gloas = self.spec.is_gloas_at_slot(slot); // Just written ⇒ no columns on disk yet, so the whole // custody set is missing. let missing = match SignedBeaconBlockView::has_data_columns(buffer, is_gloas) { true => custody_group_columns, false => 0, }; - self.history.seed(*block_root, *slot, buffer, missing, is_gloas, &self.spec); + self.history.seed(block_root, slot, buffer, missing, is_gloas, &self.spec); StorageCounters::BackfillBlocksWritten.inc(); - self.write_queue.pop_front(); } PendingWrite::BackfillEnvelope { slot, ssz } => { - let dir = self.finalized_slot_dir(Payload::Envelope, *slot); + let dir = self.finalized_slot_dir(Payload::Envelope, slot); std::fs::create_dir_all(&dir)?; let (buffer, _) = ssz.buffer().map_err(Error::other)?; open_file_write(dir.join(format!("{slot}_envelope.ssz")), false)? .write_all(buffer)?; - self.write_queue.pop_front(); } PendingWrite::PersistPeer { enr } => { let peer_file = self.peers_dir().join(format!("{}.enr", enr.public_key())); open_file_write(peer_file, false)?.write_all(enr.to_string().as_bytes())?; - self.write_queue.pop_front(); } PendingWrite::LoadPeers => { - self.write_queue.pop_front(); let peer_files = std::fs::read_dir(self.peers_dir())?; for entry in peer_files { if let Ok(entry) = entry && From aa24ab44939d51bf3fe5158468a7340f4a97daf1 Mon Sep 17 00:00:00 2001 From: vladimir-ea Date: Tue, 25 Aug 2026 16:21:31 +0100 Subject: [PATCH 2/3] fmt --- crates/storage/src/store/io.rs | 22 ++++++---------------- 1 file changed, 6 insertions(+), 16 deletions(-) diff --git a/crates/storage/src/store/io.rs b/crates/storage/src/store/io.rs index 7b3c0c80..b864a5fe 100644 --- a/crates/storage/src/store/io.rs +++ b/crates/storage/src/store/io.rs @@ -5,10 +5,7 @@ //! parent module, since `load` needs the layout constants too. use std::{ - fs::File, - io::{Error, ErrorKind, Read, Write}, - path::{Path, PathBuf}, - time::Instant, + collections::hash_map::Entry, fs::File, io::{Error, ErrorKind, Read, Write}, path::{Path, PathBuf}, time::Instant }; use flux_profiler::timed; @@ -86,8 +83,7 @@ impl Store { StorageCounters::BackfillColumnsWritten.inc(); } PendingWrite::WriteUnfinalized { slot, key, ssz } => { - let path = - self.unfinalized_dir(key.payload()).join(key.unfinalized_name(slot)); + let path = self.unfinalized_dir(key.payload()).join(key.unfinalized_name(slot)); let (buffer, _) = ssz.buffer().map_err(Error::other)?; open_file_write(&path, false)?.write_all(buffer)?; key.payload().record_written(); @@ -109,8 +105,7 @@ impl Store { payload.record_promoted(); } PendingWrite::Prune { slot, key } => { - let path = - self.unfinalized_dir(key.payload()).join(key.unfinalized_name(slot)); + let path = self.unfinalized_dir(key.payload()).join(key.unfinalized_name(slot)); remove_tolerant(&path)?; key.payload().record_pruned(); } @@ -127,12 +122,7 @@ impl Store { let epoch = finalized_slot / SLOTS_PER_EPOCH; let to_retain = Payload::Block.slots_retained(&self.spec, epoch); let start_slot = finalized_slot.saturating_sub(to_retain).max(1); - tracing::info!( - finalized_slot, - to_retain, - start_slot, - "block backfill armed" - ); + tracing::info!(finalized_slot, to_retain, start_slot, "block backfill armed"); let dir = PathBuf::new() .join(&self.store_dir) .join(Payload::Block.finalized_dir_name()); @@ -170,12 +160,12 @@ impl Store { let (buffer, _) = ssz.buffer().map_err(Error::other)?; open_file_write(path, false)?.write_all(buffer)?; - if !self.root_index.contains_key(&block_root) { + if let Entry::Vacant(e) = self.root_index.entry(block_root) { let mut record = [0u8; 40]; record[..32].copy_from_slice(&block_root); record[32..].copy_from_slice(&slot.to_le_bytes()); open_file_write(dir.join("block_index.bin"), true)?.write_all(&record)?; - self.root_index.insert(block_root, slot); + e.insert(slot); } // Set 2: a block fetched by block backfill that falls in the // column window needs its columns too (the pre-block disk From c3962c71e3d4c74c908d001108507a7d448fc5f1 Mon Sep 17 00:00:00 2001 From: vladimir-ea Date: Tue, 25 Aug 2026 16:33:50 +0100 Subject: [PATCH 3/3] fmt --- crates/bin/src/main.rs | 2 +- crates/network/src/p2p/quic/peer.rs | 2 +- crates/storage/src/store/io.rs | 6 +++++- 3 files changed, 7 insertions(+), 3 deletions(-) diff --git a/crates/bin/src/main.rs b/crates/bin/src/main.rs index 6e28d3e1..1ecbc7c8 100644 --- a/crates/bin/src/main.rs +++ b/crates/bin/src/main.rs @@ -140,7 +140,7 @@ fn main() -> Result<(), Box> { gossip_producer: incoming_gossip_producer, gossip_consumer: outgoing_gossip_producer .cache_ref() - .random_access("p2p_outgoing_gossip", true)?, + .strict_random_access("p2p_outgoing_gossip", true)?, rpc_producer: incoming_rpc_producer, rpc_consumer: outgoing_rpc_producer.cache_ref().random_access("p2p_outgoing_rpc", true)?, identify: Some(ProtoIdentify::from((&config.identify()?, &keypair))), diff --git a/crates/network/src/p2p/quic/peer.rs b/crates/network/src/p2p/quic/peer.rs index bbf6c9b4..3377dd12 100644 --- a/crates/network/src/p2p/quic/peer.rs +++ b/crates/network/src/p2p/quic/peer.rs @@ -684,7 +684,7 @@ fn id_from_connection(conn: &Connection) -> Option { fn out_buffer(id: &P2pStreamId, incoming: bool) -> OutboundBuffer { match id.protocol() { - StreamProtocol::GossipSub => OutboundBuffer::Gossip(OutBuffer::new(32 * 1024)), + StreamProtocol::GossipSub => OutboundBuffer::Gossip(OutBuffer::new(8 * 1024)), StreamProtocol::BeaconBlocksByRange | StreamProtocol::BeaconBlocksByRoot | StreamProtocol::DataColumnSidecarsByRange | diff --git a/crates/storage/src/store/io.rs b/crates/storage/src/store/io.rs index b864a5fe..17470c85 100644 --- a/crates/storage/src/store/io.rs +++ b/crates/storage/src/store/io.rs @@ -5,7 +5,11 @@ //! parent module, since `load` needs the layout constants too. use std::{ - collections::hash_map::Entry, fs::File, io::{Error, ErrorKind, Read, Write}, path::{Path, PathBuf}, time::Instant + collections::hash_map::Entry, + fs::File, + io::{Error, ErrorKind, Read, Write}, + path::{Path, PathBuf}, + time::Instant, }; use flux_profiler::timed;