fix: reduce excess allocations (ENG-2036) (#49)

prestwich · claude · web-flow · commit 6bc311d52077 · 2026-04-09T14:33:24.000+01:00
* fix: reduce excess allocations in storage workspace (ENG-2036)

Eliminate unnecessary clones and intermediate collections:
- drain_above: zip iterators instead of indexed .get().cloned()
- ColdReceipt::new: destructure and into_iter() logs instead of cloning
- write_block_to_tx: single-pass loop instead of intermediate Vec
- SqlColdBackend::new: match on &amp;str instead of .to_owned()

Add ENG-2036 comments for intentionally deferred items (Filter clone,
eager into_owned in traversals, format!() SQL queries).

Co-Authored-By: Claude Opus 4.6 (1M context) &lt;noreply@anthropic.com&gt;

* fix: add debug_assert for tx/receipt length match

Co-Authored-By: Claude Opus 4.6 (1M context) &lt;noreply@anthropic.com&gt;

---------

Co-authored-by: Claude Opus 4.6 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/crates/cold-mdbx/src/backend.rs b/crates/cold-mdbx/src/backend.rs
@@ -40,31 +40,25 @@ fn write_block_to_tx(
     // Hash-keyed indices use put (keys are not sequential)
     tx.queue_put::<ColdBlockHashIndex>(&data.header.hash(), &block)?;
 
-    // Store transactions, senders, and build hash index
-    let tx_meta: Vec<_> = data
-        .transactions
-        .iter()
-        .enumerate()
-        .map(|(idx, recovered_tx)| {
-            let tx_idx = idx as u64;
-            let sender = recovered_tx.signer();
-            let tx_signed: &TransactionSigned = recovered_tx;
-            tx.queue_append_dual::<ColdTransactions>(&block, &tx_idx, tx_signed)?;
-            tx.queue_append_dual::<ColdTxSenders>(&block, &tx_idx, &sender)?;
-            tx.queue_put::<ColdTxHashIndex>(tx_signed.hash(), &TxLocation::new(block, tx_idx))?;
-            Ok((*tx_signed.hash(), sender))
-        })
-        .collect::<Result<_, MdbxColdError>>()?;
-
-    // Compute and store IndexedReceipts with precomputed metadata
+    // Store transactions, senders, hash indices, and receipts in a single
+    // pass to avoid an intermediate Vec of (tx_hash, sender) tuples.
     let mut first_log_index = 0u64;
     let mut prior_cumulative_gas = 0u64;
-    for (idx, (receipt, (tx_hash, sender))) in data.receipts.into_iter().zip(tx_meta).enumerate() {
+    debug_assert_eq!(data.transactions.len(), data.receipts.len());
+    for (idx, (recovered_tx, receipt)) in data.transactions.iter().zip(data.receipts).enumerate() {
+        let tx_idx = idx as u64;
+        let sender = recovered_tx.signer();
+        let tx_signed: &TransactionSigned = recovered_tx;
+        tx.queue_append_dual::<ColdTransactions>(&block, &tx_idx, tx_signed)?;
+        tx.queue_append_dual::<ColdTxSenders>(&block, &tx_idx, &sender)?;
+        tx.queue_put::<ColdTxHashIndex>(tx_signed.hash(), &TxLocation::new(block, tx_idx))?;
+
+        let tx_hash = *tx_signed.hash();
         let gas_used = receipt.inner.cumulative_gas_used - prior_cumulative_gas;
         prior_cumulative_gas = receipt.inner.cumulative_gas_used;
         let ir = IndexedReceipt { receipt, tx_hash, first_log_index, gas_used, sender };
         first_log_index += ir.receipt.inner.logs.len() as u64;
-        tx.queue_append_dual::<ColdReceipts>(&block, &(idx as u64), &ir)?;
+        tx.queue_append_dual::<ColdReceipts>(&block, &tx_idx, &ir)?;
     }
 
     for (idx, event) in data.signet_events.iter().enumerate() {
@@ -731,6 +725,9 @@ impl ColdStorageRead for MdbxColdBackend {
 
     async fn produce_log_stream(&self, filter: &Filter, params: signet_cold::StreamParams) {
         let env = self.env.clone();
+        // ENG-2036: clone required to move into spawn_blocking. Eliminating
+        // this would require changing the ColdStorage trait to pass owned
+        // Filter, which is a cross-crate API change.
         let filter = filter.clone();
         let std_deadline = params.deadline.into_std();
         let _ = tokio::task::spawn_blocking(move || {
diff --git a/crates/cold-sql/src/backend.rs b/crates/cold-sql/src/backend.rs
@@ -112,21 +112,20 @@ impl SqlColdBackend {
     pub async fn new(pool: AnyPool) -> Result<Self, SqlColdError> {
         // Detect backend from a pooled connection.
         let conn = pool.acquire().await?;
-        let backend = conn.backend_name().to_owned();
-        drop(conn);
+        let backend = conn.backend_name();
 
-        let migration = match backend.as_str() {
-            "PostgreSQL" => include_str!("../migrations/001_initial_pg.sql"),
-            "SQLite" => include_str!("../migrations/001_initial.sql"),
+        let (migration, is_postgres) = match backend {
+            "PostgreSQL" => (include_str!("../migrations/001_initial_pg.sql"), true),
+            "SQLite" => (include_str!("../migrations/001_initial.sql"), false),
             other => {
                 return Err(SqlColdError::Convert(format!(
                     "unsupported database backend: {other}"
                 )));
             }
         };
+        drop(conn);
         // Execute via pool to ensure the migration uses the same
         // connection that subsequent queries will use.
-        let is_postgres = backend == "PostgreSQL";
         sqlx::raw_sql(migration).execute(&pool).await?;
         Ok(Self { pool, is_postgres })
     }
@@ -247,6 +246,10 @@ impl SqlColdBackend {
         // Build the parameterised query once. $1 is the block number
         // (bound per iteration); remaining parameters are the address
         // and topic filters from the user's request.
+        //
+        // ENG-2036: format!() allocates here, but the dynamic filter clause
+        // makes this unavoidable. sqlx's prepared-statement cache mitigates
+        // repeated execution cost.
         let (filter_clause, filter_params) = build_log_filter_clause(filter, 2);
         let data_sql = format!(
             "SELECT l.*, h.block_hash, h.timestamp AS block_timestamp, t.tx_hash, \
@@ -1348,6 +1351,8 @@ impl ColdStorageRead for SqlColdBackend {
         let to = filter.get_to_block().unwrap_or(u64::MAX);
 
         // Build WHERE clause: block range ($1, $2) + address/topic filters.
+        // ENG-2036: dynamic filter clause requires format!(); mitigated by
+        // sqlx prepared-statement cache.
         let (filter_clause, params) = build_log_filter_clause(filter, 3);
         let where_clause = format!("l.block_number >= $1 AND l.block_number <= $2{filter_clause}");
 
diff --git a/crates/cold/src/cold_receipt.rs b/crates/cold/src/cold_receipt.rs
@@ -47,40 +47,42 @@ impl ColdReceipt {
         let block_hash = header.hash();
         let block_timestamp = header.timestamp;
 
-        let rpc_logs: Vec<RpcLog> = ir
-            .receipt
+        // Destructure to move fields instead of cloning logs.
+        let IndexedReceipt { receipt: typed_receipt, tx_hash, first_log_index, gas_used, sender } =
+            ir;
+        let tx_type = typed_receipt.tx_type;
+        let status = typed_receipt.inner.status;
+        let cumulative_gas_used = typed_receipt.inner.cumulative_gas_used;
+
+        let rpc_logs: Vec<RpcLog> = typed_receipt
             .inner
             .logs
-            .iter()
+            .into_iter()
             .enumerate()
             .map(|(log_idx, log)| RpcLog {
-                inner: log.clone(),
+                inner: log,
                 block_hash: Some(block_hash),
                 block_number: Some(block_number),
                 block_timestamp: Some(block_timestamp),
-                transaction_hash: Some(ir.tx_hash),
+                transaction_hash: Some(tx_hash),
                 transaction_index: Some(transaction_index),
-                log_index: Some(ir.first_log_index + log_idx as u64),
+                log_index: Some(first_log_index + log_idx as u64),
                 removed: false,
             })
             .collect();
 
-        let receipt = ConsensusReceipt {
-            status: ir.receipt.inner.status,
-            cumulative_gas_used: ir.receipt.inner.cumulative_gas_used,
-            logs: rpc_logs,
-        };
+        let receipt = ConsensusReceipt { status, cumulative_gas_used, logs: rpc_logs };
 
         Self {
             receipt,
-            tx_type: ir.receipt.tx_type,
-            tx_hash: ir.tx_hash,
-            gas_used: ir.gas_used,
+            tx_type,
+            tx_hash,
+            gas_used,
             block_number,
             block_hash,
             block_timestamp,
             transaction_index,
-            from: ir.sender,
+            from: sender,
         }
     }
 }
diff --git a/crates/hot/src/model/traverse/dual_key.rs b/crates/hot/src/model/traverse/dual_key.rs
@@ -85,6 +85,9 @@ pub trait DualKeyTraverse<E: HotKvReadError> {
     where
         Self: Sized,
     {
+        // ENG-2036: into_owned() here eagerly materializes the first entry,
+        // wasting an allocation if the iterator is never consumed. Deferring
+        // this requires lifetime changes to the iterator struct.
         let first_entry =
             self.first()?.map(|(k1, k2, v)| (k1.into_owned(), k2.into_owned(), v.into_owned()));
         Ok(RawDualKeyIter {
@@ -107,6 +110,7 @@ pub trait DualKeyTraverse<E: HotKvReadError> {
     where
         Self: Sized,
     {
+        // ENG-2036: same eager into_owned() as iter() above.
         let first_entry = self
             .next_dual_above(k1, k2)?
             .map(|(k1, k2, v)| (k1.into_owned(), k2.into_owned(), v.into_owned()));
diff --git a/crates/hot/src/model/traverse/kv.rs b/crates/hot/src/model/traverse/kv.rs
@@ -43,6 +43,9 @@ pub trait KvTraverse<E: HotKvReadError> {
     where
         Self: Sized,
     {
+        // ENG-2036: into_owned() here eagerly materializes the first entry,
+        // wasting an allocation if the iterator is never consumed. Deferring
+        // this requires lifetime changes to the iterator struct.
         let first_entry = self.first()?.map(|(k, v)| (k.into_owned(), v.into_owned()));
         Ok(RawKvIter {
             cursor: self,
@@ -65,6 +68,7 @@ pub trait KvTraverse<E: HotKvReadError> {
     where
         Self: Sized,
     {
+        // ENG-2036: same eager into_owned() as iter() above.
         let first_entry = self.lower_bound(key)?.map(|(k, v)| (k.into_owned(), v.into_owned()));
         Ok(RawKvIter {
             cursor: self,
diff --git a/crates/storage/src/unified.rs b/crates/storage/src/unified.rs
@@ -266,14 +266,13 @@ impl<H: HotKv> UnifiedStorage<H> {
         // 3. Atomically drain cold (best-effort — failure = normal cold lag)
         let cold_receipts = self.cold.drain_above(block).await.unwrap_or_default();
 
-        // 4. Assemble drained blocks (zip headers with receipts, default empty)
+        // 4. Assemble drained blocks (zip headers with receipts, default empty).
+        // Pad cold_receipts to match headers length so zip consumes both
+        // without cloning.
         let drained = headers
             .into_iter()
-            .enumerate()
-            .map(|(i, header)| {
-                let receipts = cold_receipts.get(i).cloned().unwrap_or_default();
-                DrainedBlock { header, receipts }
-            })
+            .zip(cold_receipts.into_iter().chain(std::iter::repeat_with(Vec::new)))
+            .map(|(header, receipts)| DrainedBlock { header, receipts })
             .collect();
 
         Ok(drained)
