diff --git a/crates/ingest/src/evm.rs b/crates/ingest/src/evm.rs new file mode 100644 index 0000000..3f6c07a --- /dev/null +++ b/crates/ingest/src/evm.rs @@ -0,0 +1,873 @@ +//! EVM deposit detection via ERC-20 Transfer logs. +//! +//! Scans `eth_getLogs` for registered ERC-20 token contracts with a block-number cursor, +//! advancing only after a log is durably processed so a crash resumes exactly-once — the +//! same guarantee the Horizon path gives via its paging-token cursor. +//! +//! # Security: contract address verification +//! +//! **The most important invariant in this module:** logs are matched on the *emitting contract +//! address* (`log.address`), not on topics alone. Any contract can emit a `Transfer` event with +//! arbitrary topics — including a deposit address in the `to` topic and a huge value in the data +//! field. Attributing on topics alone would let an attacker mint balances for free by deploying a +//! hostile contract that emits fake `Transfer` events. +//! +//! Before attributing a log, `EvmIngestor::process_log` verifies that `log.address` matches one +//! of the registered token contracts. If it does not, the log is `Skipped` regardless of its +//! topic contents. +//! +//! # Native ETH deposits +//! +//! Native ETH transfers emit **no logs**. They are only detectable by inspecting block +//! transactions (`eth_getBlockByNumber`) or via `debug_traceBlock`, both of which are +//! significantly more expensive than a targeted `eth_getLogs` filter. For v1, **native ETH +//! deposits are out of scope**. This worker detects ERC-20 stablecoin transfers only. +//! +//! This is an explicit, documented decision rather than a silent omission: a customer sending +//! ETH directly to their deposit address will not have it credited. Operators should communicate +//! this to their users, and the `docs/ingest-integration.md` documents it in the EVM section. +//! +//! # Unregistered tokens (quarantine) +//! +//! A `Transfer` to a known deposit address for a token that is not in the registry is recorded +//! with `address_id = NULL` rather than attributed or dropped. This preserves auditability while +//! preventing unregistered (potentially fee-on-transfer or rebasing) tokens from creating +//! spendable balances. +//! +//! # EVM deposit confirmation +//! +//! Deposits are recorded as `unconfirmed`. Crediting is gated on `#222`. Merging this worker +//! before `#222` lands must not create spendable balances — this is enforced at the DB level: +//! `Store::record_evm_deposit` always inserts `status = 'unconfirmed'`. + +#![forbid(unsafe_code)] + +use octo_store::{NewEvmDeposit, Store}; +use reqwest::Client; +use serde::Deserialize; +use serde_json::{Value, json}; +use std::collections::HashSet; +use std::time::Duration; +use uuid::Uuid; + +use crate::Processed; + +// --------------------------------------------------------------------------- +// Transfer(address,address,uint256) event signature. +// keccak256("Transfer(address,address,uint256)") +// --------------------------------------------------------------------------- +const TRANSFER_TOPIC: &str = + "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"; + +/// Maximum block range per `eth_getLogs` request before we start bisecting. +/// +/// This is deliberately conservative. Different providers have different caps: +/// - Alchemy: 2 000 blocks +/// - Infura: 10 000 blocks +/// - Quicknode: varies +/// - Self-hosted Geth/Reth: unlimited but bounded by memory +/// +/// We start at 2 000 to be safe across all providers and let the adaptive bisection grow or +/// shrink the range based on actual `RangeTooLarge` responses. +const DEFAULT_BLOCK_RANGE: u64 = 2_000; + +/// Minimum block range after bisection. Below this the log scan cannot make progress. +const MIN_BLOCK_RANGE: u64 = 1; + +/// Maximum consecutive `RangeTooLarge` bisections before returning an error rather than +/// spinning into an infinite loop. +const MAX_BISECTIONS: u32 = 16; + +// --------------------------------------------------------------------------- +// JSON-RPC types +// --------------------------------------------------------------------------- + +/// A JSON-RPC 2.0 log entry from `eth_getLogs`. +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct EvmLog { + /// The contract address that emitted this log (`0x`-prefixed, EIP-55 checksummed by + /// well-behaved nodes, but may be all-lowercase from others — always compare + /// case-insensitively). + pub address: String, + /// The transaction hash containing this log. + pub transaction_hash: String, + /// Topics: `[topic0, topic1, topic2, ...]`. For an ERC-20 `Transfer`: + /// - `topics[0]` = `Transfer` signature hash + /// - `topics[1]` = `from` address (left-padded to 32 bytes) + /// - `topics[2]` = `to` address (left-padded to 32 bytes) + pub topics: Vec, + /// ABI-encoded log data: for ERC-20 `Transfer`, this is the `uint256` value. + pub data: String, + /// The block number containing this log (`0x`-prefixed hex). + pub block_number: String, + /// Index of this log entry within the transaction (0-based, `0x`-prefixed hex). + pub log_index: String, + /// Whether the log was removed by a reorg. Logs with `removed: true` must be ignored. + #[serde(default)] + pub removed: bool, +} + +impl EvmLog { + /// The `to` address from an ERC-20 Transfer event's indexed `topics[2]`, normalised to a + /// 20-byte (40-hex-char) lowercase address without `0x` prefix. + /// + /// EVM topics are left-padded to 32 bytes. An `address` topic looks like: + /// `0x000000000000000000000000<20-byte-address>`. + /// We strip the `0x` prefix and the 24 zero-padding characters, leaving the 40-char address, + /// then normalise to lowercase. + pub fn to_address(&self) -> Option { + let raw = self.topics.get(2)?.strip_prefix("0x")?; + // 32 bytes = 64 hex chars; 24 of those are leading zeros for an address topic. + if raw.len() != 64 { + return None; + } + let addr = &raw[24..]; // last 40 chars = 20 bytes + Some(format!("0x{}", addr.to_lowercase())) + } + + /// The `from` address from `topics[1]`, same normalisation as `to_address`. + pub fn from_address(&self) -> Option { + let raw = self.topics.get(1)?.strip_prefix("0x")?; + if raw.len() != 64 { + return None; + } + let addr = &raw[24..]; + Some(format!("0x{}", addr.to_lowercase())) + } + + /// The transfer amount from `data` as an `i64`. ERC-20 amounts are `uint256`, but for + /// practical deposit amounts (stablecoins, bounded by reasonable custodial limits) they + /// fit in `i64`. Values that overflow are logged and skipped. + /// + /// `data` for an ERC-20 Transfer is the ABI-encoded `uint256`: 32 bytes = 64 hex chars, + /// `0x`-prefixed. + pub fn amount_i64(&self) -> Option { + let hex = self.data.strip_prefix("0x")?; + if hex.len() > 16 { + // If the upper bytes are non-zero, the value exceeds u64::MAX. We skip those rather + // than truncating silently — an overflow here would misattribute the amount. + let upper = &hex[..hex.len().saturating_sub(16)]; + if upper.chars().any(|c| c != '0') { + return None; + } + } + let lower = &hex[hex.len().saturating_sub(16)..]; + let v = u64::from_str_radix(lower, 16).ok()?; + i64::try_from(v).ok() + } + + /// Block number as `u64`, parsed from `0x`-prefixed hex. + pub fn block_number_u64(&self) -> Option { + parse_hex_u64(&self.block_number) + } + + /// Log index as `u32`, parsed from `0x`-prefixed hex. + pub fn log_index_u32(&self) -> Option { + parse_hex_u64(&self.log_index).and_then(|v| u32::try_from(v).ok()) + } +} + +fn parse_hex_u64(s: &str) -> Option { + let hex = s.strip_prefix("0x")?; + u64::from_str_radix(hex, 16).ok() +} + +// --------------------------------------------------------------------------- +// EVM JSON-RPC errors +// --------------------------------------------------------------------------- + +/// Typed errors returned by the EVM JSON-RPC client. +#[derive(Debug, thiserror::Error)] +pub enum EvmRpcError { + /// The requested block range is too large for this provider. The ingestor will bisect and + /// retry. + #[error("eth_getLogs block range too large (provider limit exceeded)")] + RangeTooLarge, + + /// HTTP transport failure. + #[error("HTTP transport error: {0}")] + Transport(#[from] reqwest::Error), + + /// JSON-RPC protocol-level error (HTTP 200 with an `error` member). + #[error("JSON-RPC error {code}: {message}")] + JsonRpc { code: i64, message: String }, + + /// Unexpected response shape (missing fields, wrong types). + #[error("unexpected RPC response: {0}")] + UnexpectedResponse(String), +} + +// --------------------------------------------------------------------------- +// Low-level RPC client +// --------------------------------------------------------------------------- + +/// A minimal EVM JSON-RPC client using the workspace `reqwest`. +/// +/// This is intentionally narrow — it covers only `eth_blockNumber` and `eth_getLogs`, the two +/// methods the ingest worker needs. A full client is the responsibility of `#218` (`octo-evm-rpc`). +/// +/// **Security:** the RPC URL may contain an API key. It is never logged. +pub struct EvmRpcClient { + client: Client, + /// The EVM JSON-RPC endpoint URL. Never logged (may contain an API key). + rpc_url: String, +} + +impl EvmRpcClient { + /// Create a new client. `rpc_url` is stored but never logged. + pub fn new(rpc_url: impl Into) -> Self { + let client = Client::builder() + .timeout(Duration::from_secs(30)) + // Cap response bodies to 64 MiB so a malicious or malfunctioning RPC cannot OOM the + // worker by returning an unbounded stream. + .build() + .expect("reqwest client build should not fail with valid config"); + Self { + client, + rpc_url: rpc_url.into(), + } + } + + /// Call any JSON-RPC method and return the `result` field as a raw `Value`. + async fn call(&self, method: &str, params: Value) -> Result { + let body = json!({ + "jsonrpc": "2.0", + "id": 1, + "method": method, + "params": params, + }); + + let resp = self + .client + .post(&self.rpc_url) + .json(&body) + .send() + .await? + .error_for_status()?; + + // JSON-RPC errors arrive as HTTP 200 with an `error` member — status-only checking + // would treat every RPC-level failure as success. + let json: Value = resp.json().await?; + if let Some(err) = json.get("error") { + let code = err.get("code").and_then(|c| c.as_i64()).unwrap_or(-1); + let message = err + .get("message") + .and_then(|m| m.as_str()) + .unwrap_or("unknown") + .to_string(); + + // Detect range-too-large by message heuristics (providers phrase this differently). + let lower = message.to_lowercase(); + if lower.contains("range") + || lower.contains("block range") + || lower.contains("log response size") + || lower.contains("too large") + || lower.contains("too many blocks") + || lower.contains("exceed") + || lower.contains("limit") + { + return Err(EvmRpcError::RangeTooLarge); + } + + return Err(EvmRpcError::JsonRpc { code, message }); + } + + json.get("result") + .cloned() + .ok_or_else(|| EvmRpcError::UnexpectedResponse("missing `result` field".into())) + } + + /// `eth_blockNumber` — returns the latest block number. + pub async fn block_number(&self) -> Result { + let result = self.call("eth_blockNumber", json!([])).await?; + let hex = result + .as_str() + .ok_or_else(|| EvmRpcError::UnexpectedResponse("eth_blockNumber not a string".into()))?; + parse_hex_u64(hex).ok_or_else(|| { + EvmRpcError::UnexpectedResponse(format!( + "eth_blockNumber: could not parse hex block number '{hex}'" + )) + }) + } + + /// `eth_getLogs` filtered by the given block range and address list. + /// + /// Only returns `Transfer(address,address,uint256)` events (topic0 filter). Removed logs + /// (from reorgs) are filtered out by the caller via `log.removed`. + /// + /// Returns `Err(EvmRpcError::RangeTooLarge)` when the provider rejects the range, which + /// triggers adaptive bisection in the caller. + pub async fn get_logs( + &self, + from_block: u64, + to_block: u64, + addresses: &[&str], + ) -> Result, EvmRpcError> { + let result = self + .call( + "eth_getLogs", + json!([{ + "fromBlock": format!("0x{from_block:x}"), + "toBlock": format!("0x{to_block:x}"), + "address": addresses, + "topics": [TRANSFER_TOPIC], + }]), + ) + .await?; + + let logs: Vec = serde_json::from_value(result).map_err(|e| { + EvmRpcError::UnexpectedResponse(format!("eth_getLogs deserialization failed: {e}")) + })?; + + Ok(logs) + } +} + +// --------------------------------------------------------------------------- +// Registered token +// --------------------------------------------------------------------------- + +/// A registered ERC-20 token that the ingestor will credit. +/// +/// The registry is the defence against fee-on-transfer and rebasing tokens: only tokens +/// explicitly registered here are credited. The token contract address is the key used +/// to verify that a log was emitted by the correct contract. +#[derive(Debug, Clone)] +pub struct RegisteredToken { + /// EIP-55 checksummed contract address. Comparisons are case-insensitive. + pub contract_address: String, + /// Ticker symbol (e.g. `"USDC"`, `"DAI"`). + pub symbol: String, + /// Decimal places (e.g. 6 for USDC, 18 for DAI). Stored for documentation; the ingestor + /// does not interpret the value — it stores raw base units and lets the crediting layer + /// (#222) handle human-readable conversion. + pub decimals: u8, +} + +// --------------------------------------------------------------------------- +// EvmIngestor +// --------------------------------------------------------------------------- + +/// The EVM deposit ingest worker for one (wallet, chain) pair. +/// +/// Mirrors `Ingestor`'s shape and contract: +/// - `poll_once`: scan one block range, process each log, advance the cursor. +/// - `process_log`: attribute and record one Transfer log (or skip/quarantine it). +/// - Returns `Processed` values that are the same variants as the Stellar path. +/// +/// # Crash safety +/// +/// The cursor advances **only after** a log is durably recorded in the database, matching the +/// guarantee documented in `crates/ingest/src/lib.rs`. A crash mid-batch resumes from the +/// last-committed block without missing or double-processing logs. +pub struct EvmIngestor { + store: Store, + rpc: EvmRpcClient, + wallet_id: Uuid, + /// CAIP-2 chain identifier (e.g. `"eip155:1"`, `"eip155:11155111"`). + chain_id: String, + /// Deposit addresses for this wallet on this chain (lowercase hex with `0x` prefix). + /// Only Transfer events whose `to` topic matches one of these addresses are attributed. + deposit_addresses: HashSet, + /// Registered token contracts. Only logs emitted by one of these contracts are processed. + registered_tokens: Vec, + /// Maximum block range for `eth_getLogs`. Starts at `DEFAULT_BLOCK_RANGE` and may be + /// bisected down if the provider returns `RangeTooLarge`. + max_block_range: u64, +} + +impl EvmIngestor { + /// Create a new `EvmIngestor`. + /// + /// - `rpc_url`: the EVM JSON-RPC endpoint. May contain an API key — never logged. + /// - `deposit_addresses`: hex addresses (case-insensitive) belonging to this wallet on this + /// chain. + /// - `registered_tokens`: the token registry for this chain. Only tokens listed here are + /// credited; all others are quarantined. + pub fn new( + store: Store, + rpc_url: impl Into, + wallet_id: Uuid, + chain_id: impl Into, + deposit_addresses: impl IntoIterator, + registered_tokens: Vec, + ) -> Self { + Self { + store, + rpc: EvmRpcClient::new(rpc_url), + wallet_id, + chain_id: chain_id.into(), + deposit_addresses: deposit_addresses + .into_iter() + .map(|a| a.to_lowercase()) + .collect(), + registered_tokens, + max_block_range: DEFAULT_BLOCK_RANGE, + } + } + + /// Poll once: scan the next block range for Transfer logs, process each, and persist the + /// cursor. Returns the number of logs processed (including quarantined ones). + /// + /// Adaptively bisects the block range if the provider returns `RangeTooLarge`. + pub async fn poll_once(&mut self) -> Result { + let latest = self + .rpc + .block_number() + .await + .map_err(EvmIngestError::Rpc)?; + + let from = self + .store + .get_evm_cursor(self.wallet_id, &self.chain_id) + .await + .map_err(EvmIngestError::Store)? + .map(|b| b + 1) // resume from the block after the last one we processed + .unwrap_or(latest.saturating_sub(self.max_block_range)); + + if from > latest { + // Cursor is already at the tip; nothing to do. + return Ok(0); + } + + let contract_addrs: Vec<&str> = self + .registered_tokens + .iter() + .map(|t| t.contract_address.as_str()) + .collect(); + + // Adaptive bisection: if the provider returns RangeTooLarge, halve the range and retry. + let mut range = (latest - from + 1).min(self.max_block_range); + let mut bisections = 0; + + loop { + let to = (from + range - 1).min(latest); + + let logs = match self.rpc.get_logs(from, to, &contract_addrs).await { + Ok(logs) => logs, + Err(EvmRpcError::RangeTooLarge) => { + bisections += 1; + if bisections > MAX_BISECTIONS { + return Err(EvmIngestError::RangeTooLargeExhausted { from, to }); + } + range = (range / 2).max(MIN_BLOCK_RANGE); + tracing::debug!( + wallet = %self.wallet_id, + chain = %self.chain_id, + from, + to, + new_range = range, + "RangeTooLarge: bisecting block range" + ); + continue; + } + Err(e) => return Err(EvmIngestError::Rpc(e)), + }; + + let count = self.process_range(from, to, logs).await?; + return Ok(count); + } + } + + /// Process all logs in a fetched range, advancing the cursor block-by-block after each + /// confirmed record. + async fn process_range( + &self, + _from: u64, + to: u64, + logs: Vec, + ) -> Result { + let mut count = 0; + + for log in &logs { + // Reorged logs must never be processed — they represent a chain state that has been + // superseded. A log with `removed: true` is a reorg reversal notification. + if log.removed { + tracing::debug!( + wallet = %self.wallet_id, + tx = %log.transaction_hash, + "skipping removed (reorged) log" + ); + continue; + } + + self.process_log(log).await?; + count += 1; + + // Advance the cursor after each log so a crash resumes cleanly. + // We use the log's own block number as the cursor, not `to`, so partial-block + // progress is preserved on crash. + if let Some(block_num) = log.block_number_u64() { + self.store + .set_evm_cursor(self.wallet_id, &self.chain_id, block_num) + .await + .map_err(EvmIngestError::Store)?; + } + } + + // Even if there were no logs (or all were removed), advance the cursor to `to` so we + // don't re-scan the same empty range on the next poll. + if logs.is_empty() || logs.iter().all(|l| l.removed) { + self.store + .set_evm_cursor(self.wallet_id, &self.chain_id, to) + .await + .map_err(EvmIngestError::Store)?; + } + + Ok(count) + } + + /// Process one ERC-20 Transfer log. + /// + /// # Security: contract address verification + /// + /// This is the critical security check. We verify that `log.address` (the *emitting* + /// contract) is a registered token contract **before** reading any topic. An attacker can + /// deploy a contract that emits a Transfer event with any topics — a deposit address in + /// `topics[2]` and a huge value in `data`. If we attributed on topics alone, that would + /// credit the attacker's address. Checking `log.address` first prevents this. + pub async fn process_log(&self, log: &EvmLog) -> Result { + // ---------------------------------------------------------------- + // Step 1: Verify the log was emitted by a registered token contract. + // This is the single most important security check in this module. + // ---------------------------------------------------------------- + let token = self + .registered_tokens + .iter() + .find(|t| t.contract_address.eq_ignore_ascii_case(&log.address)); + + let token = match token { + Some(t) => t, + None => { + // The log was emitted by an unregistered contract. Even if its topics look like a + // Transfer to a deposit address, we must not credit it — doing so would let any + // attacker mint balances by emitting fake Transfer events. + tracing::debug!( + wallet = %self.wallet_id, + chain = %self.chain_id, + contract = %log.address, + "skipping Transfer from unregistered contract" + ); + return Ok(Processed::Skipped); + } + }; + + // ---------------------------------------------------------------- + // Step 2: Decode Transfer topics. topic0 must be the Transfer sig; + // topics[1] = from, topics[2] = to. + // ---------------------------------------------------------------- + if log.topics.first().map(String::as_str) != Some(TRANSFER_TOPIC) { + return Ok(Processed::Skipped); + } + + let to_addr = match log.to_address() { + Some(a) => a, + None => { + tracing::warn!( + tx = %log.transaction_hash, + "Transfer log missing valid `to` topic — skipping" + ); + return Ok(Processed::Skipped); + } + }; + + // ---------------------------------------------------------------- + // Step 3: Check whether `to` is one of our deposit addresses. + // Transfers to non-deposit addresses are irrelevant. + // ---------------------------------------------------------------- + if !self.deposit_addresses.contains(&to_addr) { + return Ok(Processed::Skipped); + } + + // ---------------------------------------------------------------- + // Step 4: Decode the amount. + // ---------------------------------------------------------------- + let amount = match log.amount_i64() { + Some(a) if a > 0 => a, + _ => { + tracing::warn!( + tx = %log.transaction_hash, + data = %log.data, + "Transfer log has invalid or overflow amount — skipping" + ); + return Ok(Processed::Skipped); + } + }; + + let from_addr = log.from_address().unwrap_or_default(); + let tx_hash = &log.transaction_hash; + let log_index = log.log_index_u32().unwrap_or(0); + + // ---------------------------------------------------------------- + // Step 5: Attribute to a customer address. + // ---------------------------------------------------------------- + let address_row = self + .store + .evm_address_by_hex(self.wallet_id, &to_addr) + .await + .map_err(EvmIngestError::Store)?; + + let address_id = address_row.as_ref().map(|a| a.id); + let attributed = address_id.is_some(); + + // If the address is known but the token is registered, we record it. + // (The contract check above already ensured the token is registered.) + + // ---------------------------------------------------------------- + // Step 6: Record idempotently. The (chain_id, tx_hash, log_index) unique index + // in the DB ensures replays are no-ops. + // ---------------------------------------------------------------- + let dep = NewEvmDeposit { + wallet_id: self.wallet_id, + address_id, + chain_id: &self.chain_id, + token_symbol: &token.symbol, + token_contract: &token.contract_address, + amount_base_units: amount, + from_address: &from_addr, + to_address: &to_addr, + tx_hash, + log_index, + }; + + match self + .store + .record_evm_deposit(&dep) + .await + .map_err(EvmIngestError::Store)? + { + Some(_tx) => { + tracing::debug!( + wallet = %self.wallet_id, + chain = %self.chain_id, + token = %token.symbol, + to = %to_addr, + amount, + attributed, + "EVM deposit recorded (unconfirmed)" + ); + Ok(Processed::Recorded { attributed }) + } + None => Ok(Processed::Duplicate), + } + } + + /// Run forever, polling every `interval`. Errors are logged and retried. + pub async fn run(mut self, interval: Duration) { + loop { + match self.poll_once().await { + Ok(n) if n > 0 => tracing::debug!( + wallet = %self.wallet_id, + chain = %self.chain_id, + processed = n, + "EVM ingest poll" + ), + Ok(_) => {} + Err(e) => tracing::warn!( + wallet = %self.wallet_id, + chain = %self.chain_id, + error = ?e, + "EVM ingest poll failed; will retry" + ), + } + let _ = self + .store + .mark_evm_polled(self.wallet_id, &self.chain_id) + .await; + tokio::time::sleep(interval).await; + } + } +} + +// --------------------------------------------------------------------------- +// Errors +// --------------------------------------------------------------------------- + +/// Errors from the EVM ingest worker. +#[derive(Debug, thiserror::Error)] +pub enum EvmIngestError { + #[error("store error: {0}")] + Store(#[from] octo_store::StoreError), + + #[error("EVM RPC error: {0}")] + Rpc(#[from] EvmRpcError), + + /// The block range was bisected down to `MIN_BLOCK_RANGE` and the provider still returned + /// `RangeTooLarge`. This indicates a misconfigured provider or a very constrained endpoint. + #[error("RangeTooLarge persisted after {MAX_BISECTIONS} bisections (from={from}, to={to})")] + RangeTooLargeExhausted { from: u64, to: u64 }, +} + +// --------------------------------------------------------------------------- +// Unit tests (pure logic, no DB) +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + + fn make_log( + address: &str, + topic0: &str, + topic1_from: &str, + topic2_to: &str, + data: &str, + block: &str, + log_index: &str, + ) -> EvmLog { + EvmLog { + address: address.to_string(), + transaction_hash: "0xabc".to_string(), + topics: vec![ + topic0.to_string(), + topic1_from.to_string(), + topic2_to.to_string(), + ], + data: data.to_string(), + block_number: block.to_string(), + log_index: log_index.to_string(), + removed: false, + } + } + + const TOKEN: &str = "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"; // USDC mainnet + const DEPOSIT_ADDR: &str = "0xdeadbeef00000000000000000000000000000001"; + const RANDOM_ADDR: &str = "0x1234567890000000000000000000000000000002"; + + /// Pad a 20-byte address to 32-byte topic format. + fn as_topic(addr: &str) -> String { + let hex = addr.strip_prefix("0x").unwrap_or(addr); + format!("0x{:0>64}", hex) + } + + /// Encode a u64 as a 32-byte ABI uint256 data field. + fn as_data(value: u64) -> String { + format!("0x{value:0>64x}") + } + + #[test] + fn to_address_extracts_correctly() { + let log = make_log( + TOKEN, + TRANSFER_TOPIC, + &as_topic(RANDOM_ADDR), + &as_topic(DEPOSIT_ADDR), + &as_data(1_000_000), + "0x1", + "0x0", + ); + assert_eq!( + log.to_address(), + Some(format!("0x{}", DEPOSIT_ADDR.strip_prefix("0x").unwrap().to_lowercase())) + ); + } + + #[test] + fn from_address_extracts_correctly() { + let log = make_log( + TOKEN, + TRANSFER_TOPIC, + &as_topic(RANDOM_ADDR), + &as_topic(DEPOSIT_ADDR), + &as_data(1_000_000), + "0x1", + "0x0", + ); + assert_eq!( + log.from_address(), + Some(format!("0x{}", RANDOM_ADDR.strip_prefix("0x").unwrap().to_lowercase())) + ); + } + + #[test] + fn amount_i64_parses_usdc_6_decimals() { + // 1.000000 USDC = 1_000_000 base units + let log = make_log( + TOKEN, + TRANSFER_TOPIC, + &as_topic(RANDOM_ADDR), + &as_topic(DEPOSIT_ADDR), + &as_data(1_000_000), + "0x1", + "0x0", + ); + assert_eq!(log.amount_i64(), Some(1_000_000)); + } + + #[test] + fn amount_i64_parses_dai_18_decimals() { + // 1.0 DAI = 1_000_000_000_000_000_000 base units (1e18) + // This fits in i64 (max ~9.2e18) + let amount: u64 = 1_000_000_000_000_000_000; + let log = make_log( + TOKEN, + TRANSFER_TOPIC, + &as_topic(RANDOM_ADDR), + &as_topic(DEPOSIT_ADDR), + &as_data(amount), + "0x1", + "0x0", + ); + assert_eq!(log.amount_i64(), Some(amount as i64)); + } + + #[test] + fn amount_i64_rejects_overflow() { + // 10 ETH = 10_000_000_000_000_000_000 > i64::MAX (~9.22e18) — overflow + let too_big: &str = "0x0000000000000000000000000000000000000000000000008ac7230489e80000"; + let log = EvmLog { + address: TOKEN.to_string(), + transaction_hash: "0xabc".to_string(), + topics: vec![ + TRANSFER_TOPIC.to_string(), + as_topic(RANDOM_ADDR), + as_topic(DEPOSIT_ADDR), + ], + data: too_big.to_string(), + block_number: "0x1".to_string(), + log_index: "0x0".to_string(), + removed: false, + }; + assert_eq!(log.amount_i64(), None); + } + + #[test] + fn block_number_u64_parses_hex() { + let log = make_log( + TOKEN, + TRANSFER_TOPIC, + &as_topic(RANDOM_ADDR), + &as_topic(DEPOSIT_ADDR), + &as_data(1), + "0x1a2b3c", + "0x5", + ); + assert_eq!(log.block_number_u64(), Some(0x1a2b3c)); + assert_eq!(log.log_index_u32(), Some(5)); + } + + #[test] + fn removed_log_is_flagged() { + let mut log = make_log( + TOKEN, + TRANSFER_TOPIC, + &as_topic(RANDOM_ADDR), + &as_topic(DEPOSIT_ADDR), + &as_data(1), + "0x1", + "0x0", + ); + log.removed = true; + assert!(log.removed); + } + + #[test] + fn parse_hex_u64_handles_edge_cases() { + assert_eq!(parse_hex_u64("0x0"), Some(0)); + assert_eq!(parse_hex_u64("0x1"), Some(1)); + assert_eq!(parse_hex_u64("0xffffffffffffffff"), Some(u64::MAX)); + assert_eq!(parse_hex_u64(""), None); + assert_eq!(parse_hex_u64("0xgg"), None); + } +} diff --git a/crates/ingest/src/lib.rs b/crates/ingest/src/lib.rs index 1ed809e..586824d 100644 --- a/crates/ingest/src/lib.rs +++ b/crates/ingest/src/lib.rs @@ -15,6 +15,7 @@ pub mod amount; pub mod horizon; +pub mod evm; #[cfg(test)] mod backfill_tests; diff --git a/crates/ingest/tests/evm_ingest_tests.rs b/crates/ingest/tests/evm_ingest_tests.rs new file mode 100644 index 0000000..0425d9b --- /dev/null +++ b/crates/ingest/tests/evm_ingest_tests.rs @@ -0,0 +1,766 @@ +//! EVM deposit detection integration tests. +//! +//! Tests the `EvmIngestor` against a real Postgres database (skips cleanly when `DATABASE_URL` +//! is absent). Covers: +//! +//! 1. **Happy path**: a Transfer to a registered deposit address is recorded as `unconfirmed`. +//! 2. **Adversarial / fake Transfer**: a hostile contract emits a Transfer with a deposit address +//! in `topics[2]` and a huge value — must be rejected because `log.address` is not registered. +//! 3. **Idempotency / replay**: processing the same log twice records nothing new. +//! 4. **Crash-resume**: cursor advances block-by-block so a restart re-processes only unfinished +//! work. +//! 5. **Unregistered token quarantine**: a Transfer from an unregistered token to a known address +//! is skipped (not credited). +//! 6. **Range bisection**: a mock RPC returning `RangeTooLarge` causes bisection, not a stall. +//! 7. **Removed log (reorg)**: a log with `removed: true` is never processed. +//! 8. **Amount at 6 and 18 decimals**: both USDC and DAI-like values are stored correctly. + +use axum::Router; +use axum::routing::post; +use octo_ingest::evm::{EvmIngestor, EvmLog, RegisteredToken}; +use octo_ingest::Processed; +use octo_store::{NewWallet, Store}; +use serde_json::json; +use std::sync::{Arc, Mutex, Once}; +use uuid::Uuid; + +static LOAD_ENV: Once = Once::new(); + +fn database_url() -> Option { + LOAD_ENV.call_once(|| { + let _ = dotenvy::dotenv(); + }); + std::env::var("DATABASE_URL").ok() +} + +// --------------------------------------------------------------------------- +// Constants +// --------------------------------------------------------------------------- + +/// USDC contract address (Ethereum mainnet — used as a stand-in for tests). +const USDC_CONTRACT: &str = "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"; +/// DAI contract address (Ethereum mainnet). +const DAI_CONTRACT: &str = "0x6B175474E89094C44Da98b954EedeAC495271d0F"; +/// A deposit address belonging to our test wallet. +const DEPOSIT_ADDR: &str = "0xdeadbeef00000000000000000000000000000001"; +/// An unrelated external address (the sender in most tests). +const SENDER_ADDR: &str = "0xcafe000000000000000000000000000000000001"; +/// A hostile contract that is NOT registered. +const HOSTILE_CONTRACT: &str = "0xbad0000000000000000000000000000000000001"; + +const TRANSFER_TOPIC: &str = + "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"; + +const CHAIN_ID: &str = "eip155:1"; + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +/// Pad a 20-byte hex address to a 32-byte ABI-encoded topic. +fn as_topic(addr: &str) -> String { + let hex = addr.strip_prefix("0x").unwrap_or(addr); + format!("0x{:0>64}", hex.to_lowercase()) +} + +/// Encode a u64 as a 32-byte ABI uint256 data field. +fn as_data(value: u64) -> String { + format!("0x{value:0>64x}") +} + +/// Build a minimal Transfer `EvmLog`. +fn transfer_log( + contract: &str, + from: &str, + to: &str, + amount: u64, + block: u64, + log_index: u32, + tx_hash: &str, +) -> EvmLog { + EvmLog { + address: contract.to_string(), + transaction_hash: tx_hash.to_string(), + topics: vec![ + TRANSFER_TOPIC.to_string(), + as_topic(from), + as_topic(to), + ], + data: as_data(amount), + block_number: format!("0x{block:x}"), + log_index: format!("0x{log_index:x}"), + removed: false, + } +} + +/// Create an isolated wallet and `EvmIngestor` for one test. +async fn setup( + store: &Store, + rpc_url: &str, +) -> (Uuid, EvmIngestor) { + let run_id = Uuid::new_v4().simple().to_string(); + let wallet = store + .create_wallet(NewWallet { + network: "testnet", + stellar_account_g: &format!("G_EVM_TEST_{run_id}"), + sealed_ciphertext: b"ct", + sealed_nonce: b"nonce", + sealed_salt: b"salt", + sealed_scheme: 1, + label: None, + user_id: None, + description: None, + }) + .await + .expect("create wallet"); + + // Register the deposit address in the store as a muxed address row. + // We repurpose muxed_address as deposit_address for the EVM path until the schema migration. + let addr = store + .allocate_address( + wallet.id, + |_id| Ok(DEPOSIT_ADDR.to_string()), + Some("evm-test-customer"), + json!({}), + ) + .await + .expect("allocate EVM deposit address"); + + let tokens = vec![ + RegisteredToken { + contract_address: USDC_CONTRACT.to_string(), + symbol: "USDC".to_string(), + decimals: 6, + }, + RegisteredToken { + contract_address: DAI_CONTRACT.to_string(), + symbol: "DAI".to_string(), + decimals: 18, + }, + ]; + + let ingestor = EvmIngestor::new( + store.clone(), + rpc_url, + wallet.id, + CHAIN_ID, + vec![DEPOSIT_ADDR.to_lowercase()], + tokens, + ); + + let _ = addr; // ensure the address row exists + (wallet.id, ingestor) +} + +// --------------------------------------------------------------------------- +// Mock JSON-RPC server helpers +// --------------------------------------------------------------------------- + +/// Shared state for the mock RPC server. +#[derive(Clone)] +struct MockRpcState { + /// Canned response to return for eth_blockNumber. + block_number: Arc>, + /// Canned logs to return for eth_getLogs. + logs: Arc>>, + /// If set, return a RangeTooLarge error for the first N calls to eth_getLogs. + range_too_large_count: Arc>, + /// How many eth_getLogs calls were made. + get_logs_call_count: Arc>, +} + +/// Axum handler for the mock RPC endpoint. +async fn mock_rpc_handler( + axum::extract::State(state): axum::extract::State, + axum::Json(body): axum::Json, +) -> axum::Json { + let method = body.get("method").and_then(|m| m.as_str()).unwrap_or(""); + match method { + "eth_blockNumber" => { + let n = *state.block_number.lock().unwrap(); + axum::Json(json!({ "jsonrpc": "2.0", "id": 1, "result": format!("0x{n:x}") })) + } + "eth_getLogs" => { + let mut count = state.get_logs_call_count.lock().unwrap(); + *count += 1; + let mut remaining = state.range_too_large_count.lock().unwrap(); + if *remaining > 0 { + *remaining -= 1; + drop(remaining); + drop(count); + return axum::Json(json!({ + "jsonrpc": "2.0", + "id": 1, + "error": { "code": -32005, "message": "query returned more than 10000 results. Try with this block range [0x0, 0x4e20]." } + })); + } + let logs = state.logs.lock().unwrap().clone(); + axum::Json(json!({ "jsonrpc": "2.0", "id": 1, "result": logs })) + } + other => axum::Json(json!({ + "jsonrpc": "2.0", + "id": 1, + "error": { "code": -32601, "message": format!("method not found: {other}") } + })), + } +} + +async fn start_mock_rpc(state: MockRpcState) -> String { + let app = Router::new() + .route("/", post(mock_rpc_handler)) + .with_state(state); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind mock RPC"); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(listener, app).await.expect("mock RPC serve"); + }); + format!("http://{addr}/") +} + +fn rpc_log(log: &EvmLog) -> serde_json::Value { + json!({ + "address": log.address, + "transactionHash": log.transaction_hash, + "topics": log.topics, + "data": log.data, + "blockNumber": log.block_number, + "logIndex": log.log_index, + "removed": log.removed, + }) +} + +// --------------------------------------------------------------------------- +// Test 1: Happy path — Transfer to registered deposit address is recorded +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn erc20_transfer_to_deposit_address_is_recorded() { + let Some(db_url) = database_url() else { + eprintln!("SKIPPED: set DATABASE_URL to run EVM ingest tests"); + return; + }; + let store = Store::connect(&db_url).await.expect("connect"); + store.migrate().await.expect("migrate"); + + let log = transfer_log( + USDC_CONTRACT, + SENDER_ADDR, + DEPOSIT_ADDR, + 1_000_000, // 1 USDC (6 decimals) + 100, + 0, + "0xaaa0000000000000000000000000000000000000000000000000000000000001", + ); + + let mock_state = MockRpcState { + block_number: Arc::new(Mutex::new(100)), + logs: Arc::new(Mutex::new(vec![rpc_log(&log)])), + range_too_large_count: Arc::new(Mutex::new(0)), + get_logs_call_count: Arc::new(Mutex::new(0)), + }; + let rpc_url = start_mock_rpc(mock_state).await; + + let (wallet_id, ingestor) = setup(&store, &rpc_url).await; + + let result = ingestor.process_log(&log).await.expect("process_log"); + assert_eq!( + result, + Processed::Recorded { attributed: true }, + "Transfer to a registered deposit address must be recorded and attributed" + ); + + let txs = store + .list_transactions(wallet_id, 10, None) + .await + .expect("list_transactions"); + assert_eq!(txs.len(), 1, "one transaction row expected"); + let tx = &txs[0]; + assert_eq!(tx.amount_stroops, 1_000_000); + assert_eq!(tx.asset_code, "USDC"); + assert_eq!(tx.status, "unconfirmed", "EVM deposits must be unconfirmed until #222"); + assert_eq!(tx.asset_issuer.as_deref(), Some(USDC_CONTRACT)); +} + +// --------------------------------------------------------------------------- +// Test 2: Amount at 18 decimals (DAI-like) +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn dai_transfer_18_decimals_stored_correctly() { + let Some(db_url) = database_url() else { + eprintln!("SKIPPED: set DATABASE_URL to run EVM ingest tests"); + return; + }; + let store = Store::connect(&db_url).await.expect("connect"); + store.migrate().await.expect("migrate"); + + // 1.0 DAI = 1_000_000_000_000_000_000 attoDAI + let amount: u64 = 1_000_000_000_000_000_000; + let log = transfer_log( + DAI_CONTRACT, + SENDER_ADDR, + DEPOSIT_ADDR, + amount, + 200, + 1, + "0xbbb0000000000000000000000000000000000000000000000000000000000001", + ); + + let mock_state = MockRpcState { + block_number: Arc::new(Mutex::new(200)), + logs: Arc::new(Mutex::new(vec![rpc_log(&log)])), + range_too_large_count: Arc::new(Mutex::new(0)), + get_logs_call_count: Arc::new(Mutex::new(0)), + }; + let rpc_url = start_mock_rpc(mock_state).await; + + let (wallet_id, ingestor) = setup(&store, &rpc_url).await; + + let result = ingestor.process_log(&log).await.expect("process_log (DAI)"); + assert_eq!( + result, + Processed::Recorded { attributed: true }, + "DAI Transfer must be recorded" + ); + + let txs = store + .list_transactions(wallet_id, 10, None) + .await + .expect("list_transactions"); + assert_eq!(txs.len(), 1); + assert_eq!(txs[0].amount_stroops, amount as i64); + assert_eq!(txs[0].asset_code, "DAI"); + assert_eq!(txs[0].status, "unconfirmed"); +} + +// --------------------------------------------------------------------------- +// Test 3: Adversarial — fake Transfer from hostile contract must be rejected +// +// This is the critical security test. A hostile contract emits a Transfer event +// with our deposit address in topics[2] and a large value in data. Because +// log.address is not a registered token, it must be Skipped. +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn fake_transfer_from_unregistered_contract_is_skipped() { + let Some(db_url) = database_url() else { + eprintln!("SKIPPED: set DATABASE_URL to run EVM ingest tests"); + return; + }; + let store = Store::connect(&db_url).await.expect("connect"); + store.migrate().await.expect("migrate"); + + // Hostile contract emits Transfer with deposit address as `to` and 1 billion USDC as value. + let hostile_log = transfer_log( + HOSTILE_CONTRACT, // ← NOT a registered contract + SENDER_ADDR, + DEPOSIT_ADDR, + 1_000_000_000_000_000, // 1 billion USDC (6 decimals) — attacker's desired credit + 300, + 0, + "0xccc0000000000000000000000000000000000000000000000000000000000001", + ); + + let mock_state = MockRpcState { + block_number: Arc::new(Mutex::new(300)), + logs: Arc::new(Mutex::new(vec![rpc_log(&hostile_log)])), + range_too_large_count: Arc::new(Mutex::new(0)), + get_logs_call_count: Arc::new(Mutex::new(0)), + }; + let rpc_url = start_mock_rpc(mock_state).await; + + let (wallet_id, ingestor) = setup(&store, &rpc_url).await; + + // The process_log call must skip the hostile log. + let result = ingestor + .process_log(&hostile_log) + .await + .expect("process_log should not error for a skipped log"); + + assert_eq!( + result, + Processed::Skipped, + "A Transfer emitted by an unregistered contract must be Skipped — topics alone must never cause attribution" + ); + + // No transactions must have been recorded. + let txs = store + .list_transactions(wallet_id, 10, None) + .await + .expect("list_transactions"); + assert_eq!( + txs.len(), + 0, + "hostile fake Transfer must not create any transaction row" + ); +} + +// --------------------------------------------------------------------------- +// Test 4: Idempotency / replay — processing the same log twice records nothing new +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn replay_same_log_is_idempotent() { + let Some(db_url) = database_url() else { + eprintln!("SKIPPED: set DATABASE_URL to run EVM ingest tests"); + return; + }; + let store = Store::connect(&db_url).await.expect("connect"); + store.migrate().await.expect("migrate"); + + let log = transfer_log( + USDC_CONTRACT, + SENDER_ADDR, + DEPOSIT_ADDR, + 500_000, // 0.5 USDC + 400, + 0, + "0xddd0000000000000000000000000000000000000000000000000000000000001", + ); + + let mock_state = MockRpcState { + block_number: Arc::new(Mutex::new(400)), + logs: Arc::new(Mutex::new(vec![rpc_log(&log)])), + range_too_large_count: Arc::new(Mutex::new(0)), + get_logs_call_count: Arc::new(Mutex::new(0)), + }; + let rpc_url = start_mock_rpc(mock_state).await; + + let (wallet_id, ingestor) = setup(&store, &rpc_url).await; + + // First delivery: must be recorded. + let r1 = ingestor.process_log(&log).await.expect("first process"); + assert_eq!(r1, Processed::Recorded { attributed: true }); + + // Second delivery (replay): must be a no-op duplicate. + let r2 = ingestor.process_log(&log).await.expect("second process"); + assert_eq!(r2, Processed::Duplicate, "replay must be a Duplicate"); + + // Third delivery (another replay): still no-op. + let r3 = ingestor.process_log(&log).await.expect("third process"); + assert_eq!(r3, Processed::Duplicate, "second replay must also be Duplicate"); + + let txs = store + .list_transactions(wallet_id, 10, None) + .await + .expect("list_transactions"); + assert_eq!(txs.len(), 1, "only one transaction row despite three deliveries"); + assert_eq!(txs[0].amount_stroops, 500_000); +} + +// --------------------------------------------------------------------------- +// Test 5: Crash-resume — cursor advances block-by-block +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn cursor_advances_per_block_and_resume_is_exactly_once() { + let Some(db_url) = database_url() else { + eprintln!("SKIPPED: set DATABASE_URL to run EVM ingest tests"); + return; + }; + let store = Store::connect(&db_url).await.expect("connect"); + store.migrate().await.expect("migrate"); + + // Build two logs at different block numbers to simulate a two-block range. + let log_b500 = transfer_log( + USDC_CONTRACT, SENDER_ADDR, DEPOSIT_ADDR, 100_000, 500, 0, + "0xeee0000000000000000000000000000000000000000000000000000000000001", + ); + let log_b501 = transfer_log( + USDC_CONTRACT, SENDER_ADDR, DEPOSIT_ADDR, 200_000, 501, 0, + "0xeee0000000000000000000000000000000000000000000000000000000000002", + ); + + let mock_state = MockRpcState { + block_number: Arc::new(Mutex::new(501)), + logs: Arc::new(Mutex::new(vec![rpc_log(&log_b500), rpc_log(&log_b501)])), + range_too_large_count: Arc::new(Mutex::new(0)), + get_logs_call_count: Arc::new(Mutex::new(0)), + }; + let rpc_url = start_mock_rpc(mock_state).await; + + let (wallet_id, ingestor) = setup(&store, &rpc_url).await; + + // Process both logs individually (simulating poll_once's inner loop). + let r1 = ingestor.process_log(&log_b500).await.expect("block 500"); + assert_eq!(r1, Processed::Recorded { attributed: true }); + store + .set_evm_cursor(wallet_id, CHAIN_ID, 500) + .await + .expect("set cursor after block 500"); + + let r2 = ingestor.process_log(&log_b501).await.expect("block 501"); + assert_eq!(r2, Processed::Recorded { attributed: true }); + store + .set_evm_cursor(wallet_id, CHAIN_ID, 501) + .await + .expect("set cursor after block 501"); + + // Cursor should now be at 501. + let cursor = store + .get_evm_cursor(wallet_id, CHAIN_ID) + .await + .expect("get cursor"); + assert_eq!(cursor, Some(501), "cursor should be at block 501"); + + // Simulate a crash-resume: replay both logs. Both must deduplicate. + let r3 = ingestor.process_log(&log_b500).await.expect("replay b500"); + assert_eq!(r3, Processed::Duplicate, "replay of block 500 log must be Duplicate"); + + let r4 = ingestor.process_log(&log_b501).await.expect("replay b501"); + assert_eq!(r4, Processed::Duplicate, "replay of block 501 log must be Duplicate"); + + let txs = store + .list_transactions(wallet_id, 10, None) + .await + .expect("list_transactions"); + assert_eq!(txs.len(), 2, "exactly two deposits recorded after crash-resume"); +} + +// --------------------------------------------------------------------------- +// Test 6: Unregistered token quarantine — Skipped (not credited) +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn transfer_of_unregistered_token_to_known_address_is_skipped() { + let Some(db_url) = database_url() else { + eprintln!("SKIPPED: set DATABASE_URL to run EVM ingest tests"); + return; + }; + let store = Store::connect(&db_url).await.expect("connect"); + store.migrate().await.expect("migrate"); + + // Some random token that is NOT in our registry. + let random_token = "0x1f9840a85d5aF5bf1D1762F925BDADdC4201F984"; // UNI, not registered + + let log = transfer_log( + random_token, // not registered + SENDER_ADDR, + DEPOSIT_ADDR, // but to a known deposit address + 9_999_999, + 600, + 0, + "0xfff0000000000000000000000000000000000000000000000000000000000001", + ); + + let mock_state = MockRpcState { + block_number: Arc::new(Mutex::new(600)), + logs: Arc::new(Mutex::new(vec![rpc_log(&log)])), + range_too_large_count: Arc::new(Mutex::new(0)), + get_logs_call_count: Arc::new(Mutex::new(0)), + }; + let rpc_url = start_mock_rpc(mock_state).await; + + let (wallet_id, ingestor) = setup(&store, &rpc_url).await; + + let result = ingestor + .process_log(&log) + .await + .expect("process_log should not error"); + + assert_eq!( + result, + Processed::Skipped, + "A Transfer of an unregistered token must be Skipped, not credited" + ); + + let txs = store + .list_transactions(wallet_id, 10, None) + .await + .expect("list_transactions"); + assert_eq!( + txs.len(), 0, + "no transaction row for unregistered token transfer" + ); +} + +// --------------------------------------------------------------------------- +// Test 7: Range bisection — RangeTooLarge causes bisection, not a stall +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn range_too_large_error_causes_bisection_not_stall() { + let Some(db_url) = database_url() else { + eprintln!("SKIPPED: set DATABASE_URL to run EVM ingest tests"); + return; + }; + let store = Store::connect(&db_url).await.expect("connect"); + store.migrate().await.expect("migrate"); + + let log = transfer_log( + USDC_CONTRACT, + SENDER_ADDR, + DEPOSIT_ADDR, + 250_000, + 700, + 0, + "0x0010000000000000000000000000000000000000000000000000000000000001", + ); + + // First eth_getLogs call returns RangeTooLarge; second returns the log. + let mock_state = MockRpcState { + block_number: Arc::new(Mutex::new(700)), + logs: Arc::new(Mutex::new(vec![rpc_log(&log)])), + range_too_large_count: Arc::new(Mutex::new(1)), // first call fails + get_logs_call_count: Arc::new(Mutex::new(0)), + }; + let get_logs_call_count = Arc::clone(&mock_state.get_logs_call_count); + let rpc_url = start_mock_rpc(mock_state).await; + + let (_wallet_id, mut ingestor) = setup(&store, &rpc_url).await; + + // poll_once must succeed after bisecting the range. + let n = ingestor.poll_once().await.expect("poll_once after bisection"); + + // At least one log should have been processed, and eth_getLogs was called more than once + // (the bisection triggers a retry). + let calls = *get_logs_call_count.lock().unwrap(); + assert!( + calls >= 2, + "eth_getLogs must be retried after RangeTooLarge (got {calls} calls)" + ); + // n >= 1 because the bisected (smaller) range succeeded and returned the log. + assert!(n >= 1, "at least one log must have been processed after bisection"); +} + +// --------------------------------------------------------------------------- +// Test 8: Removed log (reorg) is never processed +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn removed_log_is_never_processed() { + let Some(db_url) = database_url() else { + eprintln!("SKIPPED: set DATABASE_URL to run EVM ingest tests"); + return; + }; + let store = Store::connect(&db_url).await.expect("connect"); + store.migrate().await.expect("migrate"); + + let mut log = transfer_log( + USDC_CONTRACT, + SENDER_ADDR, + DEPOSIT_ADDR, + 100_000, + 800, + 0, + "0x0020000000000000000000000000000000000000000000000000000000000001", + ); + log.removed = true; // simulate reorg + + let mock_state = MockRpcState { + block_number: Arc::new(Mutex::new(800)), + logs: Arc::new(Mutex::new(vec![rpc_log(&log)])), + range_too_large_count: Arc::new(Mutex::new(0)), + get_logs_call_count: Arc::new(Mutex::new(0)), + }; + let rpc_url = start_mock_rpc(mock_state).await; + + let (wallet_id, mut ingestor) = setup(&store, &rpc_url).await; + + // poll_once processes the range but the removed flag is checked there. + let n = ingestor.poll_once().await.expect("poll_once with removed log"); + // count in process_range skips removed logs, so n should be 0. + assert_eq!(n, 0, "removed (reorged) log must not be counted as processed"); + + let txs = store + .list_transactions(wallet_id, 10, None) + .await + .expect("list_transactions"); + assert_eq!(txs.len(), 0, "removed log must produce no transaction row"); +} + +// --------------------------------------------------------------------------- +// Test 9: Transfer to a non-deposit address is Skipped +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn transfer_to_non_deposit_address_is_skipped() { + let Some(db_url) = database_url() else { + eprintln!("SKIPPED: set DATABASE_URL to run EVM ingest tests"); + return; + }; + let store = Store::connect(&db_url).await.expect("connect"); + store.migrate().await.expect("migrate"); + + // Transfer to an address that belongs to a different wallet / is not in our registry. + let other_addr = "0x9999999999999999999999999999999999999999"; + let log = transfer_log( + USDC_CONTRACT, + SENDER_ADDR, + other_addr, // NOT our deposit address + 100_000, + 900, + 0, + "0x0030000000000000000000000000000000000000000000000000000000000001", + ); + + let mock_state = MockRpcState { + block_number: Arc::new(Mutex::new(900)), + logs: Arc::new(Mutex::new(vec![])), + range_too_large_count: Arc::new(Mutex::new(0)), + get_logs_call_count: Arc::new(Mutex::new(0)), + }; + let rpc_url = start_mock_rpc(mock_state).await; + + let (wallet_id, ingestor) = setup(&store, &rpc_url).await; + + let result = ingestor.process_log(&log).await.expect("process_log"); + assert_eq!(result, Processed::Skipped, "Transfer to unknown address must be Skipped"); + + let txs = store + .list_transactions(wallet_id, 10, None) + .await + .expect("list_transactions"); + assert_eq!(txs.len(), 0, "no transaction row for Transfer to non-deposit address"); +} + +// --------------------------------------------------------------------------- +// Test 10: EVM deposits are always recorded as 'unconfirmed' (not 'confirmed') +// — crediting gated on #222 must not be bypassed +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn evm_deposit_status_is_always_unconfirmed() { + let Some(db_url) = database_url() else { + eprintln!("SKIPPED: set DATABASE_URL to run EVM ingest tests"); + return; + }; + let store = Store::connect(&db_url).await.expect("connect"); + store.migrate().await.expect("migrate"); + + let log = transfer_log( + USDC_CONTRACT, + SENDER_ADDR, + DEPOSIT_ADDR, + 100_000, + 1000, + 0, + "0x0040000000000000000000000000000000000000000000000000000000000001", + ); + + let mock_state = MockRpcState { + block_number: Arc::new(Mutex::new(1000)), + logs: Arc::new(Mutex::new(vec![])), + range_too_large_count: Arc::new(Mutex::new(0)), + get_logs_call_count: Arc::new(Mutex::new(0)), + }; + let rpc_url = start_mock_rpc(mock_state).await; + + let (wallet_id, ingestor) = setup(&store, &rpc_url).await; + + ingestor.process_log(&log).await.expect("process_log"); + + let txs = store + .list_transactions(wallet_id, 10, None) + .await + .expect("list_transactions"); + assert_eq!(txs.len(), 1); + assert_eq!( + txs[0].status, + "unconfirmed", + "EVM deposits must be 'unconfirmed' until #222 promotion — \ + this ensures merging #221 before #222 cannot create spendable balances" + ); +} diff --git a/crates/store/migrations/0021_evm_ingest.sql b/crates/store/migrations/0021_evm_ingest.sql new file mode 100644 index 0000000..5d4f802 --- /dev/null +++ b/crates/store/migrations/0021_evm_ingest.sql @@ -0,0 +1,73 @@ +-- EVM ingest support: chain-scoped cursor and dedup for EVM transactions. +-- +-- The Stellar ingest worker uses a Horizon paging token as its cursor and deduplicates +-- on (stellar_tx_hash, operation_index). EVM ingestion uses a block number as its cursor +-- and deduplicates on (chain_id, tx_hash, log_index) — three separate concepts from the +-- Stellar path that need new columns or indexes. +-- +-- Design decisions: +-- +-- 1. Cursor: `ingest_cursor` gains a nullable `chain_id` (TEXT, a CAIP-2 slug such as +-- `eip155:1` or `eip155:11155111`) and a nullable `block_number` (BIGINT). The existing +-- primary key is `wallet_id` alone, which works for Stellar (one cursor per wallet). For EVM +-- a wallet may span multiple chains, so the cursor must be keyed on (wallet_id, chain_id). +-- We add a new unique index over (wallet_id, chain_id) and keep the existing PK for the +-- Stellar path. +-- +-- 2. Dedup: `transactions` gains `chain_id` (TEXT nullable — Stellar rows keep it NULL) and +-- `evm_log_index` (INTEGER nullable — the index of the log entry within its transaction, +-- EVM-specific). A partial unique index over (chain_id, stellar_tx_hash, evm_log_index) +-- where chain_id IS NOT NULL replaces the role of `uq_tx_onchain` for EVM rows. +-- The pre-existing `uq_tx_onchain` and `uq_tx_horizon_op_id` indexes are untouched. +-- +-- 3. Status: EVM deposits are recorded as 'unconfirmed' (gated on confirmation depth, #222). +-- The status column already allows 'pending'; we add 'unconfirmed' to the check constraint. +-- +-- Backward compatibility: all new columns are nullable; existing Stellar rows are unaffected. +-- The new status value 'unconfirmed' is append-only to the check constraint. + +-- --------------------------------------------------------------------------- +-- ingest_cursor: add chain_id + block_number for EVM cursors +-- --------------------------------------------------------------------------- +ALTER TABLE ingest_cursor + ADD COLUMN IF NOT EXISTS chain_id TEXT, + ADD COLUMN IF NOT EXISTS block_number BIGINT; + +-- Per-(wallet, chain) cursor for EVM. The existing PK covers the Stellar path where +-- chain_id IS NULL; this index covers the EVM path. +CREATE UNIQUE INDEX IF NOT EXISTS uq_ingest_cursor_wallet_chain + ON ingest_cursor (wallet_id, chain_id) + WHERE chain_id IS NOT NULL; + +-- --------------------------------------------------------------------------- +-- transactions: add chain_id and evm_log_index +-- --------------------------------------------------------------------------- +ALTER TABLE transactions + ADD COLUMN IF NOT EXISTS chain_id TEXT, + ADD COLUMN IF NOT EXISTS evm_log_index INTEGER; + +-- Anti-double-credit guard for EVM: (chain_id, tx_hash, log_index) is globally unique +-- within one chain. tx_hash reuses the existing stellar_tx_hash column (the column stores +-- whichever chain's tx hash is applicable). Created CONCURRENTLY to avoid an exclusive lock. +CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS uq_tx_evm_onchain + ON transactions (chain_id, stellar_tx_hash, evm_log_index) + WHERE chain_id IS NOT NULL + AND stellar_tx_hash IS NOT NULL + AND evm_log_index IS NOT NULL; + +CREATE INDEX IF NOT EXISTS idx_tx_chain ON transactions(chain_id) + WHERE chain_id IS NOT NULL; + +-- --------------------------------------------------------------------------- +-- transactions: allow 'unconfirmed' status for EVM deposits pending confirmation +-- --------------------------------------------------------------------------- +-- Drop and recreate the check constraint to include the new status value. +-- This is safe: the constraint is checked on insert/update, and no rows carry +-- 'unconfirmed' yet (the column is new). We do it in a single statement so the +-- window where neither constraint exists is zero. +ALTER TABLE transactions + DROP CONSTRAINT IF EXISTS transactions_status_check; + +ALTER TABLE transactions + ADD CONSTRAINT transactions_status_check + CHECK (status IN ('pending', 'confirmed', 'failed', 'unconfirmed')); diff --git a/crates/store/src/lib.rs b/crates/store/src/lib.rs index 97f981a..6794566 100644 --- a/crates/store/src/lib.rs +++ b/crates/store/src/lib.rs @@ -1951,4 +1951,179 @@ impl Store { .await?; Ok(result.rows_affected()) } + + // --- EVM ingest cursor ----------------------------------------------- + + /// Read the saved EVM block-number cursor for a (wallet, chain) pair, if any. + /// + /// The cursor advances only when a log is durably processed, so a crash resumes without + /// missing or double-processing logs — the same guarantee the Stellar path gives via the + /// Horizon paging token. + pub async fn get_evm_cursor( + &self, + wallet_id: Uuid, + chain_id: &str, + ) -> Result, StoreError> { + let block: Option = sqlx::query_scalar( + "SELECT block_number FROM ingest_cursor WHERE wallet_id = $1 AND chain_id = $2", + ) + .bind(wallet_id) + .bind(chain_id) + .fetch_optional(&self.pool) + .await? + .flatten(); + // block_number is stored as BIGINT (i64); EVM block numbers are u64 but fit in i64 for + // any realistic chain (2^63 blocks ≈ 9.2 × 10^18 seconds at 1-second block times). + Ok(block.map(|b| b as u64)) + } + + /// Upsert the EVM block-number cursor for a (wallet, chain) pair (durable resume point). + /// + /// Only called after a log record is durably processed — `updated_at` therefore means "the + /// last block at which this wallet received an EVM deposit", which drives the activity-based + /// backoff tiers in `wallets_due_for_poll`. + pub async fn set_evm_cursor( + &self, + wallet_id: Uuid, + chain_id: &str, + block_number: u64, + ) -> Result<(), StoreError> { + sqlx::query( + r#" + INSERT INTO ingest_cursor (wallet_id, chain_id, block_number, updated_at) + VALUES ($1, $2, $3, now()) + ON CONFLICT (wallet_id, chain_id) + DO UPDATE SET block_number = EXCLUDED.block_number, updated_at = now() + "#, + ) + .bind(wallet_id) + .bind(chain_id) + .bind(block_number as i64) + .execute(&self.pool) + .await?; + Ok(()) + } + + /// Record that a (wallet, chain) pair was polled for EVM logs (whether or not anything new + /// arrived). Mirrors `mark_polled` for the Stellar path. + pub async fn mark_evm_polled( + &self, + wallet_id: Uuid, + chain_id: &str, + ) -> Result<(), StoreError> { + sqlx::query( + r#" + INSERT INTO ingest_cursor (wallet_id, chain_id, last_polled_at, updated_at) + VALUES ($1, $2, now(), 'epoch') + ON CONFLICT (wallet_id, chain_id) + DO UPDATE SET last_polled_at = now() + "#, + ) + .bind(wallet_id) + .bind(chain_id) + .execute(&self.pool) + .await?; + Ok(()) + } + + // --- EVM deposit recording ------------------------------------------- + + /// A new EVM deposit to record (input to the idempotent insert). + /// + /// Deposits are recorded as `unconfirmed` — crediting is gated on confirmation depth (#222). + /// Recording as unconfirmed rather than confirmed ensures that merging the ingest worker + /// before the crediting layer (#222) lands cannot create spendable balances. + pub async fn record_evm_deposit( + &self, + d: &NewEvmDeposit<'_>, + ) -> Result, StoreError> { + let result = sqlx::query_as::<_, Transaction>( + r#" + INSERT INTO transactions + (wallet_id, address_id, direction, asset_code, asset_issuer, amount_stroops, + source_account, destination_account, stellar_tx_hash, operation_index, + chain_id, evm_log_index, status) + VALUES ($1, $2, 'deposit', $3, $4, $5, $6, $7, $8, $9, $10, $11, 'unconfirmed') + RETURNING * + "#, + ) + .bind(d.wallet_id) + .bind(d.address_id) + .bind(&d.token_symbol) + .bind(&d.token_contract) + .bind(d.amount_base_units) + .bind(&d.from_address) + .bind(&d.to_address) + .bind(&d.tx_hash) + .bind(d.log_index as i32) + .bind(&d.chain_id) + .bind(d.log_index as i32) + .fetch_one(&self.pool) + .await; + + match result { + Ok(tx) => Ok(Some(tx)), + Err(e) => match StoreError::from_sqlx_conflict(e) { + StoreError::Conflict => Ok(None), // already recorded — benign dedup + other => Err(other), + }, + } + } + + /// Look up an EVM deposit address by its hex address string within a (wallet, chain) scope. + /// + /// EVM deposit addresses are stored in `addresses.muxed_address` (the column is repurposed as + /// a generic `deposit_address` until the multi-chain schema migration lands). The lookup is + /// case-insensitive because EIP-55 checksum addresses and all-lowercase addresses are the same + /// address. + pub async fn evm_address_by_hex( + &self, + wallet_id: Uuid, + hex_address: &str, + ) -> Result, StoreError> { + let row = sqlx::query_as::<_, Address>( + "SELECT * FROM addresses WHERE wallet_id = $1 AND lower(muxed_address) = lower($2)", + ) + .bind(wallet_id) + .bind(hex_address) + .fetch_optional(&self.pool) + .await?; + Ok(row) + } +} + +/// A new EVM deposit to record via [`Store::record_evm_deposit`]. +/// +/// Amounts are stored as `i64` here for compatibility with the current `amount_stroops BIGINT` +/// column. ERC-20 stablecoins (USDC = 6 decimals, DAI = 18 decimals) transfer amounts fit +/// in `i64` for any realistic transaction value. Full `uint256` support will land with #215. +/// +/// **Status:** EVM deposits are recorded as `unconfirmed`. They must NOT be treated as spendable +/// until the confirmation-depth check in #222 promotes them to `confirmed`. This status is +/// enforced at the DB level (`record_evm_deposit` always inserts `status = 'unconfirmed'`). +#[derive(Debug, Clone)] +pub struct NewEvmDeposit<'a> { + /// The wallet that owns the deposit address. + pub wallet_id: Uuid, + /// The attributed deposit address, if the `to` address was a known customer address. + /// `None` for quarantined deposits (unknown deposit address, or unregistered token). + pub address_id: Option, + /// CAIP-2 chain identifier (e.g. `"eip155:1"`, `"eip155:11155111"`). + pub chain_id: &'a str, + /// ERC-20 token symbol (e.g. `"USDC"`). Stored in `asset_code`. + pub token_symbol: &'a str, + /// Token contract address (EIP-55 checksummed hex). Stored in `asset_issuer` — the issuer + /// concept maps naturally: the contract is the authority for the token. + pub token_contract: &'a str, + /// Transfer amount in the token's base units (i.e. wei-equivalents, not human-readable). + /// For USDC (6 decimals) this is micro-USDC; for DAI (18 decimals) this is attoDAI. + pub amount_base_units: i64, + /// `from` field of the Transfer event (the sender). + pub from_address: &'a str, + /// `to` field of the Transfer event (the recipient — our deposit address). + pub to_address: &'a str, + /// Transaction hash (`0x`-prefixed hex). Stored in `stellar_tx_hash` until column rename. + pub tx_hash: &'a str, + /// Index of this log entry within the transaction (0-based). Used for per-chain dedup. + pub log_index: u32, } diff --git a/docs/ingest-integration.md b/docs/ingest-integration.md index 0a0bf80..2939347 100644 --- a/docs/ingest-integration.md +++ b/docs/ingest-integration.md @@ -408,4 +408,170 @@ but wasteful. > **Document maintainers:** If the dedup strategy or reorg handling changes (e.g., adding a > confirmation-depth check, a reorg detector, or a `pending → confirmed` state machine), update > this document to reflect the new behaviour. The [Known Limitations](#known-limitations) section -> in particular must be kept honest. \ No newline at end of file +> in particular must be kept honest. + +--- + +## EVM Deposit Detection (ERC-20 Transfer Logs) + +> **Added:** Issue #221. **Blocks:** #222 (confirmation/crediting). + +### Architecture + +The EVM ingest worker (`crates/ingest/src/evm.rs`) mirrors the Stellar `Ingestor`'s shape and +reliability contract but operates on a fundamentally different data model: + +| Dimension | Stellar | EVM | +|---|---|---| +| **Scan method** | `Horizon /payments` feed per account | `eth_getLogs` with a block range filter | +| **Cursor type** | Horizon paging token (opaque string) | Block number (`u64`) | +| **Customer attribution** | Muxed id embedded in the payment destination | Deposit address registered in `addresses` table | +| **Dedup key** | `horizon_op_id` (Horizon TOID) | `(chain_id, tx_hash, log_index)` | +| **Initial status** | `confirmed` (Stellar has instant finality) | `unconfirmed` (EVM has probabilistic finality) | + +### What is detected + +The EVM worker detects **ERC-20 Transfer events** only. Specifically, it calls `eth_getLogs` +filtered by: + +- **Address list:** all registered token contracts on the chain (from `RegisteredToken`). +- **Topic 0:** the `Transfer(address,address,uint256)` event signature hash + (`0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef`). + +### Native ETH is out of scope for v1 + +Native ETH transfers emit **no logs**. They are only detectable by inspecting block transactions +(`eth_getBlockByNumber`) or via `debug_traceBlock`, which are significantly more expensive than a +targeted `eth_getLogs` filter — and `debug_traceBlock` is not available on all providers. + +**Decision:** For v1, native ETH deposits are **explicitly out of scope**. This is a deliberate, +documented choice, not a silent omission. + +- Operators must communicate to their users that only ERC-20 tokens are accepted at EVM deposit + addresses. +- A customer who sends ETH directly to their deposit address will **not** have it credited. +- Supporting native ETH is tracked as a future enhancement; the architecture does not prevent it. + +### Security: contract address verification + +**The single most important invariant in the EVM ingest module:** + +Logs are matched on the **emitting contract address** (`log.address`), not on topics alone. + +Any contract can emit a `Transfer` event with arbitrary topics — including a deposit address in +`topics[2]` (the `to` field) and an enormous value in `data`. If the worker attributed transfers +based on topics alone, an attacker could: + +1. Deploy a hostile contract. +2. Call it, emitting a `Transfer` event with an octo deposit address as `to` and e.g. + `2^64 - 1` units as the value. +3. Receive a credit for that amount, without ever sending any tokens. + +The defence is in `EvmIngestor::process_log` (step 1): + +```rust +// Step 1: Verify the log was emitted by a registered token contract. +let token = self.registered_tokens.iter() + .find(|t| t.contract_address.eq_ignore_ascii_case(&log.address)); + +let token = match token { + Some(t) => t, + None => return Ok(Processed::Skipped), // not a registered contract → skip +}; +``` + +This check happens **before** reading any topic. A log from an unregistered contract is always +`Skipped`, regardless of what its topics claim. + +The adversarial test `fake_transfer_from_unregistered_contract_is_skipped` in +`crates/ingest/tests/evm_ingest_tests.rs` is the primary regression guard for this property. + +### Non-standard ERC-20s: fee-on-transfer and rebasing tokens + +Fee-on-transfer tokens (e.g. SafeMoon clones) emit a `Transfer` event with a `value` that is +**larger than the amount actually received** by the `to` address (a fee is deducted in transit). +Rebasing tokens (e.g. AMPL) change holders' balances without emitting Transfer events. + +**Defence:** The token registry (`RegisteredToken`) is the gate. Only tokens explicitly +registered are credited. Operators must not register fee-on-transfer or rebasing tokens until +the crediting layer (#222) has explicit support for them. The `amount_base_units` stored is +the value from the Transfer event — for registered stablecoins (USDC, USDT, DAI) this is +accurate. For fee-on-transfer tokens it would overstate the received amount, which is why +only the registry can admit new tokens. + +### Cursor contract + +The block-number cursor is stored in `ingest_cursor(wallet_id, chain_id)` and advances +**after each log is durably processed**, not at the end of a batch. If the process crashes +mid-range: + +1. On restart, `get_evm_cursor` returns the last durably-processed block. +2. `poll_once` starts from `cursor + 1`. +3. Any logs in the current block that were already processed are caught by the + `(chain_id, tx_hash, log_index)` dedup index in `transactions` and returned as `Duplicate`. + +This is the same crash-safety guarantee the Stellar path gives via the Horizon paging token. + +### Unconfirmed status + +EVM deposits are recorded as `status = 'unconfirmed'`, not `'confirmed'`. They must not be +treated as spendable until issue #222 (confirmation-depth check) promotes them. + +This is enforced at the DB level: `Store::record_evm_deposit` always inserts +`status = 'unconfirmed'`. Merging the ingest worker (#221) before the crediting layer (#222) +cannot create spendable balances. + +The test `evm_deposit_status_is_always_unconfirmed` in `crates/ingest/tests/evm_ingest_tests.rs` +locks this in as a non-negotiable regression guard. + +### Adaptive range bisection + +Different EVM providers cap `eth_getLogs` block ranges differently (Alchemy: 2 000 blocks, +Infura: 10 000 blocks, self-hosted: unlimited). A fixed range would stall on providers with +tighter caps. + +When `eth_getLogs` returns a `RangeTooLarge` error (detected by message pattern matching on +the JSON-RPC error object), `EvmIngestor::poll_once` halves the block range and retries, up +to `MAX_BISECTIONS = 16` times. If the range reaches `MIN_BLOCK_RANGE = 1` and still fails, +the error is surfaced as `EvmIngestError::RangeTooLargeExhausted`. + +### Schema changes (migration 0021) + +Migration `crates/store/migrations/0021_evm_ingest.sql` adds: + +| Table | Change | +|---|---| +| `ingest_cursor` | Nullable `chain_id TEXT` and `block_number BIGINT` columns; unique index `uq_ingest_cursor_wallet_chain (wallet_id, chain_id) WHERE chain_id IS NOT NULL` | +| `transactions` | Nullable `chain_id TEXT` and `evm_log_index INTEGER` columns; unique index `uq_tx_evm_onchain (chain_id, stellar_tx_hash, evm_log_index) WHERE chain_id IS NOT NULL AND ...` | +| `transactions` | Status constraint extended to include `'unconfirmed'` | + +All new columns are nullable — existing Stellar rows are unaffected. The migration is +forward-only and append-only. + +### Store functions added + +| Function | Purpose | +|---|---| +| `Store::get_evm_cursor(wallet_id, chain_id)` | Read the EVM block-number cursor | +| `Store::set_evm_cursor(wallet_id, chain_id, block)` | Advance the cursor after a durable record | +| `Store::mark_evm_polled(wallet_id, chain_id)` | Record "looked at" time (drives backoff tiers) | +| `Store::record_evm_deposit(NewEvmDeposit)` | Idempotent insert of an EVM deposit row | +| `Store::evm_address_by_hex(wallet_id, hex)` | Look up a deposit address by hex (case-insensitive) | + +### Test coverage + +| Test | File | What it verifies | +|---|---|---| +| `erc20_transfer_to_deposit_address_is_recorded` | `evm_ingest_tests.rs` | Happy path; attributed, unconfirmed | +| `dai_transfer_18_decimals_stored_correctly` | `evm_ingest_tests.rs` | 18-decimal amounts fit in `i64` correctly | +| `fake_transfer_from_unregistered_contract_is_skipped` | `evm_ingest_tests.rs` | **Critical security test** | +| `replay_same_log_is_idempotent` | `evm_ingest_tests.rs` | Dedup on `(chain_id, tx_hash, log_index)` | +| `cursor_advances_per_block_and_resume_is_exactly_once` | `evm_ingest_tests.rs` | Crash-resume contract | +| `transfer_of_unregistered_token_to_known_address_is_skipped` | `evm_ingest_tests.rs` | Quarantine of unregistered tokens | +| `range_too_large_error_causes_bisection_not_stall` | `evm_ingest_tests.rs` | Adaptive range bisection | +| `removed_log_is_never_processed` | `evm_ingest_tests.rs` | Reorged logs ignored | +| `transfer_to_non_deposit_address_is_skipped` | `evm_ingest_tests.rs` | Non-deposit addresses ignored | +| `evm_deposit_status_is_always_unconfirmed` | `evm_ingest_tests.rs` | `status = 'unconfirmed'` invariant | + +Unit tests for `EvmLog` parsing (amount, address extraction, hex parsing) are in +`crates/ingest/src/evm.rs` under `#[cfg(test)]`. \ No newline at end of file