perf(l1): replace synchronous disk I/O with async operations in snap sync (#6113)

## Motivation

During snap sync, multiple file system operations block the tokio
runtime:
- Directory creation (`std::fs::create_dir_all`)
- File reading (`std::fs::read`)
- Directory listing (`std::fs::read_dir`)
- Directory removal (`std::fs::remove_dir_all`)

These synchronous operations cause the async runtime to stall,
preventing network operations from proceeding while disk I/O completes.
This is particularly problematic during:
1. Snapshot file writing (account states, storage states)
2. Snapshot file reading during trie insertion
3. Cleanup after insertion completes

This PR addresses **roadmap item 1.6 (Async Disk I/O)** from the snap
sync improvement plan.

## Description

### Approach: Hybrid Async Strategy

We use a hybrid approach for async file operations:

| Operation | Method | Rationale |
|-----------|--------|-----------|
| `create_dir_all` | `tokio::fs` | Simple, no iterator complexity |
| `read` | `tokio::fs` | Simple, direct async support |
| `remove_dir_all` | `tokio::fs` | Simple, direct async support |
| `read_dir` | `spawn_blocking` | Returns iterator with complex
lifetimes; wrapping in spawn_blocking is cleaner than manual async
iteration |

### New Module: `snap/async_fs.rs`

Created a new module with async wrappers that provide consistent error
handling:

```rust
/// Ensures a directory exists, creating it if necessary.
pub async fn ensure_dir_exists(path: &Path) -> Result<(), SnapError>

/// Reads all file paths from a directory (sorted for deterministic ordering).
pub async fn read_dir_paths(dir: &Path) -> Result<Vec<PathBuf>, SnapError>

/// Reads the contents of a file asynchronously.
pub async fn read_file(path: &Path) -> Result<Vec<u8>, SnapError>

/// Removes a directory and all its contents asynchronously.
pub async fn remove_dir_all(path: &Path) -> Result<(), SnapError>

/// Writes data to a file asynchronously.
pub async fn write_file(path: &Path, contents: &[u8]) -> Result<(), SnapError>

/// Removes a directory if it exists, ignoring NotFound errors.
pub async fn remove_dir_all_if_exists(path: &Path) -> Result<(), SnapError>
```

All functions return `SnapError::FileSystem` with:
- Operation name (for debugging)
- Path (to identify which file/directory failed)
- Error kind (from `std::io::ErrorKind`)

### Changes to `snap/client.rs`

**Before (4 occurrences):**
```rust
if !std::fs::exists(account_state_snapshots_dir).map_err(|_| {
    SnapError::SnapshotDir("State snapshots directory does not exist".to_string())
})? {
    std::fs::create_dir_all(account_state_snapshots_dir).map_err(|_| {
        SnapError::SnapshotDir("Failed to create state snapshots directory".to_string())
    })?;
}
```

**After:**
```rust
async_fs::ensure_dir_exists(account_state_snapshots_dir).await?;
```

Locations updated:
1. `request_account_range()` - account state snapshot directory (loop)
2. `request_account_range()` - account state snapshot directory (final
flush)
3. `request_storage_ranges()` - storage snapshot directory (loop)
4. `request_storage_ranges()` - storage snapshot directory (final flush)

### Changes to `sync/snap_sync.rs`

#### Code Hashes Processing

**Before:**
```rust
std::fs::create_dir_all(&code_hashes_snapshot_dir).map_err(|_| SyncError::CorruptPath)?;
// ...
for entry in std::fs::read_dir(&code_hashes_dir)
    .map_err(|_| SyncError::CodeHashesSnapshotsDirNotFound)?
{
    let entry = entry.map_err(|_| SyncError::CorruptPath)?;
    let snapshot_contents = std::fs::read(entry.path())
        .map_err(|err| SyncError::SnapshotReadError(entry.path(), err))?;
    // ...
}
std::fs::remove_dir_all(code_hashes_dir)
    .map_err(|_| SyncError::CodeHashesSnapshotsDirNotFound)?;
```

**After:**
```rust
async_fs::ensure_dir_exists(&code_hashes_snapshot_dir).await?;
// ...
let code_hash_files = async_fs::read_dir_paths(&code_hashes_dir).await?;
for file_path in code_hash_files {
    let snapshot_contents = async_fs::read_file(&file_path).await?;
    // ...
}
async_fs::remove_dir_all(&code_hashes_dir).await?;
```

#### Account Insertion (non-rocksdb)

**Before:**
```rust
for entry in std::fs::read_dir(account_state_snapshots_dir)
    .map_err(|_| SyncError::AccountStateSnapshotsDirNotFound)?
{
    let entry = entry.map_err(|err| SyncError::SnapshotReadError(...))?;
    let snapshot_path = entry.path();
    let snapshot_contents = std::fs::read(&snapshot_path)
        .map_err(|err| SyncError::SnapshotReadError(...))?;
    // ...
}
std::fs::remove_dir_all(account_state_snapshots_dir)
    .map_err(|_| SyncError::AccountStoragesSnapshotsDirNotFound)?;
```

**After:**
```rust
let snapshot_files = async_fs::read_dir_paths(account_state_snapshots_dir).await?;
for snapshot_path in snapshot_files {
    let snapshot_contents = async_fs::read_file(&snapshot_path).await?;
    // ...
}
async_fs::remove_dir_all(account_state_snapshots_dir).await?;
```

#### Storage Insertion (non-rocksdb)

Same pattern as account insertion - replaced `read_dir` + `read` +
`remove_dir_all` with async equivalents.

#### Account Insertion (rocksdb)

**Before:**
```rust
let file_paths: Vec<PathBuf> = std::fs::read_dir(account_state_snapshots_dir)
    .map_err(|_| SyncError::AccountStateSnapshotsDirNotFound)?
    .collect::<Result<Vec<_>, _>>()
    .map_err(|_| SyncError::AccountStateSnapshotsDirNotFound)?
    .into_iter()
    .map(|res| res.path())
    .collect();
// ...
std::fs::remove_dir_all(account_state_snapshots_dir)
    .map_err(|_| SyncError::AccountStateSnapshotsDirNotFound)?;
std::fs::remove_dir_all(get_rocksdb_temp_accounts_dir(datadir))
    .map_err(|e| SyncError::AccountTempDBDirNotFound(e.to_string()))?;
```

**After:**
```rust
let file_paths: Vec<PathBuf> = async_fs::read_dir_paths(account_state_snapshots_dir).await?;
// ...
async_fs::remove_dir_all(account_state_snapshots_dir).await?;
async_fs::remove_dir_all(&get_rocksdb_temp_accounts_dir(datadir)).await?;
```

#### Storage Insertion (rocksdb)

Same pattern - replaced sync fs operations with async equivalents.

### Summary of Changes

| File | Before | After | Change |
|------|--------|-------|--------|
| `snap/async_fs.rs` | - | 189 lines | New module |
| `snap/mod.rs` | - | +1 line | Module export |
| `snap/client.rs` | 42 lines | 4 lines | -38 lines (4 patterns
simplified) |
| `sync/snap_sync.rs` | 82 lines | 16 lines | -66 lines (16 calls
simplified) |
| `Cargo.toml` | - | +1 line | tempfile dev-dependency |
| `Cargo.lock` | - | +1 line | tempfile entry |

**Net change:** +223 lines, -93 lines = +130 lines (including 189 lines
of new module with tests)

## How to Test

### Unit Tests

Run the async_fs module tests:
```bash
cargo test -p ethrex-p2p async_fs
```

Expected output:
```
running 6 tests
test snap::async_fs::tests::test_remove_dir_all_if_exists ... ok
test snap::async_fs::tests::test_ensure_dir_exists_idempotent ... ok
test snap::async_fs::tests::test_remove_dir_all ... ok
test snap::async_fs::tests::test_ensure_dir_exists_creates_new ... ok
test snap::async_fs::tests::test_read_write_file ... ok
test snap::async_fs::tests::test_read_dir_paths ... ok

test result: ok. 6 passed; 0 failed; 0 ignored; 0 measured; 38 filtered out
```

### Compilation Tests

Verify both feature configurations compile:
```bash
# Without rocksdb (uses non-rocksdb insert_accounts/insert_storages)
cargo check -p ethrex-p2p

# With rocksdb (uses rocksdb-specific insert_accounts/insert_storages)
cargo check -p ethrex-p2p --features rocksdb
```

### Integration Testing

Run a snap sync on testnet to verify:
1. Account state snapshots are written and read correctly
2. Storage snapshots are written and read correctly
3. Code hash snapshots are written and read correctly
4. All cleanup operations complete successfully
5. No tokio runtime blocking warnings appear in logs

```bash
# Example: Run snap sync on Holesky
make start-holesky-metrics-docker
```

### Performance Verification

To verify async I/O is working:
1. Monitor CPU usage during snapshot write/read phases
2. Verify network operations continue during disk I/O
3. Check that `tokio::time::sleep` polling loops respond promptly

## Files Changed

| File | Description |
|------|-------------|
| `crates/networking/p2p/snap/async_fs.rs` | New async filesystem
helpers module |
| `crates/networking/p2p/snap/mod.rs` | Export async_fs module |
| `crates/networking/p2p/snap/client.rs` | Use async_fs for directory
creation |
| `crates/networking/p2p/sync/snap_sync.rs` | Use async_fs for all file
operations |
| `crates/networking/p2p/Cargo.toml` | Add tempfile dev-dependency |
| `Cargo.lock` | Updated lockfile |

## Checklist

- [x] Code compiles without rocksdb feature
- [x] Code compiles with rocksdb feature
- [x] All async_fs unit tests pass
- [x] No remaining `std::fs::` calls in snap sync code paths
- [x] Error handling preserved (SnapError::FileSystem)
- [x] Deterministic ordering maintained (read_dir_paths sorts results)

## Related

- Snap Sync Roadmap: Item 1.6 (Async Disk I/O)
- Parent branch: `refactor/snapsync-healing-unification`

---------

Co-authored-by: Esteban Dimitroff Hodi <esteban.dimitroff@lambdaclass.com>
Co-authored-by: ElFantasma <estebandh@gmail.com>

Author: pablodeymo

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## Perf
 
+### 2026-05-15
+
+- Replace synchronous disk I/O with async operations in snap sync [#6113](https://github.com/lambdaclass/ethrex/pull/6113)
+
 ### 2026-05-14
 
 - Skip `vm.run_execution()` for transfers to codeless EOAs [#6570](https://github.com/lambdaclass/ethrex/pull/6570)

Cargo.lock

@@ -58,6 +58,7 @@ crossbeam.workspace = true
 
 [dev-dependencies]
 hex-literal.workspace = true
+tempfile = "3"
 criterion = { version = "0.5", features = ["html_reports"] }
 
 [lib]

crates/networking/p2p/Cargo.toml

@@ -0,0 +1,136 @@
+//! Async file system utilities for snap sync
+//!
+//! Provides async wrappers around file system operations to avoid blocking the
+//! tokio runtime during disk I/O. `tokio::fs` is used for simple operations
+//! (internally delegates to `spawn_blocking`), while explicit `spawn_blocking`
+//! is used for `read_dir` to avoid async stream complexity.
+
+use std::path::{Path, PathBuf};
+
+use super::error::SnapError;
+
+/// Ensures a directory exists, creating it if necessary.
+pub async fn ensure_dir_exists(path: &Path) -> Result<(), SnapError> {
+    tokio::fs::create_dir_all(path)
+        .await
+        .map_err(|e| SnapError::FileSystem {
+            operation: "create directory",
+            path: path.to_path_buf(),
+            kind: e.kind(),
+        })
+}
+
+/// Reads all file paths from a directory, sorted alphabetically.
+///
+/// Uses `spawn_blocking` with sync `read_dir` since we always need all paths
+/// upfront (for `ingest_external_file` or batch iteration).
+pub async fn read_dir_paths(dir: &Path) -> Result<Vec<PathBuf>, SnapError> {
+    let dir = dir.to_path_buf();
+    tokio::task::spawn_blocking(move || {
+        let mut paths: Vec<PathBuf> = std::fs::read_dir(&dir)
+            .map_err(|e| SnapError::FileSystem {
+                operation: "read directory",
+                path: dir.clone(),
+                kind: e.kind(),
+            })?
+            .map(|entry| {
+                entry.map(|e| e.path()).map_err(|e| SnapError::FileSystem {
+                    operation: "read directory entry",
+                    path: dir.clone(),
+                    kind: e.kind(),
+                })
+            })
+            .collect::<Result<Vec<_>, _>>()?;
+        paths.sort();
+        Ok(paths)
+    })
+    .await?
+}
+
+/// Reads the contents of a file asynchronously.
+pub async fn read_file(path: &Path) -> Result<Vec<u8>, SnapError> {
+    tokio::fs::read(path)
+        .await
+        .map_err(|e| SnapError::FileSystem {
+            operation: "read file",
+            path: path.to_path_buf(),
+            kind: e.kind(),
+        })
+}
+
+/// Removes a directory and all its contents asynchronously.
+pub async fn remove_dir_all(path: &Path) -> Result<(), SnapError> {
+    tokio::fs::remove_dir_all(path)
+        .await
+        .map_err(|e| SnapError::FileSystem {
+            operation: "remove directory",
+            path: path.to_path_buf(),
+            kind: e.kind(),
+        })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tempfile::tempdir;
+
+    #[tokio::test]
+    async fn test_ensure_dir_exists_creates_new() {
+        let temp = tempdir().unwrap();
+        let new_dir = temp.path().join("new_dir");
+
+        assert!(!new_dir.exists());
+        ensure_dir_exists(&new_dir).await.unwrap();
+        assert!(new_dir.exists());
+    }
+
+    #[tokio::test]
+    async fn test_ensure_dir_exists_idempotent() {
+        let temp = tempdir().unwrap();
+        let existing_dir = temp.path().join("existing");
+        std::fs::create_dir(&existing_dir).unwrap();
+
+        ensure_dir_exists(&existing_dir).await.unwrap();
+        assert!(existing_dir.exists());
+    }
+
+    #[tokio::test]
+    async fn test_read_dir_paths() {
+        let temp = tempdir().unwrap();
+
+        std::fs::write(temp.path().join("b.txt"), b"b").unwrap();
+        std::fs::write(temp.path().join("a.txt"), b"a").unwrap();
+        std::fs::write(temp.path().join("c.txt"), b"c").unwrap();
+
+        let paths = read_dir_paths(temp.path()).await.unwrap();
+
+        assert_eq!(paths.len(), 3);
+        assert!(paths[0].ends_with("a.txt"));
+        assert!(paths[1].ends_with("b.txt"));
+        assert!(paths[2].ends_with("c.txt"));
+    }
+
+    #[tokio::test]
+    async fn test_read_file() {
+        let temp = tempdir().unwrap();
+        let file_path = temp.path().join("test.bin");
+
+        let data = b"hello world";
+        std::fs::write(&file_path, data).unwrap();
+
+        let read_data = read_file(&file_path).await.unwrap();
+        assert_eq!(read_data, data);
+    }
+
+    #[tokio::test]
+    async fn test_remove_dir_all() {
+        let temp = tempdir().unwrap();
+        let dir = temp.path().join("to_remove");
+        std::fs::create_dir(&dir).unwrap();
+        std::fs::write(dir.join("file.txt"), b"data").unwrap();
+
+        assert!(dir.exists());
+        remove_dir_all(&dir).await.unwrap();
+        assert!(!dir.exists());
+    }
+}

crates/networking/p2p/snap/async_fs.rs

@@ -16,7 +16,7 @@ use crate::{
             GetStorageRanges, GetTrieNodes, StorageRanges, TrieNodes,
         },
     },
-    snap::{constants::*, encodable_to_proof, error::SnapError},
+    snap::{async_fs, constants::*, encodable_to_proof, error::SnapError},
     sync::{AccountStorageRoots, SnapBlockSyncState, block_is_stale, update_pivot},
     utils::{
         AccountsWithStorage, dump_accounts_to_file, dump_storages_to_file,
@@ -160,13 +160,7 @@ pub async fn request_account_range(
                 .zip(current_account_states)
                 .collect::<Vec<(H256, AccountState)>>();
 
-            if !std::fs::exists(account_state_snapshots_dir).map_err(|_| {
-                SnapError::SnapshotDir("State snapshots directory does not exist".to_string())
-            })? {
-                std::fs::create_dir_all(account_state_snapshots_dir).map_err(|_| {
-                    SnapError::SnapshotDir("Failed to create state snapshots directory".to_string())
-                })?;
-            }
+            async_fs::ensure_dir_exists(account_state_snapshots_dir).await?;
 
             let account_state_snapshots_dir_cloned = account_state_snapshots_dir.to_path_buf();
             write_set.spawn(async move {
@@ -289,13 +283,7 @@ pub async fn request_account_range(
             .zip(current_account_states)
             .collect::<Vec<(H256, AccountState)>>();
 
-        if !std::fs::exists(account_state_snapshots_dir).map_err(|_| {
-            SnapError::SnapshotDir("State snapshots directory does not exist".to_string())
-        })? {
-            std::fs::create_dir_all(account_state_snapshots_dir).map_err(|_| {
-                SnapError::SnapshotDir("Failed to create state snapshots directory".to_string())
-            })?;
-        }
+        async_fs::ensure_dir_exists(account_state_snapshots_dir).await?;
 
         let path = get_account_state_snapshot_file(account_state_snapshots_dir, chunk_file);
         dump_accounts_to_file(&path, account_state_chunk)
@@ -603,15 +591,8 @@ pub async fn request_storage_ranges(
             let current_account_storages = std::mem::take(&mut current_account_storages);
             let snapshot = current_account_storages.into_values().collect::<Vec<_>>();
 
-            if !std::fs::exists(account_storages_snapshots_dir).map_err(|_| {
-                SnapError::SnapshotDir("Storage snapshots directory does not exist".to_string())
-            })? {
-                std::fs::create_dir_all(account_storages_snapshots_dir).map_err(|_| {
-                    SnapError::SnapshotDir(
-                        "Failed to create storage snapshots directory".to_string(),
-                    )
-                })?;
-            }
+            async_fs::ensure_dir_exists(account_storages_snapshots_dir).await?;
+
             let account_storages_snapshots_dir_cloned =
                 account_storages_snapshots_dir.to_path_buf();
             if !disk_joinset.is_empty() {
@@ -996,13 +977,8 @@ pub async fn request_storage_ranges(
     {
         let snapshot = current_account_storages.into_values().collect::<Vec<_>>();
 
-        if !std::fs::exists(account_storages_snapshots_dir).map_err(|_| {
-            SnapError::SnapshotDir("Storage snapshots directory does not exist".to_string())
-        })? {
-            std::fs::create_dir_all(account_storages_snapshots_dir).map_err(|_| {
-                SnapError::SnapshotDir("Failed to create storage snapshots directory".to_string())
-            })?;
-        }
+        async_fs::ensure_dir_exists(account_storages_snapshots_dir).await?;
+
         let path = get_account_storages_snapshot_file(account_storages_snapshots_dir, chunk_index);
         dump_storages_to_file(&path, snapshot).map_err(|_| {
             SnapError::SnapshotDir(format!(

crates/networking/p2p/snap/client.rs

@@ -7,7 +7,7 @@ use crate::rlpx::error::PeerConnectionError;
 use ethrex_rlp::error::RLPDecodeError;
 use ethrex_storage::error::StoreError;
 use ethrex_trie::TrieError;
-use spawned_concurrency::error::ActorError;
+use spawned_concurrency::ActorError;
 use std::io::ErrorKind;
 use std::path::PathBuf;
 use thiserror::Error;

crates/networking/p2p/snap/error.rs

@@ -12,6 +12,7 @@
 //! - `constants`: Protocol constants and configuration values
 //! - `error`: Unified error types for snap protocol operations
 
+pub mod async_fs;
 pub mod client;
 pub mod constants;
 pub mod error;