Ethereum bridge milestone 1.5: upgrade get_resume_cursor() to avoid full scan on listener start

contracts/ethereum/src/Bridge.sol

@@ -2,7 +2,7 @@
 pragma solidity ^0.8.30;
 
 interface IBridge {
-    event FundsReceived(uint64 eventId, address indexed sender, uint256 amount);
+    event FundsReceived(uint64 indexed eventId, address indexed sender, uint256 amount);
     function get_config()
         external
         view

contracts/ethereum/test/Bridge.t.sol

@@ -5,7 +5,7 @@ import {Test} from "forge-std/Test.sol";
 import {Bridge} from "../src/Bridge.sol";
 
 contract BridgeTest is Test {
-    event FundsReceived(uint64 eventId, address indexed sender, uint256 amount);
+    event FundsReceived(uint64 indexed eventId, address indexed sender, uint256 amount);
     bytes constant TEST_VALIDATOR_KEY =
         hex"021111111111111111111111111111111111111111111111111111111111111111";
     bytes constant TEST_VALIDATOR_KEY_2 =
@@ -40,7 +40,7 @@ contract BridgeTest is Test {
     function test_ReceiveEthEmitsEvent() public {
         vm.deal(nonAdmin, 1 ether);
 
-        vm.expectEmit(true, false, false, true, address(bridge));
+        vm.expectEmit(true, true, false, true, address(bridge));
         emit FundsReceived(0, nonAdmin, 0.1 ether);
 
         vm.prank(nonAdmin);

packages/kolme/src/listener/ethereum.rs

@@ -2,7 +2,7 @@ use crate::*;
 use alloy::{
     contract::{ContractInstance, Interface},
     json_abi::JsonAbi,
-    primitives::{Address, U256},
+    primitives::{Address, B256, U256},
     providers::Provider,
     rpc::types::eth::{BlockNumberOrTag, Filter, Log},
     sol,
@@ -13,7 +13,7 @@ use super::get_next_bridge_event_id;
 
 const ETH_NATIVE_DENOM: &str = "eth";
 
-sol! { event FundsReceived(uint64 eventId, address indexed sender, uint256 amount); }
+sol! { event FundsReceived(uint64 indexed eventId, address indexed sender, uint256 amount); }
 
 enum EthereumBridgeEvent {
     FundsReceived(FundsReceived),
@@ -28,6 +28,11 @@ impl EthereumBridgeEvent {
 
     fn from_log(log: &Log) -> Result<Option<Self>> {
         let Some(topic) = log.topic0().copied() else {
+            // having `topic0` as None would be unusual given our filters
+            tracing::debug!(
+                "Ignoring Ethereum log without topic0 at address {:#x}",
+                log.address()
+            );
             return Ok(None);
         };
 
@@ -37,7 +42,13 @@ impl EthereumBridgeEvent {
                     .map(|decoded| decoded.inner.data)
                     .map_err(|e| anyhow::anyhow!("Failed to decode FundsReceived: {e}"))?,
             )),
-            _ => None,
+            _ => {
+                tracing::debug!(
+                    "Ignoring Ethereum log with unsupported topic {topic:#x} at address {:#x}",
+                    log.address()
+                );
+                None
+            }
         })
     }
 
@@ -154,13 +165,9 @@ async fn get_resume_cursor<P: Provider>(
     contract: &ContractInstance<P>,
     next_bridge_event_id: BridgeEventId,
 ) -> Result<(u64, Option<u64>)> {
-    // On listener connecting, we need to find a point of synchronization between
-    // the sidechain and main blockchain (Ethereum).
-    // The simplest way is to scan all the blocks starting from genesis - thats the way
-    //   how it works now.
-    // CAUTION: being this way it is not ready for mainnet!
-    // A bit more optimal would be to scan since contract deployment.
-    // But best option would be to store the last block on the sidechain.
+    // On listener startup, we find sync point by scanning from genesis while
+    // filtering by contract address and indexed event ID.
+    // TODO: Switch from "since genesis" to "since contract deployment block".
     let latest = contract.provider().get_block_number().await?;
 
     if next_bridge_event_id == BridgeEventId::start() {
@@ -170,7 +177,9 @@ async fn get_resume_cursor<P: Provider>(
     let filter = Filter::new()
         .from_block(BlockNumberOrTag::Earliest)
         .to_block(latest)
-        .address(*contract.address());
+        .address(*contract.address())
+        .event_signature(FundsReceived::SIGNATURE_HASH)
+        .topic1(event_id_topic(next_bridge_event_id));
 
     for log in contract.provider().get_logs(&filter).await? {
         if log.removed {
@@ -181,14 +190,32 @@ async fn get_resume_cursor<P: Provider>(
             continue;
         };
 
-        if event.event_id() == next_bridge_event_id {
-            return Ok((log.block_number.unwrap_or(0), log.log_index));
-        }
+        anyhow::ensure!(
+            event.event_id() == next_bridge_event_id,
+            "Ethereum filtered resume query returned mismatched event ID. Expected {}, got {}",
+            next_bridge_event_id,
+            event.event_id()
+        );
+        return Ok((log.block_number.unwrap_or(0), log.log_index));
     }
 
+    tracing::warn!(
+        "Ethereum resume scan did not find expected event ID {} on contract {:#x}; scanned blocks 0..={} and will resume from latest+1",
+        next_bridge_event_id,
+        contract.address(),
+        latest
+    );
     Ok((latest.saturating_add(1), None))
 }
 
+/// Converts kolme's event id (u64) into 32-byte Ethereum topic value for log filtering
+fn event_id_topic(event_id: BridgeEventId) -> B256 {
+    let BridgeEventId(event_id) = event_id;
+    let mut bytes = [0u8; 32];
+    bytes[24..].copy_from_slice(&event_id.to_be_bytes());
+    B256::from(bytes)
+}
+
 async fn process_event<App: KolmeApp>(
     kolme: &Kolme<App>,
     secret: &SecretKey,
@@ -245,17 +272,23 @@ mod tests {
             .parse::<Address>()
             .unwrap();
 
+        let event_id = 7u64;
+        let mut event_id_topic = [0u8; 32];
+        event_id_topic[24..].copy_from_slice(&event_id.to_be_bytes());
+
         let mut sender_topic = [0u8; 32];
         sender_topic[12..].copy_from_slice(sender.as_slice());
 
-        let event_id = 7u64;
         let amount = U256::from(42u64);
-        let mut data = [0u8; 64];
-        data[24..32].copy_from_slice(&event_id.to_be_bytes());
-        data[32..64].copy_from_slice(&amount.to_be_bytes::<32>());
+        let mut data = [0u8; 32];
+        data[0..32].copy_from_slice(&amount.to_be_bytes::<32>());
 
         let log_data = LogData::new(
-            vec![FundsReceived::SIGNATURE_HASH, B256::from(sender_topic)],
+            vec![
+                FundsReceived::SIGNATURE_HASH,
+                B256::from(event_id_topic),
+                B256::from(sender_topic),
+            ],
             Bytes::copy_from_slice(&data),
         )
         .unwrap();