feat(cold): add eth_getLogs support and first_log_index to ReceiptContext

Add log filtering to cold storage following eth_getLogs semantics, and
include block-level log indexing for RPC response construction.

- Add `LogFilter` type with block range, address, and topic filters
- Add `RichLog` type with full block/tx context and block_log_index
- Add `ColdStorage::get_logs` with implementations for in-memory, SQLite,
  and PostgreSQL backends
- Add `first_log_index` to `ReceiptContext` (cherry-picked from #23)
- Replace `idx_logs_address` with composite `idx_logs_address_block`
- Factor out `row_to_log_row` helper in SQL backend
- Wire through task channel plumbing (request, handle, runner)
- Comprehensive conformance tests covering all filter combinations

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: prestwich

crates/cold-sql/migrations/001_initial.sql

@@ -79,7 +79,7 @@ CREATE TABLE IF NOT EXISTS logs (
     PRIMARY KEY (block_number, tx_index, log_index)
 );
 
-CREATE INDEX IF NOT EXISTS idx_logs_address ON logs (address);
+CREATE INDEX IF NOT EXISTS idx_logs_address_block ON logs (address, block_number);
 CREATE INDEX IF NOT EXISTS idx_logs_topic0 ON logs (topic0);
 
 CREATE TABLE IF NOT EXISTS signet_events (

crates/cold-sql/migrations/001_initial_pg.sql

@@ -77,7 +77,7 @@ CREATE TABLE IF NOT EXISTS logs (
     PRIMARY KEY (block_number, tx_index, log_index)
 );
 
-CREATE INDEX IF NOT EXISTS idx_logs_address ON logs (address);
+CREATE INDEX IF NOT EXISTS idx_logs_address_block ON logs (address, block_number);
 CREATE INDEX IF NOT EXISTS idx_logs_topic0 ON logs (topic0);
 
 CREATE TABLE IF NOT EXISTS signet_events (

crates/cold-sql/src/backend.rs

@@ -11,8 +11,8 @@ use crate::convert::{
 };
 use alloy::{consensus::Header, primitives::BlockNumber};
 use signet_cold::{
-    BlockData, ColdResult, ColdStorage, ColdStorageError, Confirmed, HeaderSpecifier,
-    ReceiptSpecifier, SignetEventsSpecifier, TransactionSpecifier, ZenithHeaderSpecifier,
+    BlockData, ColdResult, ColdStorage, ColdStorageError, Confirmed, HeaderSpecifier, LogFilter,
+    ReceiptSpecifier, RichLog, SignetEventsSpecifier, TransactionSpecifier, ZenithHeaderSpecifier,
 };
 use signet_storage_types::{
     ConfirmationMeta, DbSignetEvent, DbZenithHeader, Receipt, TransactionSigned,
@@ -269,20 +269,7 @@ impl SqlColdBackend {
         .fetch_all(&self.pool)
         .await?;
 
-        let logs = log_rows
-            .into_iter()
-            .map(|r| LogRow {
-                block_number: r.get("block_number"),
-                tx_index: r.get("tx_index"),
-                log_index: r.get("log_index"),
-                address: r.get("address"),
-                topic0: r.get("topic0"),
-                topic1: r.get("topic1"),
-                topic2: r.get("topic2"),
-                topic3: r.get("topic3"),
-                data: r.get("data"),
-            })
-            .collect();
+        let logs = log_rows.into_iter().map(|r| row_to_log_row(&r)).collect();
 
         receipt_from_rows(receipt, logs).map(Some)
     }
@@ -538,6 +525,20 @@ fn row_to_signet_event_row(r: &sqlx::any::AnyRow) -> SignetEventRow {
     }
 }
 
+fn row_to_log_row(r: &sqlx::any::AnyRow) -> LogRow {
+    LogRow {
+        block_number: r.get("block_number"),
+        tx_index: r.get("tx_index"),
+        log_index: r.get("log_index"),
+        address: r.get("address"),
+        topic0: r.get("topic0"),
+        topic1: r.get("topic1"),
+        topic2: r.get("topic2"),
+        topic3: r.get("topic3"),
+        data: r.get("data"),
+    }
+}
+
 fn row_to_zenith_header_row(r: &sqlx::any::AnyRow) -> ZenithHeaderRow {
     ZenithHeaderRow {
         block_number: r.get("block_number"),
@@ -655,20 +656,7 @@ impl ColdStorage for SqlColdBackend {
             .await
             .map_err(SqlColdError::from)?;
 
-            let logs = log_rows
-                .into_iter()
-                .map(|r| LogRow {
-                    block_number: r.get("block_number"),
-                    tx_index: r.get("tx_index"),
-                    log_index: r.get("log_index"),
-                    address: r.get("address"),
-                    topic0: r.get("topic0"),
-                    topic1: r.get("topic1"),
-                    topic2: r.get("topic2"),
-                    topic3: r.get("topic3"),
-                    data: r.get("data"),
-                })
-                .collect();
+            let logs = log_rows.into_iter().map(|r| row_to_log_row(&r)).collect();
 
             receipts.push(receipt_from_rows(receipt, logs)?);
         }
@@ -763,6 +751,104 @@ impl ColdStorage for SqlColdBackend {
             .collect()
     }
 
+    async fn get_logs(&self, filter: LogFilter) -> ColdResult<Vec<RichLog>> {
+        // Build dynamic SQL with positional $N placeholders.
+        // The correlated subquery computes block_log_index: the absolute
+        // position of each log among all logs in its block, leveraging the
+        // PK index on (block_number, tx_index, log_index).
+        let mut sql = String::from(
+            "SELECT l.*, h.block_hash, t.tx_hash, \
+               (SELECT COUNT(*) FROM logs l2 \
+                WHERE l2.block_number = l.block_number \
+                  AND (l2.tx_index < l.tx_index \
+                       OR (l2.tx_index = l.tx_index AND l2.log_index < l.log_index)) \
+               ) AS block_log_index \
+             FROM logs l \
+             JOIN headers h ON l.block_number = h.block_number \
+             JOIN transactions t ON l.block_number = t.block_number \
+               AND l.tx_index = t.tx_index \
+             WHERE l.block_number >= $1 AND l.block_number <= $2",
+        );
+        let mut params: Vec<Vec<u8>> = Vec::new();
+        let mut idx = 3u32;
+
+        // Address filter
+        if let Some(ref addrs) = filter.address {
+            if addrs.len() == 1 {
+                sql.push_str(&format!(" AND l.address = ${idx}"));
+                params.push(addrs[0].as_slice().to_vec());
+                idx += 1;
+            } else if !addrs.is_empty() {
+                let placeholders: String = addrs
+                    .iter()
+                    .enumerate()
+                    .map(|(i, _)| format!("${}", idx + i as u32))
+                    .collect::<Vec<_>>()
+                    .join(", ");
+                sql.push_str(&format!(" AND l.address IN ({placeholders})"));
+                for addr in addrs {
+                    params.push(addr.as_slice().to_vec());
+                }
+                idx += addrs.len() as u32;
+            }
+        }
+
+        // Topic filters
+        let topic_cols = ["l.topic0", "l.topic1", "l.topic2", "l.topic3"];
+        for (i, topic_filter) in filter.topics.iter().enumerate() {
+            let Some(values) = topic_filter else { continue };
+            if values.is_empty() {
+                continue;
+            }
+            if values.len() == 1 {
+                sql.push_str(&format!(" AND {} = ${idx}", topic_cols[i]));
+                params.push(values[0].as_slice().to_vec());
+                idx += 1;
+            } else {
+                let placeholders: String = values
+                    .iter()
+                    .enumerate()
+                    .map(|(j, _)| format!("${}", idx + j as u32))
+                    .collect::<Vec<_>>()
+                    .join(", ");
+                sql.push_str(&format!(" AND {} IN ({placeholders})", topic_cols[i]));
+                for v in values {
+                    params.push(v.as_slice().to_vec());
+                }
+                idx += values.len() as u32;
+            }
+        }
+
+        sql.push_str(" ORDER BY l.block_number, l.tx_index, l.log_index");
+
+        // Bind parameters and execute.
+        let mut query =
+            sqlx::query(&sql).bind(to_i64(filter.from_block)).bind(to_i64(filter.to_block));
+        for param in &params {
+            query = query.bind(param.as_slice());
+        }
+
+        let rows = query.fetch_all(&self.pool).await.map_err(SqlColdError::from)?;
+
+        rows.into_iter()
+            .map(|r| {
+                let log = row_to_log_row(&r).into_log();
+                let block_number = from_i64(r.get::<i64, _>("block_number"));
+                let block_hash_bytes: Vec<u8> = r.get("block_hash");
+                let tx_hash_bytes: Vec<u8> = r.get("tx_hash");
+                Ok(RichLog {
+                    log,
+                    block_number,
+                    block_hash: alloy::primitives::B256::from_slice(&block_hash_bytes),
+                    tx_hash: alloy::primitives::B256::from_slice(&tx_hash_bytes),
+                    tx_index: from_i64(r.get::<i64, _>("tx_index")),
+                    block_log_index: from_i64(r.get::<i64, _>("block_log_index")),
+                    tx_log_index: from_i64(r.get::<i64, _>("log_index")),
+                })
+            })
+            .collect::<ColdResult<Vec<_>>>()
+    }
+
     async fn get_latest_block(&self) -> ColdResult<Option<BlockNumber>> {
         let row = sqlx::query("SELECT block_number FROM metadata WHERE key = $1")
             .bind("latest_block")