perf(l1): two-CF receipts migration

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## Perf
 
+### 2026-04-29
+
+- Use fixed-width RECEIPTS keys and cursor iteration with early stop for eth_getTransactionReceipt; fix prefix iterator scan in get_transaction_location [#6548](https://github.com/lambdaclass/ethrex/pull/6548)
+
 ### 2026-04-27
 
 - Reduce peak disk usage during snap sync by moving SST files into the temp DB instead of copying [#6532](https://github.com/lambdaclass/ethrex/pull/6532)

crates/networking/p2p/rlpx/connection/server.rs

@@ -1242,7 +1242,7 @@ async fn handle_incoming_message(
                 let start_index = if i == 0 { first_block_receipt_index } else { 0 };
                 let block_receipts = state
                     .storage
-                    .get_receipts_for_block_from_index(hash, start_index)
+                    .get_receipts_for_block_from_index(hash, start_index, None)
                     .await?;
 
                 let mut block_receipt_list = Vec::new();

crates/networking/rpc/eth/block.rs

@@ -12,7 +12,7 @@ use crate::{
     utils::RpcErr,
 };
 use ethrex_common::types::{
-    Block, BlockBody, BlockHash, BlockHeader, BlockNumber, Receipt, calculate_base_fee_per_blob_gas,
+    Block, BlockBody, BlockHash, BlockHeader, Receipt, calculate_base_fee_per_blob_gas,
 };
 use ethrex_storage::Store;
 
@@ -177,7 +177,7 @@ impl RpcHandler for GetBlockReceiptsRequest {
             // Block not found
             _ => return Ok(Value::Null),
         };
-        let receipts = get_all_block_rpc_receipts(block_number, header, body, storage).await?;
+        let receipts = get_all_block_rpc_receipts(header, body, storage, None).await?;
 
         serde_json::to_value(&receipts).map_err(|error| RpcErr::Internal(error.to_string()))
     }
@@ -270,11 +270,11 @@ impl RpcHandler for GetRawReceipts {
         };
         let header = storage.get_block_header(block_number)?;
         let body = storage.get_block_body(block_number).await?;
-        let (header, body) = match (header, body) {
+        let (header, _body) = match (header, body) {
             (Some(header), Some(body)) => (header, body),
             _ => return Ok(Value::Null),
         };
-        let receipts: Vec<String> = get_all_block_receipts(block_number, header, body, storage)
+        let receipts: Vec<String> = get_all_block_receipts(header, storage)
             .await?
             .iter()
             .map(|receipt| {
@@ -329,11 +329,18 @@ impl RpcHandler for GetBlobBaseFee {
     }
 }
 
+/// Fetches RPC receipts for a block, optionally stopping after `target_index`.
+///
+/// When `target_index` is `Some(n)`, only receipts 0..=n are fetched using a
+/// cursor pass — this is the fast path for `eth_getTransactionReceipt` which
+/// only needs one receipt but requires preceding cumulative gas values.
+///
+/// When `target_index` is `None`, all receipts are fetched (for `eth_getBlockReceipts`).
 pub async fn get_all_block_rpc_receipts(
-    block_number: BlockNumber,
     header: BlockHeader,
     body: BlockBody,
     storage: &Store,
+    target_index: Option<u64>,
 ) -> Result<Vec<RpcReceipt>, RpcErr> {
     let mut receipts = Vec::new();
     // Check if this is the genesis block
@@ -353,16 +360,32 @@ pub async fn get_all_block_rpc_receipts(
         .try_into()
         .map_err(|_| RpcErr::Internal("blob_base_fee does not fit in u64".to_owned()))?;
     // Fetch receipt info from block
+    let block_hash = header.hash();
     let block_info = RpcReceiptBlockInfo::from_block_header(header);
-    // Fetch receipt for each tx in the block and add block and tx info
+    // Fetch receipts: only up to target_index+1 when set, otherwise all
+    let fetch_count = target_index
+        .map(|ti| (ti + 1) as usize)
+        .unwrap_or(body.transactions.len());
+    let all_receipts = storage
+        .get_receipts_for_block_from_index(&block_hash, 0, Some(fetch_count))
+        .await?;
+    if all_receipts.len() != fetch_count {
+        return Err(RpcErr::Internal(format!(
+            "Expected {} receipts, got {}",
+            fetch_count,
+            all_receipts.len()
+        )));
+    }
     let mut last_cumulative_gas_used = 0;
     let mut current_log_index = 0;
-    for (index, tx) in body.transactions.iter().enumerate() {
+    for (index, (tx, receipt)) in body
+        .transactions
+        .iter()
+        .zip(all_receipts.iter())
+        .enumerate()
+        .take(fetch_count)
+    {
         let index = index as u64;
-        let receipt = match storage.get_receipt(block_number, index).await? {
-            Some(receipt) => receipt,
-            _ => return Err(RpcErr::Internal("Could not get receipt".to_owned())),
-        };
         let gas_used = receipt.cumulative_gas_used - last_cumulative_gas_used;
         let tx_info = RpcReceiptTxInfo::from_transaction(
             tx.clone(),
@@ -385,23 +408,13 @@ pub async fn get_all_block_rpc_receipts(
 }
 
 pub async fn get_all_block_receipts(
-    block_number: BlockNumber,
     header: BlockHeader,
-    body: BlockBody,
     storage: &Store,
 ) -> Result<Vec<Receipt>, RpcErr> {
-    let mut receipts = Vec::new();
     // Check if this is the genesis block
     if header.parent_hash.is_zero() {
-        return Ok(receipts);
+        return Ok(Vec::new());
     }
-    for (index, _) in body.transactions.iter().enumerate() {
-        let index = index as u64;
-        let receipt = match storage.get_receipt(block_number, index).await? {
-            Some(receipt) => receipt,
-            _ => return Err(RpcErr::Internal("Could not get receipt".to_owned())),
-        };
-        receipts.push(receipt);
-    }
-    Ok(receipts)
+    let block_hash = header.hash();
+    Ok(storage.get_receipts_for_block(&block_hash).await?)
 }

crates/networking/rpc/eth/transaction.rs

@@ -292,7 +292,7 @@ impl RpcHandler for GetTransactionReceiptRequest {
             "Requested receipt for transaction {:#x}",
             self.transaction_hash,
         );
-        let (block_number, block_hash, index) = match storage
+        let (_block_number, block_hash, index) = match storage
             .get_transaction_location(self.transaction_hash)
             .await?
         {
@@ -304,11 +304,10 @@ impl RpcHandler for GetTransactionReceiptRequest {
             None => return Ok(Value::Null),
         };
         let receipts =
-            block::get_all_block_rpc_receipts(block_number, block.header, block.body, storage)
+            block::get_all_block_rpc_receipts(block.header, block.body, storage, Some(index))
                 .await?;
 
-        serde_json::to_value(receipts.get(index as usize))
-            .map_err(|error| RpcErr::Internal(error.to_string()))
+        serde_json::to_value(receipts.last()).map_err(|error| RpcErr::Internal(error.to_string()))
     }
 }
 

crates/storage/Cargo.toml

@@ -26,6 +26,8 @@ tokio = { workspace = true, features = ["rt", "sync"] }
 fastbloom = "0.14"
 rayon.workspace = true
 lru.workspace = true
+libc = "0.2"
+tracing-subscriber = { workspace = true }
 
 [features]
 default = []
@@ -35,9 +37,20 @@ rocksdb = ["dep:rocksdb"]
 tempfile.workspace = true
 tokio = { workspace = true, features = ["full"] }
 
+
 [lib]
 path = "./lib.rs"
 
+[[bin]]
+name = "seed_migration_test"
+path = "seed_migration_test.rs"
+required-features = ["rocksdb"]
+
+[[bin]]
+name = "bench_migration"
+path = "bench_migration.rs"
+required-features = ["rocksdb"]
+
 [lints.clippy]
 unwrap_used = "deny"
 redundant_clone = "warn"

crates/storage/api/tables.rs

@@ -30,11 +30,16 @@ pub const ACCOUNT_CODES: &str = "account_codes";
 /// - [`u8; 8`] = `code_length.to_be_bytes()`
 pub const ACCOUNT_CODE_METADATA: &str = "account_code_metadata";
 
-/// Receipts column family: [`Vec<u8>`] => [`Vec<u8>`]
-/// - [`Vec<u8>`] = `(block_hash, index).encode_to_vec()`
-/// - [`Vec<u8>`] = `receipt.encode_to_vec()`
+/// Receipts column family (legacy, pre-v2): [`Vec<u8>`] => [`Vec<u8>`]
+/// Kept only for migration reads; dropped automatically on next startup.
 pub const RECEIPTS: &str = "receipts";
 
+/// Receipts v2 column family: [`Vec<u8>`] => [`Vec<u8>`]
+/// - Key: `block_hash (32B) || index (8B big-endian u64)` — fixed-width raw key
+///   enabling cursor-based prefix iteration by block hash.
+/// - Value: `receipt.encode_to_vec()`
+pub const RECEIPTS_V2: &str = "receipts_v2";
+
 /// Transaction locations column family: [`Vec<u8>`] => [`Vec<u8>`]
 /// - [`Vec<u8>`] = Composite key
 ///    ```rust,no_run
@@ -102,7 +107,7 @@ pub const MISC_VALUES: &str = "misc_values";
 /// - [`Vec<u8>`] = `serde_json::to_vec(&witness)`
 pub const EXECUTION_WITNESSES: &str = "execution_witnesses";
 
-pub const TABLES: [&str; 19] = [
+pub const TABLES: [&str; 20] = [
     CHAIN_DATA,
     ACCOUNT_CODES,
     ACCOUNT_CODE_METADATA,
@@ -113,6 +118,7 @@ pub const TABLES: [&str; 19] = [
     PENDING_BLOCKS,
     TRANSACTION_LOCATIONS,
     RECEIPTS,
+    RECEIPTS_V2,
     SNAP_STATE,
     INVALID_CHAINS,
     ACCOUNT_TRIE_NODES,

crates/storage/backend/rocksdb.rs

@@ -1,6 +1,6 @@
 use crate::api::tables::{
     ACCOUNT_CODES, ACCOUNT_FLATKEYVALUE, ACCOUNT_TRIE_NODES, BLOCK_NUMBERS, BODIES,
-    CANONICAL_BLOCK_HASHES, FULLSYNC_HEADERS, HEADERS, RECEIPTS, STORAGE_FLATKEYVALUE,
+    CANONICAL_BLOCK_HASHES, FULLSYNC_HEADERS, HEADERS, RECEIPTS_V2, STORAGE_FLATKEYVALUE,
     STORAGE_TRIE_NODES, TRANSACTION_LOCATIONS,
 };
 use crate::api::{
@@ -68,7 +68,7 @@ impl RocksDBBackend {
             BLOCK_NUMBERS,
             HEADERS,
             BODIES,
-            RECEIPTS,
+            RECEIPTS_V2,
             TRANSACTION_LOCATIONS,
             FULLSYNC_HEADERS,
         ];
@@ -165,7 +165,7 @@ impl RocksDBBackend {
                     block_opts.set_block_cache(&block_cache);
                     cf_opts.set_block_based_table_factory(&block_opts);
                 }
-                RECEIPTS => {
+                RECEIPTS_V2 => {
                     cf_opts.set_write_buffer_size(128 * 1024 * 1024); // 128MB
                     cf_opts.set_max_write_buffer_number(3);
                     cf_opts.set_target_file_size_base(256 * 1024 * 1024); // 256MB

crates/storage/bench_migration.rs

@@ -0,0 +1,100 @@
+/// Standalone binary to benchmark the v1→v2 RECEIPTS migration.
+///
+/// Usage: bench_migration <db_path>
+///
+/// Opens the RocksDB database, runs the two-CF migration (receipts → receipts_v2),
+/// and reports wall-clock time and peak RSS.
+///
+/// Prerequisites:
+///   1. Run `seed_migration_test <db_path>` to seed 150M old-format entries
+///   2. Ensure metadata.json has {"schema_version": 1} (or just don't create one)
+///   3. Run this binary
+use std::time::Instant;
+
+fn get_rss_mb() -> Option<f64> {
+    #[cfg(target_os = "macos")]
+    {
+        use std::mem;
+        let mut info: libc::rusage = unsafe { mem::zeroed() };
+        let ret = unsafe { libc::getrusage(libc::RUSAGE_SELF, &mut info) };
+        if ret == 0 {
+            // macOS reports maxrss in bytes
+            Some(info.ru_maxrss as f64 / (1024.0 * 1024.0))
+        } else {
+            None
+        }
+    }
+    #[cfg(target_os = "linux")]
+    {
+        use std::mem;
+        let mut info: libc::rusage = unsafe { mem::zeroed() };
+        let ret = unsafe { libc::getrusage(libc::RUSAGE_SELF, &mut info) };
+        if ret == 0 {
+            // Linux reports maxrss in kilobytes
+            Some(info.ru_maxrss as f64 / 1024.0)
+        } else {
+            None
+        }
+    }
+    #[cfg(not(any(target_os = "macos", target_os = "linux")))]
+    {
+        None
+    }
+}
+
+fn main() {
+    // Initialize tracing so migration progress logs are visible
+    tracing_subscriber::fmt()
+        .with_target(false)
+        .with_env_filter(
+            tracing_subscriber::EnvFilter::try_from_default_env()
+                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
+        )
+        .init();
+
+    let args: Vec<String> = std::env::args().collect();
+    if args.len() != 2 {
+        eprintln!("Usage: {} <db_path>", args[0]);
+        std::process::exit(1);
+    }
+    let db_path = &args[1];
+
+    println!("Opening database at: {db_path}");
+    let rss_before = get_rss_mb();
+
+    let backend = ethrex_storage::backend::rocksdb::RocksDBBackend::open(db_path)
+        .expect("Failed to open RocksDB");
+
+    let rss_after_open = get_rss_mb();
+    println!(
+        "Database opened. RSS after open: {:.1} MB",
+        rss_after_open.unwrap_or(0.0)
+    );
+
+    // Run the migration
+    println!("Starting migration v1→v2 (two-CF: receipts → receipts_v2)...");
+    let start = Instant::now();
+
+    ethrex_storage::migrations::run_pending_migrations(
+        &backend,
+        std::path::Path::new(db_path),
+        1, // pretend we're at v1
+    )
+    .expect("Migration failed");
+
+    let elapsed = start.elapsed();
+    let rss_after = get_rss_mb();
+
+    println!("\n=== Migration Benchmark Results ===");
+    println!("Wall-clock time: {:.1}s", elapsed.as_secs_f64());
+    if let Some(before) = rss_before {
+        println!("RSS before open:     {:.1} MB", before);
+    }
+    if let Some(after_open) = rss_after_open {
+        println!("RSS after open:      {:.1} MB", after_open);
+    }
+    if let Some(after) = rss_after {
+        println!("Peak RSS (maxrss):   {:.1} MB", after);
+    }
+    println!("===================================");
+}

crates/storage/lib.rs

@@ -85,7 +85,7 @@ pub use store::{
 /// When bumping this version, add a corresponding migration function to
 /// `migrations::MIGRATIONS`. The migration framework will automatically
 /// upgrade existing databases instead of requiring a full resync.
-pub const STORE_SCHEMA_VERSION: u64 = 1;
+pub const STORE_SCHEMA_VERSION: u64 = 2;
 
 /// Name of the file storing the metadata about the database.
 ///