fix(exex): skip missing notification files when filling the WAL block cache

crates/exex/exex/src/wal/mod.rs

@@ -23,7 +23,7 @@ use alloy_eips::BlockNumHash;
 use alloy_primitives::B256;
 use parking_lot::{RwLock, RwLockReadGuard};
 use reth_exex_types::ExExNotification;
-use reth_tracing::tracing::{debug, instrument};
+use reth_tracing::tracing::{debug, instrument, warn};
 
 /// WAL is a write-ahead log (WAL) that stores the notifications sent to ExExes.
 ///
@@ -120,7 +120,21 @@ where
         let mut notifications_size = 0;
 
         for entry in self.storage.iter_notifications(files_range) {
-            let (file_id, size, notification) = entry?;
+            // `commit` consumes the next file ID before writing, so a failed write (e.g. a full
+            // disk) leaves a permanent hole in the ID range. Aborting here would make the node
+            // unbootable until the WAL directory is deleted by hand.
+            let (file_id, size, notification) = match entry {
+                Ok(entry) => entry,
+                Err(WalError::FileNotFound(file_id)) => {
+                    warn!(
+                        target: "exex::wal",
+                        ?file_id,
+                        "Notification file is missing from the WAL, skipping"
+                    );
+                    continue
+                }
+                Err(err) => return Err(err),
+            };
 
             notifications_size += size;
 
@@ -238,12 +252,13 @@ mod tests {
     use crate::wal::{cache::CachedBlock, error::WalResult, Wal};
     use alloy_primitives::B256;
     use itertools::Itertools;
+    use reth_ethereum_primitives::EthPrimitives;
     use reth_exex_types::ExExNotification;
     use reth_provider::Chain;
     use reth_testing_utils::generators::{
         self, random_block, random_block_range, BlockParams, BlockRangeParams,
     };
-    use std::{collections::BTreeMap, sync::Arc};
+    use std::{collections::BTreeMap, fs, sync::Arc};
 
     fn read_notifications(wal: &Wal) -> WalResult<Vec<ExExNotification>> {
         wal.inner.storage.files_range()?.map_or(Ok(Vec::new()), |range| {
@@ -522,4 +537,35 @@ mod tests {
 
         Ok(())
     }
+
+    #[test]
+    fn test_fill_block_cache_skips_missing_notification_file() -> eyre::Result<()> {
+        reth_tracing::init_test_tracing();
+
+        let mut rng = generators::rng();
+
+        let temp_dir = tempfile::tempdir()?;
+        let wal = Wal::<EthPrimitives>::new(&temp_dir)?;
+
+        let blocks = random_block_range(&mut rng, 0..=2, BlockRangeParams::default())
+            .into_iter()
+            .map(|block| block.try_recover())
+            .collect::<Result<Vec<_>, _>>()?;
+
+        for block in &blocks {
+            wal.commit(&ExExNotification::ChainCommitted {
+                new: Arc::new(Chain::new(vec![block.clone()], Default::default(), BTreeMap::new())),
+            })?;
+        }
+
+        fs::remove_file(temp_dir.path().join("1.wal"))?;
+
+        let wal = Wal::<EthPrimitives>::new(&temp_dir)?;
+        assert_eq!(
+            wal.inner.block_cache().blocks_sorted(),
+            [(blocks[2].number, 2), (blocks[0].number, 0)]
+        );
+
+        Ok(())
+    }
 }

crates/net/network/src/test_utils/testnet.rs

@@ -386,8 +386,7 @@ impl<C, Pool> TestnetHandle<C, Pool> {
 
         // add all peers to each other
         for (idx, handle) in self.peers.iter().enumerate().take(self.peers.len() - 1) {
-            for idx in (idx + 1)..self.peers.len() {
-                let neighbour = &self.peers[idx];
+            for neighbour in self.peers.iter().skip(idx + 1) {
                 handle.network.add_peer(*neighbour.peer_id(), neighbour.local_addr());
             }
         }

crates/net/network/src/transactions/fetcher.rs

@@ -184,10 +184,7 @@ impl<N: NetworkPrimitives> TransactionFetcher<N> {
     #[inline]
     pub fn is_idle(&self, peer_id: &PeerId) -> bool {
         let Some(inflight_count) = self.active_peers.peek(peer_id) else { return true };
-        if *inflight_count < self.info.max_inflight_requests_per_peer {
-            return true
-        }
-        false
+        *inflight_count < self.info.max_inflight_requests_per_peer
     }
 
     /// Returns any idle peer for the given hash.