diff --git a/Cargo.lock b/Cargo.lock index e3e1e9d3d..c7e24bac6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -12477,6 +12477,7 @@ dependencies = [ "ark-bn254 0.5.0", "ark-ff 0.5.0", "axum", + "backon", "clap", "dotenvy", "eyre", diff --git a/crates/registries/src/world_id.rs b/crates/registries/src/world_id.rs index 30e7dcf0a..1b4a13e52 100644 --- a/crates/registries/src/world_id.rs +++ b/crates/registries/src/world_id.rs @@ -12,7 +12,7 @@ use alloy::{ sol!( /// The registry of World IDs. Each World ID is represented as a leaf in the Merkle tree. #[allow(clippy::too_many_arguments)] - #[sol(rpc, ignore_unlinked)] + #[sol(rpc, ignore_unlinked, extra_derives(Debug, PartialEq, Eq))] WorldIdRegistry, "abi/WorldIDRegistryAbi.json" ); diff --git a/local-setup.sh b/local-setup.sh index 7757e6810..8665e9bb3 100755 --- a/local-setup.sh +++ b/local-setup.sh @@ -150,7 +150,7 @@ start_node() { run_indexer_and_gateway() { # remove the tree_cache_file as we have a new DB everytime we run local_setup rm -f /tmp/tree.mmap - REGISTRY_ADDRESS=$world_id_registry RPC_URL=http://localhost:8545 WS_URL=ws://localhost:8545 DATABASE_URL=postgres://postgres:postgres@localhost:5432/postgres TREE_CACHE_FILE=/tmp/tree.mmap cargo run --release -p world-id-indexer -- --http --indexer > logs/world-id-indexer.log 2>&1 & + REGISTRY_ADDRESS=$world_id_registry RPC_URL=http://localhost:8545 DATABASE_URL=postgres://postgres:postgres@localhost:5432/postgres TREE_CACHE_FILE=/tmp/tree.mmap cargo run --release -p world-id-indexer -- --http --indexer > logs/world-id-indexer.log 2>&1 & indexer_pid=$! echo "started indexer with PID $indexer_pid" wait_for_health 8080 "world-id-indexer" 300 diff --git a/run-indexer-stage.sh b/run-indexer-stage.sh index 1dfed1c34..f25eef3d4 100755 --- a/run-indexer-stage.sh +++ b/run-indexer-stage.sh @@ -11,7 +11,6 @@ REGISTRY_ADDRESS="${REGISTRY_ADDRESS:-0x969947cFED008bFb5e3F32a25A1A2CDdf64d46fe BATCH_SIZE="${BATCH_SIZE:-512}" START_BLOCK="${START_BLOCK:-22827118}" RPC_URL="${RPC_URL:-https://worldchain-mainnet.g.alchemy.com/public}" -WS_URL="${WS_URL:-wss://worldchain-mainnet.g.alchemy.com/public}" DATABASE_URL="${DATABASE_URL:-postgres://postgres:postgres@localhost:5433/indexer_stage}" TREE_CACHE_FILE="${TREE_CACHE_FILE:-$ROOT_DIR/tree-cache-stage.mmap}" DB_NAME="${DB_NAME:-indexer_stage}" @@ -69,7 +68,6 @@ env \ RUN_MODE="$RUN_MODE" \ DATABASE_URL="$DATABASE_URL" \ RPC_URL="$RPC_URL" \ - WS_URL="$WS_URL" \ REGISTRY_ADDRESS="$REGISTRY_ADDRESS" \ START_BLOCK="$START_BLOCK" \ BATCH_SIZE="$BATCH_SIZE" \ diff --git a/services/indexer/.env.example b/services/indexer/.env.example index 16ed6d7f9..df4142ef0 100644 --- a/services/indexer/.env.example +++ b/services/indexer/.env.example @@ -5,10 +5,13 @@ DATABASE_URL=postgres://postgres:postgres@localhost:5432/postgres # Indexer Config RPC_URL=http://localhost:8545 -WS_URL=ws://localhost:8545 REGISTRY_ADDRESS=0x0000000000000000000000000000000000000000 START_BLOCK=0 BATCH_SIZE=64 +# Blocks to stay behind the chain head when polling (avoids ingesting reorg-prone unconfirmed blocks) +CONFIRMATIONS=8 +# Seconds to wait between RPC poll cycles once caught up +POLL_INTERVAL_SECS=2 # HTTP Config HTTP_ADDR=0.0.0.0:8080 diff --git a/services/indexer/Cargo.toml b/services/indexer/Cargo.toml index 8006157a1..0fbf2b9d5 100644 --- a/services/indexer/Cargo.toml +++ b/services/indexer/Cargo.toml @@ -7,7 +7,8 @@ publish = false [dependencies] world-id-services-common = { workspace = true } -tokio = { workspace = true, features = ["rt-multi-thread", "macros"] } +tokio = { workspace = true, features = ["rt-multi-thread", "macros", "sync"] } +backon = { workspace = true } tracing = { workspace = true } tracing-subscriber = { workspace = true, features = ["json"] } telemetry-batteries = { workspace = true } diff --git a/services/indexer/README.md b/services/indexer/README.md index 903b498ee..bc4f4ddb2 100644 --- a/services/indexer/README.md +++ b/services/indexer/README.md @@ -14,7 +14,7 @@ The indexer supports three run modes, selected via the `RUN_MODE` environment va ### Event Processing -The indexer connects to the chain via HTTP + WebSocket RPC. On startup it backfills from the last indexed block (or `START_BLOCK` if the DB is empty), then transitions to live WebSocket streaming. +The indexer connects to the chain via HTTP RPC and polls for events. On startup it backfills from the last indexed block (or `START_BLOCK` if the DB is empty), then keeps polling for newly confirmed logs. Only logs at or below the confirmed head (`chain_head - CONFIRMATIONS`) are ingested, so short-lived reorgs of unconfirmed blocks are never committed. Events are processed through `EventsCommitter`, which: @@ -65,12 +65,13 @@ This restart-on-reorg pattern — detect, rollback state, exit cleanly, re-initi |---|---|---| | `DATABASE_URL` | required | PostgreSQL connection string | | `RPC_URL` | required | HTTP RPC endpoint | -| `WS_URL` | required | WebSocket RPC endpoint | | `REGISTRY_ADDRESS` | required | `AccountRegistry` contract address | | `TREE_CACHE_FILE` | required | Path to mmap-backed tree cache file | | `RUN_MODE` | `both` | `both`, `indexer`, or `http` | | `START_BLOCK` | `0` | Block to start indexing from if DB is empty | | `BATCH_SIZE` | `64` | Blocks per RPC batch during backfill | +| `CONFIRMATIONS` | `8` | Blocks to stay behind the chain head when polling | +| `POLL_INTERVAL_SECS` | `2` | Seconds between RPC poll cycles once caught up | | `TREE_DEPTH` | `30` | Merkle tree depth | | `TREE_MAX_BLOCK_AGE` | `1000` | Blocks of per-leaf history kept for rollback | | `HTTP_ADDR` | `0.0.0.0:8080` | Address for the HTTP server | diff --git a/services/indexer/src/blockchain/events.rs b/services/indexer/src/blockchain/events.rs index 38124d28f..e2ec88a16 100644 --- a/services/indexer/src/blockchain/events.rs +++ b/services/indexer/src/blockchain/events.rs @@ -1,11 +1,24 @@ use alloy::{ - primitives::{Address, FixedBytes, Log, U256}, - sol_types::SolEvent, + primitives::{FixedBytes, U256}, + sol_types::{SolEvent, SolEventInterface}, }; use world_id_registries::world_id::WorldIdRegistry; use super::{BlockchainError, BlockchainResult}; +/// Re-export the alloy-generated registry event types under stable local names. +/// +/// These are the types produced by the `sol!` bindings in `world_id_registries`; +/// the indexer no longer hand-rolls parallel copies of them. +pub use world_id_registries::world_id::WorldIdRegistry::{ + AccountCreated as AccountCreatedEvent, AccountRecovered as AccountRecoveredEvent, + AccountUpdated as AccountUpdatedEvent, AuthenticatorInserted as AuthenticatorInsertedEvent, + AuthenticatorRemoved as AuthenticatorRemovedEvent, RootRecorded as RootRecordedEvent, + WorldIdRegistryEvents as RegistryEvent, +}; + +/// A decoded registry event together with the block metadata of the log it was +/// decoded from. #[derive(Debug, Clone, PartialEq)] pub struct BlockchainEvent { pub block_number: u64, @@ -15,73 +28,22 @@ pub struct BlockchainEvent { pub details: T, } -#[derive(Debug, Clone, PartialEq)] -pub struct AccountCreatedEvent { - pub leaf_index: u64, - pub recovery_address: Address, - pub authenticator_addresses: Vec
, - pub authenticator_pubkeys: Vec, - pub offchain_signer_commitment: U256, -} - -#[derive(Debug, Clone, PartialEq)] -pub struct AccountUpdatedEvent { - pub leaf_index: u64, - pub pubkey_id: u32, - pub new_authenticator_pubkey: U256, - pub old_authenticator_address: Address, - pub new_authenticator_address: Address, - pub old_offchain_signer_commitment: U256, - pub new_offchain_signer_commitment: U256, -} - -#[derive(Debug, Clone, PartialEq)] -pub struct AuthenticatorInsertedEvent { - pub leaf_index: u64, - pub pubkey_id: u32, - pub authenticator_address: Address, - pub new_authenticator_pubkey: U256, - pub old_offchain_signer_commitment: U256, - pub new_offchain_signer_commitment: U256, -} - -#[derive(Debug, Clone, PartialEq)] -pub struct AuthenticatorRemovedEvent { - pub leaf_index: u64, - pub pubkey_id: u32, - pub authenticator_address: Address, - pub authenticator_pubkey: U256, - pub old_offchain_signer_commitment: U256, - pub new_offchain_signer_commitment: U256, -} - -#[derive(Debug, Clone, PartialEq)] -pub struct AccountRecoveredEvent { - pub leaf_index: u64, - pub new_authenticator_address: Address, - pub new_authenticator_pubkey: U256, - pub old_offchain_signer_commitment: U256, - pub new_offchain_signer_commitment: U256, +/// Extension methods on the generated [`RegistryEvent`] enum. +/// +/// `RegistryEvent` aliases a type defined in `world_id_registries`, so these +/// cannot be inherent methods (orphan rule) and live on this trait instead. +pub trait RegistryEventExt: Sized + Clone { + /// The topic-0 signature hashes of every registry event variant, used to + /// build log filters. + fn signatures() -> Vec>; + + /// Decode an RPC log into a [`BlockchainEvent`], pulling block metadata from + /// the log and delegating event-body decoding to the generated bindings. + fn decode(log: &alloy::rpc::types::Log) -> BlockchainResult>; } -#[derive(Debug, Clone, PartialEq)] -pub struct RootRecordedEvent { - pub root: U256, - pub timestamp: U256, -} - -#[derive(Debug, Clone, PartialEq)] -pub enum RegistryEvent { - AccountCreated(AccountCreatedEvent), - AccountUpdated(AccountUpdatedEvent), - AuthenticatorInserted(AuthenticatorInsertedEvent), - AuthenticatorRemoved(AuthenticatorRemovedEvent), - AccountRecovered(AccountRecoveredEvent), - RootRecorded(RootRecordedEvent), -} - -impl RegistryEvent { - pub fn signatures() -> Vec> { +impl RegistryEventExt for RegistryEvent { + fn signatures() -> Vec> { vec![ WorldIdRegistry::AccountCreated::SIGNATURE_HASH, WorldIdRegistry::AccountUpdated::SIGNATURE_HASH, @@ -92,39 +54,15 @@ impl RegistryEvent { ] } - pub fn decode(lg: &alloy::rpc::types::Log) -> BlockchainResult> { - if lg.topics().is_empty() { - return Err(BlockchainError::EmptyTopics); - } - - let event_sig = lg.topics()[0]; + fn decode(log: &alloy::rpc::types::Log) -> BlockchainResult> { + let block_number = log + .block_number + .ok_or(BlockchainError::MissingBlockNumber)?; + let block_hash = log.block_hash.ok_or(BlockchainError::MissingBlockHash)?; + let tx_hash = log.transaction_hash.ok_or(BlockchainError::MissingTxHash)?; + let log_index = log.log_index.ok_or(BlockchainError::MissingLogIndex)?; - let block_number = lg.block_number.ok_or(BlockchainError::MissingBlockNumber)?; - let block_hash = lg.block_hash.ok_or(BlockchainError::MissingBlockHash)?; - let tx_hash = lg.transaction_hash.ok_or(BlockchainError::MissingTxHash)?; - let log_index = lg.log_index.ok_or(BlockchainError::MissingLogIndex)?; - - let details = match event_sig { - WorldIdRegistry::AccountCreated::SIGNATURE_HASH => { - RegistryEvent::AccountCreated(Self::decode_account_created(lg)?) - } - WorldIdRegistry::AccountUpdated::SIGNATURE_HASH => { - RegistryEvent::AccountUpdated(Self::decode_account_updated(lg)?) - } - WorldIdRegistry::AuthenticatorInserted::SIGNATURE_HASH => { - RegistryEvent::AuthenticatorInserted(Self::decode_authenticator_inserted(lg)?) - } - WorldIdRegistry::AuthenticatorRemoved::SIGNATURE_HASH => { - RegistryEvent::AuthenticatorRemoved(Self::decode_authenticator_removed(lg)?) - } - WorldIdRegistry::AccountRecovered::SIGNATURE_HASH => { - RegistryEvent::AccountRecovered(Self::decode_account_recovered(lg)?) - } - WorldIdRegistry::RootRecorded::SIGNATURE_HASH => { - RegistryEvent::RootRecorded(Self::decode_root_recorded(lg)?) - } - _ => return Err(BlockchainError::UnknownEventSignature(event_sig)), - }; + let details = RegistryEvent::decode_log(&log.inner)?.data; Ok(BlockchainEvent { block_number, @@ -134,601 +72,119 @@ impl RegistryEvent { details, }) } - - fn decode_account_created( - lg: &alloy::rpc::types::Log, - ) -> BlockchainResult { - let prim = Log::new(lg.address(), lg.topics().to_vec(), lg.data().data.clone()) - .ok_or_else(|| BlockchainError::InvalidLog)?; - let typed = WorldIdRegistry::AccountCreated::decode_log(&prim) - .map_err(BlockchainError::LogDecode)?; - - // TODO: Validate pubkey is valid affine compressed - Ok(AccountCreatedEvent { - leaf_index: typed.data.leafIndex, - recovery_address: typed.data.recoveryAddress, - authenticator_addresses: typed.data.authenticatorAddresses, - authenticator_pubkeys: typed.data.authenticatorPubkeys, - offchain_signer_commitment: typed.data.offchainSignerCommitment, - }) - } - - fn decode_account_updated( - lg: &alloy::rpc::types::Log, - ) -> BlockchainResult { - let prim = Log::new(lg.address(), lg.topics().to_vec(), lg.data().data.clone()) - .ok_or_else(|| BlockchainError::InvalidLog)?; - let typed = WorldIdRegistry::AccountUpdated::decode_log(&prim) - .map_err(BlockchainError::LogDecode)?; - - Ok(AccountUpdatedEvent { - leaf_index: typed.data.leafIndex, - pubkey_id: typed.data.pubkeyId, - new_authenticator_pubkey: typed.data.newAuthenticatorPubkey, - old_authenticator_address: typed.data.oldAuthenticatorAddress, - new_authenticator_address: typed.data.newAuthenticatorAddress, - old_offchain_signer_commitment: typed.data.oldOffchainSignerCommitment, - new_offchain_signer_commitment: typed.data.newOffchainSignerCommitment, - }) - } - - fn decode_authenticator_inserted( - lg: &alloy::rpc::types::Log, - ) -> BlockchainResult { - let prim = Log::new(lg.address(), lg.topics().to_vec(), lg.data().data.clone()) - .ok_or_else(|| BlockchainError::InvalidLog)?; - let typed = WorldIdRegistry::AuthenticatorInserted::decode_log(&prim) - .map_err(BlockchainError::LogDecode)?; - - Ok(AuthenticatorInsertedEvent { - leaf_index: typed.data.leafIndex, - pubkey_id: typed.data.pubkeyId, - authenticator_address: typed.data.authenticatorAddress, - new_authenticator_pubkey: typed.data.newAuthenticatorPubkey, - old_offchain_signer_commitment: typed.data.oldOffchainSignerCommitment, - new_offchain_signer_commitment: typed.data.newOffchainSignerCommitment, - }) - } - - fn decode_authenticator_removed( - lg: &alloy::rpc::types::Log, - ) -> BlockchainResult { - let prim = Log::new(lg.address(), lg.topics().to_vec(), lg.data().data.clone()) - .ok_or_else(|| BlockchainError::InvalidLog)?; - let typed = WorldIdRegistry::AuthenticatorRemoved::decode_log(&prim) - .map_err(BlockchainError::LogDecode)?; - - Ok(AuthenticatorRemovedEvent { - leaf_index: typed.data.leafIndex, - pubkey_id: typed.data.pubkeyId, - authenticator_address: typed.data.authenticatorAddress, - authenticator_pubkey: typed.data.authenticatorPubkey, - old_offchain_signer_commitment: typed.data.oldOffchainSignerCommitment, - new_offchain_signer_commitment: typed.data.newOffchainSignerCommitment, - }) - } - - fn decode_account_recovered( - lg: &alloy::rpc::types::Log, - ) -> BlockchainResult { - let prim = Log::new(lg.address(), lg.topics().to_vec(), lg.data().data.clone()) - .ok_or_else(|| BlockchainError::InvalidLog)?; - let typed = WorldIdRegistry::AccountRecovered::decode_log(&prim) - .map_err(BlockchainError::LogDecode)?; - - Ok(AccountRecoveredEvent { - leaf_index: typed.data.leafIndex, - new_authenticator_address: typed.data.newAuthenticatorAddress, - new_authenticator_pubkey: typed.data.newAuthenticatorPubkey, - old_offchain_signer_commitment: typed.data.oldOffchainSignerCommitment, - new_offchain_signer_commitment: typed.data.newOffchainSignerCommitment, - }) - } - - fn decode_root_recorded(lg: &alloy::rpc::types::Log) -> BlockchainResult { - let prim = Log::new(lg.address(), lg.topics().to_vec(), lg.data().data.clone()) - .ok_or_else(|| BlockchainError::InvalidLog)?; - let typed = - WorldIdRegistry::RootRecorded::decode_log(&prim).map_err(BlockchainError::LogDecode)?; - - Ok(RootRecordedEvent { - root: typed.data.root, - timestamp: typed.data.timestamp, - }) - } } #[cfg(test)] mod tests { use super::*; use alloy::{ - primitives::{Bytes, FixedBytes, Log, LogData}, + primitives::{Address, Bytes, FixedBytes, Log, LogData, address}, rpc::types::Log as RpcLog, - sol_types::SolEvent, }; - use world_id_registries::world_id::WorldIdRegistry; - /// Helper to create a mock RPC log - fn create_mock_rpc_log( - address: Address, - topics: Vec>, - data: Bytes, - block_number: u64, - tx_hash: FixedBytes<32>, - log_index: u64, - ) -> RpcLog { + /// Helper to create a mock RPC log carrying the given topics and data. + fn mock_rpc_log(topics: Vec>, data: Bytes) -> RpcLog { RpcLog { inner: Log { - address, + address: Address::ZERO, data: LogData::new_unchecked(topics, data), }, block_hash: Some(FixedBytes::ZERO), - block_number: Some(block_number), + block_number: Some(100), block_timestamp: None, - transaction_hash: Some(tx_hash), + transaction_hash: Some(FixedBytes::from([1u8; 32])), transaction_index: Some(0), - log_index: Some(log_index), + log_index: Some(3), removed: false, } } #[test] - fn test_decode_account_created_event() { + fn decodes_account_created_with_metadata() { let leaf_index = 1u64; - let recovery_address = Address::from([1u8; 20]); - let auth_addresses = vec![Address::from([2u8; 20])]; - let auth_pubkeys = vec![U256::from(123)]; - let commitment = U256::from(456); - - let event_data = WorldIdRegistry::AccountCreated { + let recovery_address = address!("0x0000000000000000000000000000000000000011"); + let event = WorldIdRegistry::AccountCreated { leafIndex: leaf_index, recoveryAddress: recovery_address, - authenticatorAddresses: auth_addresses.clone(), - authenticatorPubkeys: auth_pubkeys.clone(), - offchainSignerCommitment: commitment, + authenticatorAddresses: vec![address!("0x0000000000000000000000000000000000000022")], + authenticatorPubkeys: vec![U256::from(123)], + offchainSignerCommitment: U256::from(456), }; - let encoded = event_data.encode_data(); - - let log = create_mock_rpc_log( - Address::ZERO, + let log = mock_rpc_log( vec![ WorldIdRegistry::AccountCreated::SIGNATURE_HASH, FixedBytes::from(U256::from(leaf_index)), recovery_address.into_word(), ], - encoded.into(), - 100, - FixedBytes::from([1u8; 32]), - 0, + event.encode_data().into(), ); - let blockchain_event = RegistryEvent::decode(&log).expect("Decoding should succeed"); - - assert_eq!(blockchain_event.block_number, 100); - assert_eq!(blockchain_event.log_index, 0); - - match blockchain_event.details { + let decoded = RegistryEvent::decode(&log).expect("decoding should succeed"); + assert_eq!(decoded.block_number, 100); + assert_eq!(decoded.log_index, 3); + match decoded.details { RegistryEvent::AccountCreated(ev) => { - assert_eq!(ev.leaf_index, leaf_index); - assert_eq!(ev.recovery_address, recovery_address); - assert_eq!(ev.authenticator_addresses, auth_addresses); - assert_eq!(ev.authenticator_pubkeys, auth_pubkeys); - assert_eq!(ev.offchain_signer_commitment, commitment); - } - _ => panic!("Expected AccountCreated event"), - } - } - - #[test] - fn test_decode_account_updated_event() { - let leaf_index = 1u64; - let pubkey_id = 0u32; - let new_pubkey = U256::from(789); - let old_address = Address::from([1u8; 20]); - let new_address = Address::from([2u8; 20]); - let old_commitment = U256::from(100); - let new_commitment = U256::from(200); - - let event_data = WorldIdRegistry::AccountUpdated { - leafIndex: leaf_index, - pubkeyId: pubkey_id, - newAuthenticatorPubkey: new_pubkey, - oldAuthenticatorAddress: old_address, - newAuthenticatorAddress: new_address, - oldOffchainSignerCommitment: old_commitment, - newOffchainSignerCommitment: new_commitment, - }; - - let encoded = event_data.encode_data(); - - let log = create_mock_rpc_log( - Address::ZERO, - vec![ - WorldIdRegistry::AccountUpdated::SIGNATURE_HASH, - FixedBytes::from(U256::from(leaf_index)), - old_address.into_word(), - new_address.into_word(), - ], - encoded.into(), - 100, - FixedBytes::from([1u8; 32]), - 0, - ); - - let blockchain_event = RegistryEvent::decode(&log).expect("Decoding should succeed"); - - match blockchain_event.details { - RegistryEvent::AccountUpdated(ev) => { - assert_eq!(ev.leaf_index, leaf_index); - assert_eq!(ev.pubkey_id, pubkey_id); - assert_eq!(ev.new_authenticator_pubkey, new_pubkey); - assert_eq!(ev.old_authenticator_address, old_address); - assert_eq!(ev.new_authenticator_address, new_address); - assert_eq!(ev.old_offchain_signer_commitment, old_commitment); - assert_eq!(ev.new_offchain_signer_commitment, new_commitment); - } - _ => panic!("Expected AccountUpdated event"), - } - } - - #[test] - fn test_decode_authenticator_inserted_event() { - let leaf_index = 1u64; - let pubkey_id = 1u32; - let auth_address = Address::from([3u8; 20]); - let new_pubkey = U256::from(999); - let old_commitment = U256::from(100); - let new_commitment = U256::from(200); - - let event_data = WorldIdRegistry::AuthenticatorInserted { - leafIndex: leaf_index, - pubkeyId: pubkey_id, - authenticatorAddress: auth_address, - newAuthenticatorPubkey: new_pubkey, - oldOffchainSignerCommitment: old_commitment, - newOffchainSignerCommitment: new_commitment, - }; - - let encoded = event_data.encode_data(); - - let log = create_mock_rpc_log( - Address::ZERO, - vec![ - WorldIdRegistry::AuthenticatorInserted::SIGNATURE_HASH, - FixedBytes::from(U256::from(leaf_index)), - auth_address.into_word(), - FixedBytes::from(new_pubkey), - ], - encoded.into(), - 100, - FixedBytes::from([1u8; 32]), - 0, - ); - - let blockchain_event = RegistryEvent::decode(&log).expect("Decoding should succeed"); - - match blockchain_event.details { - RegistryEvent::AuthenticatorInserted(ev) => { - assert_eq!(ev.leaf_index, leaf_index); - assert_eq!(ev.pubkey_id, pubkey_id); - assert_eq!(ev.authenticator_address, auth_address); - assert_eq!(ev.new_authenticator_pubkey, new_pubkey); - assert_eq!(ev.old_offchain_signer_commitment, old_commitment); - assert_eq!(ev.new_offchain_signer_commitment, new_commitment); - } - _ => panic!("Expected AuthenticatorInserted event"), - } - } - - #[test] - fn test_decode_authenticator_removed_event() { - let leaf_index = 1u64; - let pubkey_id = 0u32; - let auth_address = Address::from([4u8; 20]); - let pubkey = U256::from(888); - let old_commitment = U256::from(100); - let new_commitment = U256::from(200); - - let event_data = WorldIdRegistry::AuthenticatorRemoved { - leafIndex: leaf_index, - pubkeyId: pubkey_id, - authenticatorAddress: auth_address, - authenticatorPubkey: pubkey, - oldOffchainSignerCommitment: old_commitment, - newOffchainSignerCommitment: new_commitment, - }; - - let encoded = event_data.encode_data(); - - let log = create_mock_rpc_log( - Address::ZERO, - vec![ - WorldIdRegistry::AuthenticatorRemoved::SIGNATURE_HASH, - FixedBytes::from(U256::from(leaf_index)), - auth_address.into_word(), - FixedBytes::from(pubkey), - ], - encoded.into(), - 100, - FixedBytes::from([1u8; 32]), - 0, - ); - - let blockchain_event = RegistryEvent::decode(&log).expect("Decoding should succeed"); - - match blockchain_event.details { - RegistryEvent::AuthenticatorRemoved(ev) => { - assert_eq!(ev.leaf_index, leaf_index); - assert_eq!(ev.pubkey_id, pubkey_id); - assert_eq!(ev.authenticator_address, auth_address); - assert_eq!(ev.authenticator_pubkey, pubkey); - assert_eq!(ev.old_offchain_signer_commitment, old_commitment); - assert_eq!(ev.new_offchain_signer_commitment, new_commitment); + assert_eq!(ev.leafIndex, leaf_index); + assert_eq!(ev.recoveryAddress, recovery_address); + assert_eq!(ev.offchainSignerCommitment, U256::from(456)); } - _ => panic!("Expected AuthenticatorRemoved event"), + other => panic!("expected AccountCreated, got {other:?}"), } } #[test] - fn test_decode_account_recovered_event() { - let leaf_index = 1u64; - let new_address = Address::from([5u8; 20]); - let new_pubkey = U256::from(777); - let old_commitment = U256::from(100); - let new_commitment = U256::from(200); - - let event_data = WorldIdRegistry::AccountRecovered { - leafIndex: leaf_index, - newAuthenticatorAddress: new_address, - newAuthenticatorPubkey: new_pubkey, - oldOffchainSignerCommitment: old_commitment, - newOffchainSignerCommitment: new_commitment, + fn decodes_root_recorded() { + let event = WorldIdRegistry::RootRecorded { + root: U256::from(123456), + timestamp: U256::from(1_000_000), }; - - let encoded = event_data.encode_data(); - - let log = create_mock_rpc_log( - Address::ZERO, - vec![ - WorldIdRegistry::AccountRecovered::SIGNATURE_HASH, - FixedBytes::from(U256::from(leaf_index)), - new_address.into_word(), - FixedBytes::from(new_pubkey), - ], - encoded.into(), - 100, - FixedBytes::from([1u8; 32]), - 0, - ); - - let blockchain_event = RegistryEvent::decode(&log).expect("Decoding should succeed"); - - match blockchain_event.details { - RegistryEvent::AccountRecovered(ev) => { - assert_eq!(ev.leaf_index, leaf_index); - assert_eq!(ev.new_authenticator_address, new_address); - assert_eq!(ev.new_authenticator_pubkey, new_pubkey); - assert_eq!(ev.old_offchain_signer_commitment, old_commitment); - assert_eq!(ev.new_offchain_signer_commitment, new_commitment); - } - _ => panic!("Expected AccountRecovered event"), - } - } - - #[test] - fn test_decode_root_recorded_event() { - let root = U256::from(123456); - let timestamp = U256::from(1000000); - - let event_data = WorldIdRegistry::RootRecorded { root, timestamp }; - - let encoded = event_data.encode_data(); - - let log = create_mock_rpc_log( - Address::ZERO, + let log = mock_rpc_log( vec![ WorldIdRegistry::RootRecorded::SIGNATURE_HASH, - FixedBytes::from(root), + FixedBytes::from(U256::from(123456)), ], - encoded.into(), - 100, - FixedBytes::from([1u8; 32]), - 0, + event.encode_data().into(), ); - let blockchain_event = RegistryEvent::decode(&log).expect("Decoding should succeed"); - - match blockchain_event.details { + let decoded = RegistryEvent::decode(&log).expect("decoding should succeed"); + match decoded.details { RegistryEvent::RootRecorded(ev) => { - assert_eq!(ev.root, root); - assert_eq!(ev.timestamp, timestamp); + assert_eq!(ev.root, U256::from(123456)); + assert_eq!(ev.timestamp, U256::from(1_000_000)); } - _ => panic!("Expected RootRecorded event"), + other => panic!("expected RootRecorded, got {other:?}"), } } #[test] - fn test_decode_log_with_missing_block_number() { - let event_data = WorldIdRegistry::RootRecorded { - root: U256::from(123), - timestamp: U256::from(456), - }; - - let encoded = event_data.encode_data(); - - let mut log = create_mock_rpc_log( - Address::ZERO, - vec![ - WorldIdRegistry::RootRecorded::SIGNATURE_HASH, - FixedBytes::from(U256::from(123)), - ], - encoded.into(), - 100, - FixedBytes::from([1u8; 32]), - 0, - ); - - log.block_number = None; - - let result = RegistryEvent::decode(&log); - assert!(result.is_err(), "Decoding should fail without block number"); + fn signatures_cover_all_variants() { + assert_eq!(RegistryEvent::signatures().len(), 6); } #[test] - fn test_decode_log_with_missing_tx_hash() { - let event_data = WorldIdRegistry::RootRecorded { - root: U256::from(123), - timestamp: U256::from(456), - }; - - let encoded = event_data.encode_data(); - - let mut log = create_mock_rpc_log( - Address::ZERO, - vec![ - WorldIdRegistry::RootRecorded::SIGNATURE_HASH, - FixedBytes::from(U256::from(123)), - ], - encoded.into(), - 100, - FixedBytes::from([1u8; 32]), - 0, - ); - - log.transaction_hash = None; - - let result = RegistryEvent::decode(&log); - assert!( - result.is_err(), - "Decoding should fail without transaction hash" - ); + fn decode_unknown_signature_errors() { + let log = mock_rpc_log(vec![FixedBytes::from([0xFFu8; 32])], Bytes::new()); + assert!(RegistryEvent::decode(&log).is_err()); } #[test] - fn test_decode_log_with_missing_log_index() { - let event_data = WorldIdRegistry::RootRecorded { - root: U256::from(123), - timestamp: U256::from(456), + fn decode_missing_block_number_errors() { + let event = WorldIdRegistry::RootRecorded { + root: U256::from(1), + timestamp: U256::from(2), }; - - let encoded = event_data.encode_data(); - - let mut log = create_mock_rpc_log( - Address::ZERO, + let mut log = mock_rpc_log( vec![ WorldIdRegistry::RootRecorded::SIGNATURE_HASH, - FixedBytes::from(U256::from(123)), + FixedBytes::from(U256::from(1)), ], - encoded.into(), - 100, - FixedBytes::from([1u8; 32]), - 0, + event.encode_data().into(), ); - - log.log_index = None; - - let result = RegistryEvent::decode(&log); - assert!(result.is_err(), "Decoding should fail without log index"); - } - - #[test] - fn test_decode_unknown_event_signature() { - let log = create_mock_rpc_log( - Address::ZERO, - vec![FixedBytes::from([0xFFu8; 32])], - Bytes::new(), - 100, - FixedBytes::from([1u8; 32]), - 0, - ); - - let result = RegistryEvent::decode(&log); - assert!( - result.is_err(), - "Decoding should fail for unknown event signature" - ); - } - - #[test] - fn test_decode_log_with_no_topics() { - let log = RpcLog { - inner: Log { - address: Address::ZERO, - data: LogData::new_unchecked(vec![], Bytes::new()), - }, - block_hash: Some(FixedBytes::ZERO), - block_number: Some(100), - block_timestamp: None, - transaction_hash: Some(FixedBytes::from([1u8; 32])), - transaction_index: Some(0), - log_index: Some(0), - removed: false, - }; - - let result = RegistryEvent::decode(&log); - assert!(result.is_err(), "Decoding should fail with no topics"); - } - - #[test] - fn test_registry_event_signatures() { - let signatures = RegistryEvent::signatures(); - - assert_eq!(signatures.len(), 6); - - assert!(signatures.contains(&WorldIdRegistry::AccountCreated::SIGNATURE_HASH)); - assert!(signatures.contains(&WorldIdRegistry::AccountUpdated::SIGNATURE_HASH)); - assert!(signatures.contains(&WorldIdRegistry::AuthenticatorInserted::SIGNATURE_HASH)); - assert!(signatures.contains(&WorldIdRegistry::AuthenticatorRemoved::SIGNATURE_HASH)); - assert!(signatures.contains(&WorldIdRegistry::AccountRecovered::SIGNATURE_HASH)); - assert!(signatures.contains(&WorldIdRegistry::RootRecorded::SIGNATURE_HASH)); - } - - #[test] - fn test_decode_account_created_with_multiple_authenticators() { - let leaf_index = 1u64; - let recovery_address = Address::from([1u8; 20]); - let auth_addresses = vec![ - Address::from([2u8; 20]), - Address::from([3u8; 20]), - Address::from([4u8; 20]), - ]; - let auth_pubkeys = vec![U256::from(123), U256::from(456), U256::from(789)]; - let commitment = U256::from(999); - - let event_data = WorldIdRegistry::AccountCreated { - leafIndex: leaf_index, - recoveryAddress: recovery_address, - authenticatorAddresses: auth_addresses.clone(), - authenticatorPubkeys: auth_pubkeys.clone(), - offchainSignerCommitment: commitment, - }; - - let encoded = event_data.encode_data(); - - let log = create_mock_rpc_log( - Address::ZERO, - vec![ - WorldIdRegistry::AccountCreated::SIGNATURE_HASH, - FixedBytes::from(U256::from(leaf_index)), - recovery_address.into_word(), - ], - encoded.into(), - 100, - FixedBytes::from([1u8; 32]), - 0, - ); - - let blockchain_event = RegistryEvent::decode(&log).expect("Decoding should succeed"); - - match blockchain_event.details { - RegistryEvent::AccountCreated(ev) => { - assert_eq!(ev.authenticator_addresses.len(), 3); - assert_eq!(ev.authenticator_pubkeys.len(), 3); - assert_eq!(ev.authenticator_addresses, auth_addresses); - assert_eq!(ev.authenticator_pubkeys, auth_pubkeys); - } - _ => panic!("Expected AccountCreated event"), - } + log.block_number = None; + assert!(matches!( + RegistryEvent::decode(&log), + Err(BlockchainError::MissingBlockNumber) + )); } } diff --git a/services/indexer/src/blockchain/mod.rs b/services/indexer/src/blockchain/mod.rs index 1518d06bd..ae7bae8b9 100644 --- a/services/indexer/src/blockchain/mod.rs +++ b/services/indexer/src/blockchain/mod.rs @@ -1,32 +1,30 @@ -use std::sync::{ - Arc, - atomic::{AtomicU64, Ordering}, -}; +use std::{sync::Arc, time::Duration}; use alloy::{ primitives::{Address, FixedBytes}, - providers::{DynProvider, Provider, ProviderBuilder, WsConnect}, - pubsub::Subscription, + providers::{DynProvider, Provider}, rpc::types::{Filter, Log}, }; +use backon::{ExponentialBuilder, Retryable}; use futures_util::{Stream, StreamExt, TryStreamExt, stream}; use thiserror::Error; +use tokio::{sync::Semaphore, task::JoinSet}; pub use crate::blockchain::events::{ AccountCreatedEvent, AccountRecoveredEvent, AccountUpdatedEvent, AuthenticatorInsertedEvent, - AuthenticatorRemovedEvent, BlockchainEvent, RegistryEvent, RootRecordedEvent, + AuthenticatorRemovedEvent, BlockchainEvent, RegistryEvent, RegistryEventExt, RootRecordedEvent, }; mod events; -static WS_BUFFER_SIZE: usize = 1024; +/// Maximum number of concurrent `eth_getLogs` requests in flight while +/// fetching a block range. Bounds RPC fan-out across the whole program. +const SEMAPHORE_PERMIT: usize = 16; pub type BlockchainResult = Result; #[derive(Debug, Error)] pub enum BlockchainError { - #[error("failed to connect ws provider: {0}")] - WsProvider(#[source] Box), #[error("rpc error: {0}")] Rpc(#[source] Box), #[error("invalid log for decoding")] @@ -45,56 +43,32 @@ pub enum BlockchainError { MissingLogIndex, #[error("unknown event signature: {0:?}")] UnknownEventSignature(FixedBytes<32>), - #[error("websocket subscription closed unexpectedly")] - WsSubscriptionClosed, + #[error(transparent)] + AlloySolTypes(#[from] alloy::sol_types::Error), } pub struct Blockchain { http_provider: DynProvider, - ws_provider: DynProvider, world_id_registry: Address, } impl Blockchain { - /// Creates a new [`Blockchain`] instance used to stream events from the blockchain. + /// Creates a new [`Blockchain`] instance used to poll events from the blockchain. /// Note: We consider any errors as fatal and stop the stream. /// /// # Arguments /// /// * `http_provider` - A pre-built HTTP provider (e.g. from [`ProviderArgs::http()`]). - /// * `ws_rpc_url` - The WebSocket RPC URL to use for the blockchain. /// * `world_id_registry` - The address of the World ID registry. /// /// # Returns /// /// A new [`Blockchain`] instance. - pub async fn new( - http_provider: DynProvider, - ws_rpc_url: &str, - world_id_registry: Address, - ) -> BlockchainResult { - // Disable internal WS reconnect so drops surface immediately as errors. - let ws_connect = WsConnect::new(ws_rpc_url).with_max_retries(0); - let ws_provider = DynProvider::new( - ProviderBuilder::new() - .connect_ws(ws_connect) - .await - .map_err(|err| BlockchainError::WsProvider(Box::new(err)))?, - ); - - ws_provider - .client() - .pubsub_frontend() - .ok_or_else(|| { - BlockchainError::WsProvider("missing pubsub frontend on ws provider".into()) - })? - .set_channel_size(WS_BUFFER_SIZE); // Increase buffer size to avoid losing events - - Ok(Self { + pub fn new(http_provider: DynProvider, world_id_registry: Address) -> Self { + Self { http_provider, - ws_provider, world_id_registry, - }) + } } /// Returns a [`WorldIdRegistryInstance`] bound to the HTTP provider. @@ -107,197 +81,67 @@ impl Blockchain { ) } - /// Streams World Tree events from the blockchain. - /// - /// Concatenates [`Self::backfill_stream`] with [`Self::websocket_stream`] and - /// decodes each log into a [`BlockchainEvent`]. The last - /// block number seen during backfill is tracked via a shared atomic counter - /// so that [`Self::websocket_stream`] knows where to pick up. - /// - /// # Arguments - /// - /// * `from_block` - The block number to start streaming from. - /// * `batch_size` - The batch size to use for the backfill stage. + /// Polls the blockchain for registry events over HTTP and decodes each log + /// into a [`BlockchainEvent`]. /// - /// # Returns - /// - /// A stream of [`BlockchainEvent`]. The stream terminates - /// after the first error. - pub fn backfill_and_stream_events( - &self, - from_block: u64, - batch_size: u64, - ) -> impl Stream>> + Unpin + '_ { - let last_block = Arc::new(AtomicU64::new(from_block.saturating_sub(1))); - - let backfill = self.backfill_stream(from_block, batch_size, last_block.clone()); - let ws = self.websocket_stream(last_block, batch_size); - - backfill.chain(ws).stop_after_first_error() - } - - /// Returns a decoded backfill event stream and a shared atomic holding the - /// last block number that was fetched. The atomic is updated during - /// streaming and should be read after the stream is fully consumed. + /// Only logs at or below the *confirmed* head (`chain_head - confirmations`) + /// are emitted. This keeps the indexer behind the unsafe head so that + /// short-lived reorgs of preconfirmed blocks never reach the committer, + /// which is the failure mode the previous WebSocket subscription suffered + /// from. A single HTTP provider determines both the head and the logs, so + /// there is no cross-provider view mismatch. /// - /// Each raw log is decoded into a [`BlockchainEvent`]; - /// decode failures are surfaced as [`BlockchainError`] stream items. + /// The first poll fetches the full backlog (`from_block ..= confirmed_head`) + /// in one step, internally chunked by [`Self::fetch_logs_in_batches`]. Each + /// subsequent poll waits `poll_interval`, then emits any newly confirmed + /// logs. /// /// # Arguments /// - /// * `from_block` - The block number to start backfilling from. - /// * `batch_size` - The batch size to use for the backfill stage. + /// * `from_block` - The block number to start polling from. + /// * `batch_size` - The batch size to use when fetching log ranges. + /// * `confirmations` - Number of blocks to stay behind the chain head. + /// * `poll_interval` - How long to wait between polls once caught up. /// /// # Returns /// - /// A tuple of (stream, last_block_atomic). The stream terminates after - /// the first error. - pub fn backfill_events( + /// A stream of decoded [`BlockchainEvent`]. The stream + /// terminates after the first error (e.g. an RPC failure). + pub fn poll_events( &self, from_block: u64, batch_size: u64, - ) -> ( - impl Stream>> + Unpin, - Arc, - ) { - let last_block = Arc::new(AtomicU64::new(from_block.saturating_sub(1))); - let stream = self - .backfill_stream(from_block, batch_size, last_block.clone()) - .stop_after_first_error(); - (stream, last_block) - } - - /// Starts a WebSocket log subscription and bridges the gap between the - /// backfill stage and live events. - /// - /// 1. Subscribes to new logs via WebSocket (buffered to [`WS_BUFFER_SIZE`]). - /// 2. Waits for the first live event and extracts its block number. - /// 3. Fetches any logs that may have been missed between - /// `backfill_to_block + 1` and the first event's block (exclusive). - /// 4. Returns a stream that emits, in order: the gap-fill logs, the first - /// live event, then the remaining live WebSocket events. - /// - /// All errors (including setup failures) are emitted as stream items rather - /// than returned as an outer `Result`, consistent with how - /// [`Self::fetch_logs_in_batches`] emits RPC errors. - /// - /// # Arguments - /// - /// * `backfill_to_block` - Shared atomic holding the last block the backfill - /// stage processed. Loaded lazily when the stream is first polled so the - /// backfill has time to update it. - /// * `batch_size` - The batch size to use for the websocket stage. - /// - /// # Returns - /// - /// A stream of decoded [`BlockchainResult>`]. - /// Returns [`BlockchainError::WsSubscriptionClosed`] and stops in case the websocket is dropped. - fn websocket_stream( - &self, - backfill_to_block: Arc, - batch_size: u64, - ) -> impl Stream>> + Unpin { - stream::once(async move { - let backfill_to_block = backfill_to_block.load(Ordering::Relaxed); - let filter = Filter::new() - .address(self.world_id_registry) - .event_signature(RegistryEvent::signatures()); - - let sub = self.subscribe_logs(&filter).await?; - - let mut ws_stream = sub.into_stream(); - - // Wait for the first live event to determine how far we need to gap-fill. - let first_log = ws_stream - .next() - .await - .ok_or(BlockchainError::WsSubscriptionClosed)?; - - let block_number = first_log - .block_number - .ok_or(BlockchainError::MissingBlockNumber)?; - crate::metrics::set_chain_head_block(block_number); - - // Fetch any logs between the end of the backfill and the first live - // event's block (exclusive to avoid duplicates). - // Edge case: Use `saturating_sub` to avoid underflow when `block_number` is 0. - let missed_logs = self - .fetch_logs_in_batches( - backfill_to_block.saturating_add(1), - block_number.saturating_sub(1), - batch_size, - ) - .map(|r| r.and_then(|log| RegistryEvent::decode(&log))); - - // Chain: gap-fill logs → first live event → remaining WS events - let first_decoded = RegistryEvent::decode(&first_log); - Ok::<_, BlockchainError>( - missed_logs - .chain(stream::iter(std::iter::once(first_decoded))) - .chain(ws_stream.map(|log| { - if let Some(block_number) = log.block_number { - crate::metrics::set_chain_head_block(block_number); - } - RegistryEvent::decode(&log) - })) - // If the websocket subscription is closed, we return an error. - .chain(stream::once(async { - Err(BlockchainError::WsSubscriptionClosed) - })), - ) - }) - .try_flatten() - .boxed() - } + confirmations: u64, + poll_interval: Duration, + ) -> impl Stream>> + Unpin + '_ { + tracing::info!(?from_block, confirmations, "polling from block"); - /// Iteratively backfills logs from the blockchain. - /// - /// The idea here is that while we fetch logs, the head of the chain moves. - /// To prevent having to buffer a lot of incoming logs from the websocket, - /// we try to get within batch size distance from the head of the chain - /// before switching to the websocket stage. When we reach that point, we - /// fetch one final range covering the remaining blocks, store the last - /// fetched block in `last_block`, and then terminate. - /// - /// # Arguments - /// - /// * `from_block` - The block number to start backfilling from. - /// * `batch_size` - The batch size to use for the backfill stage. - /// * `last_block` - A shared atomic counter to store the last fetched block. - /// - /// # Returns - /// - /// A stream of decoded [`BlockchainResult>`]. - fn backfill_stream( - &self, - from_block: u64, - batch_size: u64, - last_block: Arc, - ) -> impl Stream>> + Unpin { - tracing::info!(?from_block, "backfilling from block"); - - stream::try_unfold((from_block, false), move |(current_from, done)| { - let last_block = last_block.clone(); - async move { - if done { - return Ok::<_, BlockchainError>(None); + stream::try_unfold( + (from_block, true), + move |(current_from, first)| async move { + // Wait between poll cycles once we have caught up. The first cycle + // runs immediately so the initial backfill isn't delayed. + if !first { + tokio::time::sleep(poll_interval).await; } - let latest_block_number = self.get_block_number().await?; - crate::metrics::set_chain_head_block(latest_block_number); + loop { + let chain_head = self.get_block_number().await?; + crate::metrics::set_chain_head_block(chain_head); - // We emit one more range here - let is_last = latest_block_number.saturating_sub(current_from) < batch_size; - if is_last { - last_block.store(latest_block_number, Ordering::Relaxed); - } + let confirmed_head = chain_head.saturating_sub(confirmations); + if confirmed_head >= current_from { + return Ok::<_, BlockchainError>(Some(( + (current_from, confirmed_head), + (confirmed_head + 1, false), + ))); + } - Ok(Some(( - (current_from, latest_block_number), - (latest_block_number + 1, is_last), - ))) - } - }) + // No newly confirmed blocks yet; wait and re-check. + tokio::time::sleep(poll_interval).await; + } + }, + ) .map_ok(move |(from, to)| { self.fetch_logs_in_batches(from, to, batch_size) .map(|r| r.and_then(|log| RegistryEvent::decode(&log))) @@ -306,15 +150,18 @@ impl Blockchain { .boxed() } - /// Fetches logs in batches to avoid exceeding RPC provider's max range limits. + /// Fetches logs across a block range, concurrently and in batches. /// - /// This function chunks the block range into smaller batches and streams logs - /// for each batch sequentially, yielding logs as they're fetched. + /// The range is processed in bounded windows of `SEMAPHORE_PERMIT * + /// batch_size` blocks. Each window fans out one task per `batch_size` chunk + /// (capped at [`SEMAPHORE_PERMIT`] concurrent requests), then emits the + /// window's logs in block order before moving to the next window. Windowing + /// keeps memory and task count bounded even for a large initial backfill. /// /// # Arguments /// * `from_block` - The block number to start fetching logs from. /// * `to_block` - The block number to stop fetching logs at. - /// * `batch_size` - The batch size to use for the fetch stage. + /// * `batch_size` - The number of blocks per `eth_getLogs` request. /// /// # Returns /// @@ -325,56 +172,102 @@ impl Blockchain { from_block: u64, to_block: u64, batch_size: u64, - ) -> impl Stream> + Unpin { - let initial_state = ( - from_block, // start block - 0usize, // total logs fetched; we keep this only for logging - ); - - stream::try_unfold( - initial_state, - move |(current_from, total_logs)| async move { - if current_from > to_block { - tracing::info!( - "Backfill step complete: fetched {} total logs from block {} to {}", - total_logs, - from_block, - to_block - ); - return Ok::<_, BlockchainError>(None); - } - - let current_to = std::cmp::min(current_from + batch_size - 1, to_block); - - tracing::debug!( - "Fetching logs from block {} to {}", - current_from, - current_to - ); + ) -> impl Stream> + Unpin + '_ { + let window = SEMAPHORE_PERMIT as u64 * batch_size; - let batch_filter = Filter::new() - .address(self.world_id_registry) - .event_signature(RegistryEvent::signatures()) - .from_block(current_from) - .to_block(current_to); + stream::try_unfold(from_block, move |current_from| async move { + if current_from > to_block { + return Ok::<_, BlockchainError>(None); + } - let logs = self.get_logs(&batch_filter).await?; + let window_end = current_from.saturating_add(window - 1).min(to_block); - tracing::debug!("Fetched {} logs in batch", logs.len()); + let logs = self + .fetch_window_parallel(current_from, window_end, batch_size) + .await?; - let new_total = total_logs + logs.len(); - let next_from = current_to + 1; + tracing::debug!( + from = current_from, + to = window_end, + count = logs.len(), + "fetched log window" + ); - Ok(Some(( - stream::iter(logs.into_iter().map(Ok)), - (next_from, new_total), - ))) - }, - ) + Ok(Some(( + stream::iter(logs.into_iter().map(Ok)), + window_end + 1, + ))) + }) .try_flatten() .boxed() } + /// Fetches all logs in `[from_block, to_block]` concurrently, one task per + /// `batch_size` chunk, bounded by a [`SEMAPHORE_PERMIT`]-permit semaphore. + /// Each task retries transient RPC failures with exponential backoff. + /// + /// Tasks complete out of order, so the collected batches are re-sorted by + /// their chunk index to restore block order before being flattened. + async fn fetch_window_parallel( + &self, + from_block: u64, + to_block: u64, + batch_size: u64, + ) -> BlockchainResult> { + // Build the ordered batch ranges covering the window. + let mut ranges = Vec::new(); + let mut cursor = from_block; + while cursor <= to_block { + let end = cursor.saturating_add(batch_size - 1).min(to_block); + ranges.push((cursor, end)); + if end == u64::MAX { + break; + } + cursor = end + 1; + } + + let semaphore = Arc::new(Semaphore::new(SEMAPHORE_PERMIT)); + let mut join_set: JoinSet)>> = JoinSet::new(); + + for (idx, (start, end)) in ranges.into_iter().enumerate() { + let provider = self.http_provider.clone(); + let registry = self.world_id_registry; + let semaphore = semaphore.clone(); + + join_set.spawn(async move { + // Cap the number of concurrent in-flight requests. + let _permit = semaphore + .acquire_owned() + .await + .map_err(|e| BlockchainError::Rpc(Box::new(e)))?; + + let filter = Filter::new() + .address(registry) + .event_signature(RegistryEvent::signatures()) + .from_block(start) + .to_block(end); + + let logs = (|| async { provider.get_logs(&filter).await }) + .retry(ExponentialBuilder::default().with_jitter()) + .await + .map_err(|e| BlockchainError::Rpc(Box::new(e)))?; + + Ok((idx, logs)) + }); + } + + // Collect all batches; fail fast on the first error (dropping the + // JoinSet aborts any still-running tasks). + let mut batches: Vec<(usize, Vec)> = Vec::new(); + while let Some(joined) = join_set.join_next().await { + let batch = joined.map_err(|e| BlockchainError::Rpc(Box::new(e)))??; + batches.push(batch); + } + + batches.sort_by_key(|(idx, _)| *idx); + Ok(batches.into_iter().flat_map(|(_, logs)| logs).collect()) + } + async fn get_block_number(&self) -> BlockchainResult { self.http_provider .get_block_number() @@ -382,41 +275,38 @@ impl Blockchain { .map_err(|err| BlockchainError::Rpc(Box::new(err))) } - async fn get_logs(&self, filter: &Filter) -> BlockchainResult> { - self.http_provider - .get_logs(filter) - .await - .map_err(|err| BlockchainError::Rpc(Box::new(err))) + /// Current chain head (tip) block number. + pub(crate) async fn current_block(&self) -> BlockchainResult { + self.get_block_number().await } - async fn subscribe_logs(&self, filter: &Filter) -> BlockchainResult> { - self.ws_provider - .subscribe_logs(filter) - .await - .map_err(|err| BlockchainError::Rpc(Box::new(err))) + /// Fetch and decode all registry events in `[from_block, to_block]`, + /// concurrently and in order. The caller is responsible for keeping the + /// range bounded (e.g. one poll window) to bound memory and task count. + pub(crate) async fn fetch_events( + &self, + from_block: u64, + to_block: u64, + batch_size: u64, + ) -> BlockchainResult>> { + let logs = self + .fetch_window_parallel(from_block, to_block, batch_size) + .await?; + logs.iter().map(|log| RegistryEvent::decode(log)).collect() } -} - -/// Small extension trait to stop a stream after the first error. -pub trait StopAfterFirstErrorExt: Stream> + Sized { - fn stop_after_first_error(self) -> impl Stream> + Unpin; -} - -impl StopAfterFirstErrorExt for S -where - S: Stream> + Sized + Unpin, -{ - fn stop_after_first_error(self) -> impl Stream> + Unpin { - self.scan(false, |seen_err, item| { - if *seen_err { - return std::future::ready(None); - } - if item.is_err() { - *seen_err = true; - } - - std::future::ready(Some(item)) - }) + /// Fetch the canonical block hash for `number`, or `None` if the block is + /// not (yet) present on the canonical chain. Used to detect reorgs of + /// already-ingested, not-yet-confirmed blocks. + pub(crate) async fn canonical_block_hash( + &self, + number: u64, + ) -> BlockchainResult> { + let block = self + .http_provider + .get_block_by_number(number.into()) + .await + .map_err(|err| BlockchainError::Rpc(Box::new(err)))?; + Ok(block.map(|b| b.header.hash.into())) } } diff --git a/services/indexer/src/config.rs b/services/indexer/src/config.rs index 177f193a0..14333014b 100644 --- a/services/indexer/src/config.rs +++ b/services/indexer/src/config.rs @@ -97,7 +97,6 @@ pub struct GlobalConfig { pub run_mode: RunMode, pub db_url: String, pub provider: ProviderArgs, - pub ws_rpc_url: String, pub registry_address: Address, pub tree_cache: TreeCacheConfig, } @@ -164,6 +163,11 @@ pub struct IndexerConfig { pub start_block: u64, pub batch_size: u64, pub tree_max_block_age: u64, + /// Number of blocks to stay behind the chain head when polling, so that + /// short-lived reorgs of unconfirmed blocks are never ingested. + pub confirmations: u64, + /// How long to wait between RPC poll cycles once caught up. + pub poll_interval_secs: u64, } impl IndexerConfig { @@ -195,6 +199,15 @@ impl IndexerConfig { .ok() .and_then(|s| s.parse().ok()) .unwrap_or(1000), + confirmations: std::env::var("CONFIRMATIONS") + .ok() + .and_then(|s| s.parse().ok()) + .unwrap_or(8), + poll_interval_secs: std::env::var("POLL_INTERVAL_SECS") + .ok() + .and_then(|s| s.parse::().ok()) + .filter(|&v| v > 0) + .unwrap_or(2), }; tracing::info!("✔️ Indexer config loaded from env"); config @@ -248,8 +261,6 @@ pub enum ConfigError { MissingDatabaseUrl, #[error("RPC_URL environment variable is required")] MissingRpcUrl, - #[error("WS_URL environment variable is required")] - MissingWsUrl, #[error("REGISTRY_ADDRESS environment variable is required")] MissingRegistryAddress, #[error( @@ -289,8 +300,6 @@ impl GlobalConfig { return Err(ConfigError::MissingRpcUrl); } - let ws_rpc_url = std::env::var("WS_URL").map_err(|_| ConfigError::MissingWsUrl)?; - let registry_address_str = std::env::var("REGISTRY_ADDRESS").map_err(|_| ConfigError::MissingRegistryAddress)?; let registry_address = registry_address_str.parse::
().map_err(|e| { @@ -304,7 +313,6 @@ impl GlobalConfig { run_mode, db_url, provider, - ws_rpc_url, registry_address, tree_cache, }) @@ -331,7 +339,6 @@ mod tests { env::remove_var("ENVIRONMENT"); env::remove_var("DATABASE_URL"); env::remove_var("RPC_URL"); - env::remove_var("WS_URL"); env::remove_var("REGISTRY_ADDRESS"); env::remove_var("RUN_MODE"); @@ -551,7 +558,6 @@ mod tests { clear_all_config_env(); set_env("DATABASE_URL", "postgresql://localhost/test"); set_env("RPC_URL", "http://localhost:8545"); - set_env("WS_URL", "ws://localhost:8545"); set_env( "REGISTRY_ADDRESS", "0x0000000000000000000000000000000000000001", @@ -566,7 +572,6 @@ mod tests { assert_eq!(config.environment, Environment::Development); assert_eq!(config.db_url, "postgresql://localhost/test"); assert!(config.provider.http.is_some()); - assert_eq!(config.ws_rpc_url, "ws://localhost:8545"); assert_eq!( config.registry_address.to_string(), "0x0000000000000000000000000000000000000001" @@ -579,7 +584,6 @@ mod tests { fn test_global_config_missing_database_url() { clear_all_config_env(); set_env("RPC_URL", "http://localhost:8545"); - set_env("WS_URL", "ws://localhost:8545"); set_env( "REGISTRY_ADDRESS", "0x0000000000000000000000000000000000000001", @@ -596,7 +600,6 @@ mod tests { fn test_global_config_missing_rpc_url() { clear_all_config_env(); set_env("DATABASE_URL", "postgresql://localhost/test"); - set_env("WS_URL", "ws://localhost:8545"); set_env( "REGISTRY_ADDRESS", "0x0000000000000000000000000000000000000001", @@ -608,30 +611,12 @@ mod tests { assert!(result.unwrap_err().to_string().contains("RPC_URL")); } - #[test] - #[serial] - fn test_global_config_missing_ws_url() { - clear_all_config_env(); - set_env("DATABASE_URL", "postgresql://localhost/test"); - set_env("RPC_URL", "http://localhost:8545"); - set_env( - "REGISTRY_ADDRESS", - "0x0000000000000000000000000000000000000001", - ); - set_env("TREE_CACHE_FILE", "/tmp/test_cache"); - - let result = GlobalConfig::from_env(); - assert!(result.is_err()); - assert!(result.unwrap_err().to_string().contains("WS_URL")); - } - #[test] #[serial] fn test_global_config_missing_registry_address() { clear_all_config_env(); set_env("DATABASE_URL", "postgresql://localhost/test"); set_env("RPC_URL", "http://localhost:8545"); - set_env("WS_URL", "ws://localhost:8545"); set_env("TREE_CACHE_FILE", "/tmp/test_cache"); let result = GlobalConfig::from_env(); @@ -645,7 +630,6 @@ mod tests { clear_all_config_env(); set_env("DATABASE_URL", "postgresql://localhost/test"); set_env("RPC_URL", "http://localhost:8545"); - set_env("WS_URL", "ws://localhost:8545"); set_env("REGISTRY_ADDRESS", "invalid_address"); set_env("TREE_CACHE_FILE", "/tmp/test_cache"); diff --git a/services/indexer/src/db/mod.rs b/services/indexer/src/db/mod.rs index c50fa509e..387134d73 100644 --- a/services/indexer/src/db/mod.rs +++ b/services/indexer/src/db/mod.rs @@ -20,6 +20,8 @@ pub enum DBError { Migrate(#[from] sqlx::migrate::MigrateError), #[error("unknown event type: {0}")] UnknownEventType(String), + #[error("unsupported registry event: {0}")] + UnsupportedEvent(String), #[error("missing required field '{field}' in event data")] MissingEventField { field: String }, #[error("invalid value for field '{field}' in event data: {reason}")] diff --git a/services/indexer/src/db/world_id_registry_events.rs b/services/indexer/src/db/world_id_registry_events.rs index cb5462633..259bb5fb6 100644 --- a/services/indexer/src/db/world_id_registry_events.rs +++ b/services/indexer/src/db/world_id_registry_events.rs @@ -85,54 +85,54 @@ impl<'a> TryFrom<&'a str> for WorldIdRegistryEventType { pub fn serialize_account_created(ev: &AccountCreatedEvent) -> serde_json::Value { serde_json::json!({ - "recovery_address": format!("{:?}", ev.recovery_address), - "authenticator_addresses": ev.authenticator_addresses.iter() + "recovery_address": format!("{:?}", ev.recoveryAddress), + "authenticator_addresses": ev.authenticatorAddresses.iter() .map(|a| format!("{:?}", a)) .collect::>(), - "authenticator_pubkeys": ev.authenticator_pubkeys.iter() + "authenticator_pubkeys": ev.authenticatorPubkeys.iter() .map(|p| format!("{:#x}", p)) .collect::>(), - "offchain_signer_commitment": format!("{:#x}", ev.offchain_signer_commitment), + "offchain_signer_commitment": format!("{:#x}", ev.offchainSignerCommitment), }) } pub fn serialize_account_updated(ev: &AccountUpdatedEvent) -> serde_json::Value { serde_json::json!({ - "pubkey_id": ev.pubkey_id, - "new_authenticator_pubkey": format!("{:#x}", ev.new_authenticator_pubkey), - "old_authenticator_address": format!("{:?}", ev.old_authenticator_address), - "new_authenticator_address": format!("{:?}", ev.new_authenticator_address), - "old_offchain_signer_commitment": format!("{:#x}", ev.old_offchain_signer_commitment), - "new_offchain_signer_commitment": format!("{:#x}", ev.new_offchain_signer_commitment), + "pubkey_id": ev.pubkeyId, + "new_authenticator_pubkey": format!("{:#x}", ev.newAuthenticatorPubkey), + "old_authenticator_address": format!("{:?}", ev.oldAuthenticatorAddress), + "new_authenticator_address": format!("{:?}", ev.newAuthenticatorAddress), + "old_offchain_signer_commitment": format!("{:#x}", ev.oldOffchainSignerCommitment), + "new_offchain_signer_commitment": format!("{:#x}", ev.newOffchainSignerCommitment), }) } pub fn serialize_authenticator_inserted(ev: &AuthenticatorInsertedEvent) -> serde_json::Value { serde_json::json!({ - "pubkey_id": ev.pubkey_id, - "authenticator_address": format!("{:?}", ev.authenticator_address), - "new_authenticator_pubkey": format!("{:#x}", ev.new_authenticator_pubkey), - "old_offchain_signer_commitment": format!("{:#x}", ev.old_offchain_signer_commitment), - "new_offchain_signer_commitment": format!("{:#x}", ev.new_offchain_signer_commitment), + "pubkey_id": ev.pubkeyId, + "authenticator_address": format!("{:?}", ev.authenticatorAddress), + "new_authenticator_pubkey": format!("{:#x}", ev.newAuthenticatorPubkey), + "old_offchain_signer_commitment": format!("{:#x}", ev.oldOffchainSignerCommitment), + "new_offchain_signer_commitment": format!("{:#x}", ev.newOffchainSignerCommitment), }) } pub fn serialize_authenticator_removed(ev: &AuthenticatorRemovedEvent) -> serde_json::Value { serde_json::json!({ - "pubkey_id": ev.pubkey_id, - "authenticator_address": format!("{:?}", ev.authenticator_address), - "authenticator_pubkey": format!("{:#x}", ev.authenticator_pubkey), - "old_offchain_signer_commitment": format!("{:#x}", ev.old_offchain_signer_commitment), - "new_offchain_signer_commitment": format!("{:#x}", ev.new_offchain_signer_commitment), + "pubkey_id": ev.pubkeyId, + "authenticator_address": format!("{:?}", ev.authenticatorAddress), + "authenticator_pubkey": format!("{:#x}", ev.authenticatorPubkey), + "old_offchain_signer_commitment": format!("{:#x}", ev.oldOffchainSignerCommitment), + "new_offchain_signer_commitment": format!("{:#x}", ev.newOffchainSignerCommitment), }) } pub fn serialize_account_recovered(ev: &AccountRecoveredEvent) -> serde_json::Value { serde_json::json!({ - "new_authenticator_address": format!("{:?}", ev.new_authenticator_address), - "new_authenticator_pubkey": format!("{:#x}", ev.new_authenticator_pubkey), - "old_offchain_signer_commitment": format!("{:#x}", ev.old_offchain_signer_commitment), - "new_offchain_signer_commitment": format!("{:#x}", ev.new_offchain_signer_commitment), + "new_authenticator_address": format!("{:?}", ev.newAuthenticatorAddress), + "new_authenticator_pubkey": format!("{:#x}", ev.newAuthenticatorPubkey), + "old_offchain_signer_commitment": format!("{:#x}", ev.oldOffchainSignerCommitment), + "new_offchain_signer_commitment": format!("{:#x}", ev.newOffchainSignerCommitment), }) } @@ -143,18 +143,6 @@ pub fn serialize_root_recorded(ev: &RootRecordedEvent) -> serde_json::Value { }) } -/// Serialize event data to JSON for storage -pub fn serialize_event_data(event: &RegistryEvent) -> serde_json::Value { - match event { - RegistryEvent::AccountCreated(ev) => serialize_account_created(ev), - RegistryEvent::AccountUpdated(ev) => serialize_account_updated(ev), - RegistryEvent::AuthenticatorInserted(ev) => serialize_authenticator_inserted(ev), - RegistryEvent::AuthenticatorRemoved(ev) => serialize_authenticator_removed(ev), - RegistryEvent::AccountRecovered(ev) => serialize_account_recovered(ev), - RegistryEvent::RootRecorded(ev) => serialize_root_recorded(ev), - } -} - pub fn deserialize_account_created( leaf_index: Option, event_data: &serde_json::Value, @@ -186,11 +174,11 @@ pub fn deserialize_account_created( .map_err(|_| invalid_field!("offchain_signer_commitment", "failed to parse U256"))?; Ok(AccountCreatedEvent { - leaf_index: leaf_index.ok_or_else(|| missing_field!("leaf_index"))?, - recovery_address, - authenticator_addresses, - authenticator_pubkeys, - offchain_signer_commitment, + leafIndex: leaf_index.ok_or_else(|| missing_field!("leaf_index"))?, + recoveryAddress: recovery_address, + authenticatorAddresses: authenticator_addresses, + authenticatorPubkeys: authenticator_pubkeys, + offchainSignerCommitment: offchain_signer_commitment, }) } @@ -233,13 +221,13 @@ pub fn deserialize_account_updated( .map_err(|_| invalid_field!("new_offchain_signer_commitment", "failed to parse U256"))?; Ok(AccountUpdatedEvent { - leaf_index: leaf_index.ok_or_else(|| missing_field!("leaf_index"))?, - pubkey_id, - new_authenticator_pubkey, - old_authenticator_address, - new_authenticator_address, - old_offchain_signer_commitment, - new_offchain_signer_commitment, + leafIndex: leaf_index.ok_or_else(|| missing_field!("leaf_index"))?, + pubkeyId: pubkey_id, + newAuthenticatorPubkey: new_authenticator_pubkey, + oldAuthenticatorAddress: old_authenticator_address, + newAuthenticatorAddress: new_authenticator_address, + oldOffchainSignerCommitment: old_offchain_signer_commitment, + newOffchainSignerCommitment: new_offchain_signer_commitment, }) } @@ -276,12 +264,12 @@ pub fn deserialize_authenticator_inserted( .map_err(|_| invalid_field!("new_offchain_signer_commitment", "failed to parse U256"))?; Ok(AuthenticatorInsertedEvent { - leaf_index: leaf_index.ok_or_else(|| missing_field!("leaf_index"))?, - pubkey_id, - authenticator_address, - new_authenticator_pubkey, - old_offchain_signer_commitment, - new_offchain_signer_commitment, + leafIndex: leaf_index.ok_or_else(|| missing_field!("leaf_index"))?, + pubkeyId: pubkey_id, + authenticatorAddress: authenticator_address, + newAuthenticatorPubkey: new_authenticator_pubkey, + oldOffchainSignerCommitment: old_offchain_signer_commitment, + newOffchainSignerCommitment: new_offchain_signer_commitment, }) } @@ -318,12 +306,12 @@ pub fn deserialize_authenticator_removed( .map_err(|_| invalid_field!("new_offchain_signer_commitment", "failed to parse U256"))?; Ok(AuthenticatorRemovedEvent { - leaf_index: leaf_index.ok_or_else(|| missing_field!("leaf_index"))?, - pubkey_id, - authenticator_address, - authenticator_pubkey, - old_offchain_signer_commitment, - new_offchain_signer_commitment, + leafIndex: leaf_index.ok_or_else(|| missing_field!("leaf_index"))?, + pubkeyId: pubkey_id, + authenticatorAddress: authenticator_address, + authenticatorPubkey: authenticator_pubkey, + oldOffchainSignerCommitment: old_offchain_signer_commitment, + newOffchainSignerCommitment: new_offchain_signer_commitment, }) } @@ -356,11 +344,11 @@ pub fn deserialize_account_recovered( .map_err(|_| invalid_field!("new_offchain_signer_commitment", "failed to parse U256"))?; Ok(AccountRecoveredEvent { - leaf_index: leaf_index.ok_or_else(|| missing_field!("leaf_index"))?, - new_authenticator_address, - new_authenticator_pubkey, - old_offchain_signer_commitment, - new_offchain_signer_commitment, + leafIndex: leaf_index.ok_or_else(|| missing_field!("leaf_index"))?, + newAuthenticatorAddress: new_authenticator_address, + newAuthenticatorPubkey: new_authenticator_pubkey, + oldOffchainSignerCommitment: old_offchain_signer_commitment, + newOffchainSignerCommitment: new_offchain_signer_commitment, }) } @@ -440,30 +428,43 @@ where /// Insert a full registry event #[instrument(level = "info", skip(self, event))] pub async fn insert_event(self, event: &BlockchainEvent) -> DBResult<()> { - let event_type = match &event.details { - RegistryEvent::AccountCreated(_) => WorldIdRegistryEventType::AccountCreated, - RegistryEvent::AccountUpdated(_) => WorldIdRegistryEventType::AccountUpdated, - RegistryEvent::AuthenticatorInserted(_) => { - WorldIdRegistryEventType::AuthenticatorInserted - } - RegistryEvent::AuthenticatorRemoved(_) => { - WorldIdRegistryEventType::AuthenticatorRemoved - } - RegistryEvent::AccountRecovered(_) => WorldIdRegistryEventType::AccountRecovered, - RegistryEvent::RootRecorded(_) => WorldIdRegistryEventType::RootRecorded, - }; - - let leaf_index = match &event.details { - RegistryEvent::AccountCreated(ev) => Some(ev.leaf_index as i64), - RegistryEvent::AccountUpdated(ev) => Some(ev.leaf_index as i64), - RegistryEvent::AuthenticatorInserted(ev) => Some(ev.leaf_index as i64), - RegistryEvent::AuthenticatorRemoved(ev) => Some(ev.leaf_index as i64), - RegistryEvent::AccountRecovered(ev) => Some(ev.leaf_index as i64), - RegistryEvent::RootRecorded(_) => None, + // Map the decoded event to its stored representation. Only the six + // indexed event types are ever fetched (see `RegistryEvent::signatures`); + // any other variant is rejected rather than silently stored. + let (event_type, leaf_index, event_data) = match &event.details { + RegistryEvent::AccountCreated(ev) => ( + WorldIdRegistryEventType::AccountCreated, + Some(ev.leafIndex as i64), + serialize_account_created(ev), + ), + RegistryEvent::AccountUpdated(ev) => ( + WorldIdRegistryEventType::AccountUpdated, + Some(ev.leafIndex as i64), + serialize_account_updated(ev), + ), + RegistryEvent::AuthenticatorInserted(ev) => ( + WorldIdRegistryEventType::AuthenticatorInserted, + Some(ev.leafIndex as i64), + serialize_authenticator_inserted(ev), + ), + RegistryEvent::AuthenticatorRemoved(ev) => ( + WorldIdRegistryEventType::AuthenticatorRemoved, + Some(ev.leafIndex as i64), + serialize_authenticator_removed(ev), + ), + RegistryEvent::AccountRecovered(ev) => ( + WorldIdRegistryEventType::AccountRecovered, + Some(ev.leafIndex as i64), + serialize_account_recovered(ev), + ), + RegistryEvent::RootRecorded(ev) => ( + WorldIdRegistryEventType::RootRecorded, + None, + serialize_root_recorded(ev), + ), + other => return Err(DBError::UnsupportedEvent(format!("{other:?}"))), }; - let event_data = serialize_event_data(&event.details); - sqlx::query( r#" INSERT INTO world_id_registry_events ( diff --git a/services/indexer/src/events_processor.rs b/services/indexer/src/events_processor.rs index 707dee1e2..4a6128189 100644 --- a/services/indexer/src/events_processor.rs +++ b/services/indexer/src/events_processor.rs @@ -20,11 +20,11 @@ impl EventsProcessor { .accounts() .await? .insert( - ev.leaf_index, - &ev.recovery_address, - &ev.authenticator_addresses, - &ev.authenticator_pubkeys, - &ev.offchain_signer_commitment, + ev.leafIndex, + &ev.recoveryAddress, + &ev.authenticatorAddresses, + &ev.authenticatorPubkeys, + &ev.offchainSignerCommitment, event.block_number, event.log_index, ) @@ -35,11 +35,11 @@ impl EventsProcessor { .accounts() .await? .update_authenticator_at_index( - ev.leaf_index, - ev.pubkey_id, - &ev.new_authenticator_address, - &ev.new_authenticator_pubkey, - &ev.new_offchain_signer_commitment, + ev.leafIndex, + ev.pubkeyId, + &ev.newAuthenticatorAddress, + &ev.newAuthenticatorPubkey, + &ev.newOffchainSignerCommitment, event.block_number, event.log_index, ) @@ -50,11 +50,11 @@ impl EventsProcessor { .accounts() .await? .insert_authenticator_at_index( - ev.leaf_index, - ev.pubkey_id, - &ev.authenticator_address, - &ev.new_authenticator_pubkey, - &ev.new_offchain_signer_commitment, + ev.leafIndex, + ev.pubkeyId, + &ev.authenticatorAddress, + &ev.newAuthenticatorPubkey, + &ev.newOffchainSignerCommitment, event.block_number, event.log_index, ) @@ -65,9 +65,9 @@ impl EventsProcessor { .accounts() .await? .remove_authenticator_at_index( - ev.leaf_index, - ev.pubkey_id, - &ev.new_offchain_signer_commitment, + ev.leafIndex, + ev.pubkeyId, + &ev.newOffchainSignerCommitment, event.block_number, event.log_index, ) @@ -78,16 +78,19 @@ impl EventsProcessor { .accounts() .await? .reset_authenticator( - ev.leaf_index, - &ev.new_authenticator_address, - &ev.new_authenticator_pubkey, - &ev.new_offchain_signer_commitment, + ev.leafIndex, + &ev.newAuthenticatorAddress, + &ev.newAuthenticatorPubkey, + &ev.newOffchainSignerCommitment, event.block_number, event.log_index, ) .await?; } RegistryEvent::RootRecorded(_ev) => {} + // Other registry events (ownership, fee, recovery-agent, etc.) are + // not indexed and are never fetched by the log filter. + _ => {} } Ok(()) diff --git a/services/indexer/src/lib.rs b/services/indexer/src/lib.rs index 6c947f4c9..40b8b5edc 100644 --- a/services/indexer/src/lib.rs +++ b/services/indexer/src/lib.rs @@ -27,6 +27,7 @@ pub mod metrics; pub mod rollback_executor; mod routes; mod sanity_check; +pub mod tip_cache; pub mod tree; static BLOCKCHAIN_RETRY_DELAY: Duration = Duration::from_secs(1); @@ -135,7 +136,6 @@ pub async unsafe fn run_indexer(cfg: GlobalConfig) -> eyre::Result<()> { run_indexer_only( db, http_provider, - &cfg.ws_rpc_url, cfg.registry_address, indexer_config, tree_state, @@ -169,7 +169,6 @@ pub async unsafe fn run_indexer(cfg: GlobalConfig) -> eyre::Result<()> { run_both( db, http_provider, - &cfg.ws_rpc_url, cfg.registry_address, indexer_config, http_config, @@ -184,14 +183,12 @@ pub async unsafe fn run_indexer(cfg: GlobalConfig) -> eyre::Result<()> { async fn run_indexer_only( db: DB, http_provider: DynProvider, - ws_rpc_url: &str, registry_address: Address, indexer_cfg: IndexerConfig, tree_state: tree::TreeState, ) -> eyre::Result<()> { process_registry_events( http_provider, - ws_rpc_url, registry_address, indexer_cfg, &db, @@ -273,7 +270,6 @@ async fn run_http_only( async fn run_both( db: DB, http_provider: DynProvider, - ws_rpc_url: &str, registry_address: Address, indexer_cfg: IndexerConfig, http_cfg: HttpConfig, @@ -320,7 +316,6 @@ async fn run_both( tokio::select! { result = process_registry_events( http_provider, - ws_rpc_url, registry_address, indexer_cfg, &db, @@ -348,38 +343,36 @@ pub async fn handle_registry_event<'a>( Ok(()) } -/// Stream registry events from the blockchain and process them. -/// Restart when websocket connection is dropped. +/// Poll registry events from the blockchain and process them. +/// Restart when the poll stream returns an error (e.g. an RPC failure). #[instrument(level = "info", skip_all, fields(start_from))] pub async fn process_registry_events( http_provider: DynProvider, - ws_rpc_url: &str, registry_address: Address, indexer_cfg: IndexerConfig, db: &DB, tree_state: &tree::TreeState, ) -> IndexerResult<()> { - // We re-create the blockchain connection (including backfill and websocket) when the stream - // returns an error or the websocket connection is dropped. + let poll_interval = Duration::from_secs(indexer_cfg.poll_interval_secs); + + // We re-create the blockchain connection and poll stream when the stream + // returns an error. loop { tracing::info!("starting blockchain connection"); - let blockchain = - match Blockchain::new(http_provider.clone(), ws_rpc_url, registry_address).await { - Ok(b) => b, - Err(e) => { - tracing::error!(?e, "failed to create blockchain connection, retrying"); - tokio::time::sleep(BLOCKCHAIN_RETRY_DELAY).await; - continue; - } - }; + let blockchain = Blockchain::new(http_provider.clone(), registry_address); let from = match db.world_id_registry_events().get_latest_block().await? { Some(block) => block + 1, None => indexer_cfg.start_block, }; - let mut stream = blockchain.backfill_and_stream_events(from, indexer_cfg.batch_size); + let mut stream = blockchain.poll_events( + from, + indexer_cfg.batch_size, + indexer_cfg.confirmations, + poll_interval, + ); let versioned_tree = tree::VersionedTreeState::new(tree_state.clone(), indexer_cfg.tree_max_block_age); @@ -438,5 +431,6 @@ pub async fn process_registry_events( } tracing::warn!("restarting blockchain connection"); + tokio::time::sleep(BLOCKCHAIN_RETRY_DELAY).await; } } diff --git a/services/indexer/src/rollback_executor.rs b/services/indexer/src/rollback_executor.rs index e1545f403..b0ae38e62 100644 --- a/services/indexer/src/rollback_executor.rs +++ b/services/indexer/src/rollback_executor.rs @@ -7,7 +7,7 @@ use alloy::{ use world_id_registries::world_id::WorldIdRegistry; use crate::{ - blockchain::RegistryEvent, + blockchain::{RegistryEvent, RegistryEventExt}, db::{DB, DBResult, IsolationLevel, PostgresDBTransaction, WorldIdRegistryEventId}, error::{IndexerError, IndexerResult}, events_processor::EventsProcessor, diff --git a/services/indexer/src/tip_cache.rs b/services/indexer/src/tip_cache.rs new file mode 100644 index 000000000..fc28b5fda --- /dev/null +++ b/services/indexer/src/tip_cache.rs @@ -0,0 +1,281 @@ +//! In-memory tip cache for the indexer. +//! +//! The indexer polls registry events at the chain **tip** (no confirmation lag) +//! and applies them immediately to the in-memory Merkle tree, so inclusion +//! proofs reflect the latest state. Those updates are held in a **pending +//! buffer** and are only persisted to the database once their block is at least +//! `confirmations` deep — that is the durable, "confirmed" store. +//! +//! Because nothing unconfirmed is written to the database, a reorg inside the +//! unconfirmed window is cheap to recover from: the in-memory tree is rolled +//! back to the common ancestor (via [`VersionedTreeState::rollback_to`]) and the +//! canonical events are replayed. The database is never touched. +//! +//! Reorg *detection* is driven by the poll loop (which has RPC access) using the +//! per-block canonical hashes tracked here; see [`TipCache::block_hashes_desc`]. + +use std::collections::VecDeque; + +use alloy::primitives::U256; + +use crate::{ + blockchain::{BlockchainEvent, RegistryEvent}, + db::{DB, IsolationLevel, WorldIdRegistryEventId}, + error::{IndexerError, IndexerResult}, + events_processor::EventsProcessor, + tree::{VersionedTreeState, apply_event_to_tree}, +}; + +/// Holds the tip view of the tree plus the buffer of applied-but-unconfirmed +/// events awaiting database commit. +pub struct TipCache { + /// The in-memory tree, kept at the tip. Shared with the HTTP server, so + /// proofs served reflect unconfirmed updates held here. + tree: VersionedTreeState, + /// Applied-but-uncommitted events, ordered ascending by `(block, log_index)`. + pending: VecDeque>, + /// Canonical block hash of each event-bearing pending block, ascending by + /// block number (one entry per block). Used for reorg detection. + block_hashes: VecDeque<(u64, U256)>, + /// Number of confirmations a block must reach before its events are + /// committed to the database. + confirmations: u64, + /// `(block, log_index)` of the most recent event committed to the database. + committed: WorldIdRegistryEventId, +} + +impl TipCache { + pub fn new( + tree: VersionedTreeState, + confirmations: u64, + committed: WorldIdRegistryEventId, + ) -> Self { + Self { + tree, + pending: VecDeque::new(), + block_hashes: VecDeque::new(), + confirmations, + committed, + } + } + + /// The in-memory tree (tip view). + pub fn tree(&self) -> &VersionedTreeState { + &self.tree + } + + /// Highest block currently held in the pending buffer, if any. + pub fn highest_pending_block(&self) -> Option { + self.block_hashes.back().map(|(b, _)| *b) + } + + /// `(block, canonical_hash)` for every event-bearing pending block, newest + /// first. The poll loop walks this to locate the common ancestor on a reorg. + pub fn block_hashes_desc(&self) -> Vec<(u64, U256)> { + self.block_hashes.iter().rev().copied().collect() + } + + /// Apply a single canonical event to the tip tree and record it as pending. + /// + /// On a `RootRecorded` event the resulting tree root is checked against the + /// on-chain root; a mismatch indicates a missed/extra event (or a reorg + /// observed mid-fetch) and is surfaced as [`IndexerError::ReorgDetected`]. + pub async fn apply(&mut self, event: BlockchainEvent) -> IndexerResult<()> { + apply_event_to_tree(&self.tree, &event).await?; + + // Track one canonical hash per block for reorg detection. + match self.block_hashes.back() { + Some((b, _)) if *b == event.block_number => {} + _ => self + .block_hashes + .push_back((event.block_number, event.block_hash)), + } + + let block_number = event.block_number; + let expected_root = match &event.details { + RegistryEvent::RootRecorded(rr) => Some(rr.root), + _ => None, + }; + + self.pending.push_back(event); + crate::metrics::set_chain_processed_block(block_number); + + if let Some(expected) = expected_root { + let actual = self.tree.root().await; + if actual != expected { + return Err(IndexerError::ReorgDetected { + block_number, + reason: format!( + "tip tree root (0x{actual:x}) does not match RootRecorded root (0x{expected:x})" + ), + }); + } + } + + Ok(()) + } + + /// Roll the tip tree and pending buffer back so that no event from a block + /// after `block` remains applied. Used to discard a reorged suffix. + pub async fn rollback_after_block(&mut self, block: u64) -> IndexerResult<()> { + // `log_index: u64::MAX` keeps every event in `block` and discards + // everything in later blocks. + let target = WorldIdRegistryEventId { + block_number: block, + log_index: u64::MAX, + }; + self.tree.rollback_to(target).await?; + + while self.pending.back().is_some_and(|e| e.block_number > block) { + self.pending.pop_back(); + } + while self.block_hashes.back().is_some_and(|(b, _)| *b > block) { + self.block_hashes.pop_back(); + } + + Ok(()) + } + + /// Commit the contiguous prefix of pending events whose block is at or below + /// `head - confirmations` to the database, then drop them from the buffer. + /// + /// The tree is **not** touched here — those updates were already applied at + /// the tip; this only makes them durable. + pub async fn flush_confirmed(&mut self, db: &DB, head: u64) -> IndexerResult<()> { + let confirmed_through = head.saturating_sub(self.confirmations); + + if self + .pending + .front() + .is_none_or(|e| e.block_number > confirmed_through) + { + return Ok(()); + } + + let mut tx = db.transaction(IsolationLevel::Serializable).await?; + let mut last = self.committed; + let mut flushed = 0usize; + + while let Some(front) = self.pending.front() { + if front.block_number > confirmed_through { + break; + } + let event = self.pending.pop_front().expect("front exists"); + tx.world_id_registry_events() + .await? + .insert_event(&event) + .await?; + EventsProcessor::process_event(&mut tx, &event).await?; + last = WorldIdRegistryEventId { + block_number: event.block_number, + log_index: event.log_index, + }; + flushed += 1; + } + + tx.commit().await?; + + self.committed = last; + while self + .block_hashes + .front() + .is_some_and(|(b, _)| *b <= confirmed_through) + { + self.block_hashes.pop_front(); + } + self.tree.set_last_synced_event_id(last).await; + + tracing::debug!( + flushed, + confirmed_through, + committed = ?self.committed, + "flushed confirmed events to DB" + ); + + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::tree::TreeState; + + fn tmp_file() -> std::path::PathBuf { + let mut p = std::env::temp_dir(); + p.push(format!("tip_cache_test_{}.tmp", uuid::Uuid::new_v4())); + p + } + + fn cache(confirmations: u64) -> TipCache { + let tree = unsafe { TreeState::new_empty(6, tmp_file()).unwrap() }; + let versioned = VersionedTreeState::new(tree, 10_000); + TipCache::new(versioned, confirmations, WorldIdRegistryEventId::default()) + } + + fn account_created(block: u64, log_index: u64, leaf: u64, commitment: u64) -> BlockchainEvent { + BlockchainEvent { + block_number: block, + block_hash: U256::from(block), // distinct per block by default + tx_hash: U256::from(1), + log_index, + details: RegistryEvent::AccountCreated( + world_id_registries::world_id::WorldIdRegistry::AccountCreated { + leafIndex: leaf, + recoveryAddress: alloy::primitives::Address::ZERO, + authenticatorAddresses: vec![], + authenticatorPubkeys: vec![], + offchainSignerCommitment: U256::from(commitment), + }, + ), + } + } + + #[tokio::test] + async fn apply_updates_tree_and_tracks_block_hash() { + let mut c = cache(2); + let root0 = c.tree().root().await; + + c.apply(account_created(10, 0, 1, 42)).await.unwrap(); + + assert_ne!(c.tree().root().await, root0, "tip tree should reflect update"); + assert_eq!(c.highest_pending_block(), Some(10)); + } + + #[tokio::test] + async fn one_block_hash_entry_per_block() { + let mut c = cache(2); + c.apply(account_created(10, 0, 1, 42)).await.unwrap(); + c.apply(account_created(10, 1, 2, 43)).await.unwrap(); + c.apply(account_created(11, 0, 3, 44)).await.unwrap(); + + let hashes = c.block_hashes_desc(); + assert_eq!(hashes.iter().map(|(b, _)| *b).collect::>(), vec![11, 10]); + } + + #[tokio::test] + async fn rollback_after_block_restores_tip() { + let mut c = cache(2); + c.apply(account_created(10, 0, 1, 42)).await.unwrap(); + let root_at_10 = c.tree().root().await; + c.apply(account_created(11, 0, 2, 43)).await.unwrap(); + assert_ne!(c.tree().root().await, root_at_10); + + c.rollback_after_block(10).await.unwrap(); + + assert_eq!(c.tree().root().await, root_at_10, "tree rolled back to block 10"); + assert_eq!(c.highest_pending_block(), Some(10)); + } + + #[tokio::test] + async fn rollback_clears_all_when_before_first() { + let mut c = cache(2); + let root_empty = c.tree().root().await; + c.apply(account_created(10, 0, 1, 42)).await.unwrap(); + + c.rollback_after_block(9).await.unwrap(); + + assert_eq!(c.tree().root().await, root_empty); + assert_eq!(c.highest_pending_block(), None); + } +} diff --git a/services/indexer/src/tree/mod.rs b/services/indexer/src/tree/mod.rs index a660f2e06..1154f6934 100644 --- a/services/indexer/src/tree/mod.rs +++ b/services/indexer/src/tree/mod.rs @@ -90,20 +90,17 @@ pub fn extract_leaf_commitment(event: &crate::blockchain::RegistryEvent) -> Opti use crate::blockchain::RegistryEvent; match event { - RegistryEvent::AccountCreated(ev) => Some((ev.leaf_index, ev.offchain_signer_commitment)), - RegistryEvent::AccountUpdated(ev) => { - Some((ev.leaf_index, ev.new_offchain_signer_commitment)) - } + RegistryEvent::AccountCreated(ev) => Some((ev.leafIndex, ev.offchainSignerCommitment)), + RegistryEvent::AccountUpdated(ev) => Some((ev.leafIndex, ev.newOffchainSignerCommitment)), RegistryEvent::AuthenticatorInserted(ev) => { - Some((ev.leaf_index, ev.new_offchain_signer_commitment)) + Some((ev.leafIndex, ev.newOffchainSignerCommitment)) } RegistryEvent::AuthenticatorRemoved(ev) => { - Some((ev.leaf_index, ev.new_offchain_signer_commitment)) - } - RegistryEvent::AccountRecovered(ev) => { - Some((ev.leaf_index, ev.new_offchain_signer_commitment)) + Some((ev.leafIndex, ev.newOffchainSignerCommitment)) } - RegistryEvent::RootRecorded(_) => None, + RegistryEvent::AccountRecovered(ev) => Some((ev.leafIndex, ev.newOffchainSignerCommitment)), + // RootRecorded and all non-indexed events carry no leaf data. + _ => None, } } diff --git a/services/indexer/tests/db_tests.rs b/services/indexer/tests/db_tests.rs index 96e0f7161..8fd3e9daf 100644 --- a/services/indexer/tests/db_tests.rs +++ b/services/indexer/tests/db_tests.rs @@ -451,11 +451,11 @@ async fn test_insert_world_tree_event() { log_index, details: world_id_indexer::blockchain::RegistryEvent::AccountCreated( world_id_indexer::blockchain::AccountCreatedEvent { - leaf_index, - recovery_address: Address::ZERO, - authenticator_addresses: vec![], - authenticator_pubkeys: vec![], - offchain_signer_commitment: commitment, + leafIndex: leaf_index, + recoveryAddress: Address::ZERO, + authenticatorAddresses: vec![], + authenticatorPubkeys: vec![], + offchainSignerCommitment: commitment, }, ), }; @@ -499,11 +499,11 @@ async fn test_duplicate_world_tree_event_insert() { log_index, details: world_id_indexer::blockchain::RegistryEvent::AccountCreated( world_id_indexer::blockchain::AccountCreatedEvent { - leaf_index, - recovery_address: Address::ZERO, - authenticator_addresses: vec![], - authenticator_pubkeys: vec![], - offchain_signer_commitment: commitment, + leafIndex: leaf_index, + recoveryAddress: Address::ZERO, + authenticatorAddresses: vec![], + authenticatorPubkeys: vec![], + offchainSignerCommitment: commitment, }, ), }; @@ -547,11 +547,11 @@ async fn test_get_world_tree_event() { log_index, details: world_id_indexer::blockchain::RegistryEvent::AccountCreated( world_id_indexer::blockchain::AccountCreatedEvent { - leaf_index, - recovery_address: Address::ZERO, - authenticator_addresses: vec![], - authenticator_pubkeys: vec![], - offchain_signer_commitment: commitment, + leafIndex: leaf_index, + recoveryAddress: Address::ZERO, + authenticatorAddresses: vec![], + authenticatorPubkeys: vec![], + offchainSignerCommitment: commitment, }, ), }; @@ -590,11 +590,11 @@ async fn test_get_latest_world_tree_events() { log_index: i, details: world_id_indexer::blockchain::RegistryEvent::AccountCreated( world_id_indexer::blockchain::AccountCreatedEvent { - leaf_index: i, - recovery_address: Address::ZERO, - authenticator_addresses: vec![], - authenticator_pubkeys: vec![], - offchain_signer_commitment: U256::from(i * 100), + leafIndex: i, + recoveryAddress: Address::ZERO, + authenticatorAddresses: vec![], + authenticatorPubkeys: vec![], + offchainSignerCommitment: U256::from(i * 100), }, ), }; @@ -794,11 +794,11 @@ async fn test_multiple_operations_in_transaction() { log_index: 0, details: world_id_indexer::blockchain::RegistryEvent::AccountCreated( world_id_indexer::blockchain::AccountCreatedEvent { - leaf_index: 1, - recovery_address: Address::ZERO, - authenticator_addresses: vec![], - authenticator_pubkeys: vec![], - offchain_signer_commitment: U256::from(123), + leafIndex: 1, + recoveryAddress: Address::ZERO, + authenticatorAddresses: vec![], + authenticatorPubkeys: vec![], + offchainSignerCommitment: U256::from(123), }, ), }; diff --git a/services/indexer/tests/edge_case_tests.rs b/services/indexer/tests/edge_case_tests.rs index 67a45c619..1974626a2 100644 --- a/services/indexer/tests/edge_case_tests.rs +++ b/services/indexer/tests/edge_case_tests.rs @@ -93,11 +93,11 @@ async fn test_max_block_number() { log_index: 0, details: world_id_indexer::blockchain::RegistryEvent::AccountCreated( world_id_indexer::blockchain::AccountCreatedEvent { - leaf_index: 1, - recovery_address: Address::ZERO, - authenticator_addresses: vec![], - authenticator_pubkeys: vec![], - offchain_signer_commitment: U256::from(100), + leafIndex: 1, + recoveryAddress: Address::ZERO, + authenticatorAddresses: vec![], + authenticatorPubkeys: vec![], + offchainSignerCommitment: U256::from(100), }, ), }; @@ -130,11 +130,11 @@ async fn test_max_log_index() { log_index: max_log_index, details: world_id_indexer::blockchain::RegistryEvent::AccountCreated( world_id_indexer::blockchain::AccountCreatedEvent { - leaf_index: 1, - recovery_address: Address::ZERO, - authenticator_addresses: vec![], - authenticator_pubkeys: vec![], - offchain_signer_commitment: U256::from(100), + leafIndex: 1, + recoveryAddress: Address::ZERO, + authenticatorAddresses: vec![], + authenticatorPubkeys: vec![], + offchainSignerCommitment: U256::from(100), }, ), }; @@ -197,59 +197,59 @@ async fn test_all_event_types() { WorldIdRegistryEventType::AccountCreated => { world_id_indexer::blockchain::RegistryEvent::AccountCreated( world_id_indexer::blockchain::AccountCreatedEvent { - leaf_index: i as u64, - recovery_address: Address::ZERO, - authenticator_addresses: vec![], - authenticator_pubkeys: vec![], - offchain_signer_commitment: U256::from(i * 100), + leafIndex: i as u64, + recoveryAddress: Address::ZERO, + authenticatorAddresses: vec![], + authenticatorPubkeys: vec![], + offchainSignerCommitment: U256::from(i * 100), }, ) } WorldIdRegistryEventType::AccountUpdated => { world_id_indexer::blockchain::RegistryEvent::AccountUpdated( world_id_indexer::blockchain::AccountUpdatedEvent { - leaf_index: i as u64, - pubkey_id: 0, - new_authenticator_pubkey: U256::from(i * 100), - old_authenticator_address: Address::ZERO, - new_authenticator_address: Address::ZERO, - old_offchain_signer_commitment: U256::ZERO, - new_offchain_signer_commitment: U256::from(i * 100), + leafIndex: i as u64, + pubkeyId: 0, + newAuthenticatorPubkey: U256::from(i * 100), + oldAuthenticatorAddress: Address::ZERO, + newAuthenticatorAddress: Address::ZERO, + oldOffchainSignerCommitment: U256::ZERO, + newOffchainSignerCommitment: U256::from(i * 100), }, ) } WorldIdRegistryEventType::AuthenticatorInserted => { world_id_indexer::blockchain::RegistryEvent::AuthenticatorInserted( world_id_indexer::blockchain::AuthenticatorInsertedEvent { - leaf_index: i as u64, - pubkey_id: 0, - authenticator_address: Address::ZERO, - new_authenticator_pubkey: U256::from(i * 100), - old_offchain_signer_commitment: U256::ZERO, - new_offchain_signer_commitment: U256::from(i * 100), + leafIndex: i as u64, + pubkeyId: 0, + authenticatorAddress: Address::ZERO, + newAuthenticatorPubkey: U256::from(i * 100), + oldOffchainSignerCommitment: U256::ZERO, + newOffchainSignerCommitment: U256::from(i * 100), }, ) } WorldIdRegistryEventType::AuthenticatorRemoved => { world_id_indexer::blockchain::RegistryEvent::AuthenticatorRemoved( world_id_indexer::blockchain::AuthenticatorRemovedEvent { - leaf_index: i as u64, - pubkey_id: 0, - authenticator_address: Address::ZERO, - authenticator_pubkey: U256::from(i * 100), - old_offchain_signer_commitment: U256::ZERO, - new_offchain_signer_commitment: U256::from(i * 100), + leafIndex: i as u64, + pubkeyId: 0, + authenticatorAddress: Address::ZERO, + authenticatorPubkey: U256::from(i * 100), + oldOffchainSignerCommitment: U256::ZERO, + newOffchainSignerCommitment: U256::from(i * 100), }, ) } WorldIdRegistryEventType::AccountRecovered => { world_id_indexer::blockchain::RegistryEvent::AccountRecovered( world_id_indexer::blockchain::AccountRecoveredEvent { - leaf_index: i as u64, - new_authenticator_address: Address::ZERO, - new_authenticator_pubkey: U256::from(i * 100), - old_offchain_signer_commitment: U256::ZERO, - new_offchain_signer_commitment: U256::from(i * 100), + leafIndex: i as u64, + newAuthenticatorAddress: Address::ZERO, + newAuthenticatorPubkey: U256::from(i * 100), + oldOffchainSignerCommitment: U256::ZERO, + newOffchainSignerCommitment: U256::from(i * 100), }, ) } diff --git a/services/indexer/tests/helpers/common.rs b/services/indexer/tests/helpers/common.rs index c4737fa6e..423563210 100644 --- a/services/indexer/tests/helpers/common.rs +++ b/services/indexer/tests/helpers/common.rs @@ -57,10 +57,6 @@ impl TestSetup { self._anvil.endpoint().to_string() } - pub fn ws_url(&self) -> String { - self._anvil.ws_endpoint().to_string() - } - pub async fn create_account(&self, auth_addr: Address, pubkey: U256, commitment: u64) { let deployer = self._anvil.signer(0).unwrap(); let registry = WorldIdRegistry::new( diff --git a/services/indexer/tests/helpers/mock_blockchain.rs b/services/indexer/tests/helpers/mock_blockchain.rs index 74ad9d09a..7cee747cf 100644 --- a/services/indexer/tests/helpers/mock_blockchain.rs +++ b/services/indexer/tests/helpers/mock_blockchain.rs @@ -38,11 +38,11 @@ pub fn mock_account_created_event_with_authenticators( tx_hash: U256::from(1), log_index, details: RegistryEvent::AccountCreated(world_id_indexer::blockchain::AccountCreatedEvent { - leaf_index, - recovery_address, - authenticator_addresses, - authenticator_pubkeys, - offchain_signer_commitment: commitment, + leafIndex: leaf_index, + recoveryAddress: recovery_address, + authenticatorAddresses: authenticator_addresses, + authenticatorPubkeys: authenticator_pubkeys, + offchainSignerCommitment: commitment, }), } } @@ -65,13 +65,13 @@ pub fn mock_account_updated_event( tx_hash: U256::from(1), log_index, details: RegistryEvent::AccountUpdated(world_id_indexer::blockchain::AccountUpdatedEvent { - leaf_index, - pubkey_id, - new_authenticator_pubkey: new_pubkey, - old_authenticator_address: Address::ZERO, - new_authenticator_address: new_address, - old_offchain_signer_commitment: old_commitment, - new_offchain_signer_commitment: new_commitment, + leafIndex: leaf_index, + pubkeyId: pubkey_id, + newAuthenticatorPubkey: new_pubkey, + oldAuthenticatorAddress: Address::ZERO, + newAuthenticatorAddress: new_address, + oldOffchainSignerCommitment: old_commitment, + newOffchainSignerCommitment: new_commitment, }), } } @@ -95,12 +95,12 @@ pub fn mock_authenticator_inserted_event( log_index, details: RegistryEvent::AuthenticatorInserted( world_id_indexer::blockchain::AuthenticatorInsertedEvent { - leaf_index, - pubkey_id, - authenticator_address, - new_authenticator_pubkey: new_pubkey, - old_offchain_signer_commitment: old_commitment, - new_offchain_signer_commitment: new_commitment, + leafIndex: leaf_index, + pubkeyId: pubkey_id, + authenticatorAddress: authenticator_address, + newAuthenticatorPubkey: new_pubkey, + oldOffchainSignerCommitment: old_commitment, + newOffchainSignerCommitment: new_commitment, }, ), } @@ -125,12 +125,12 @@ pub fn mock_authenticator_removed_event( log_index, details: RegistryEvent::AuthenticatorRemoved( world_id_indexer::blockchain::AuthenticatorRemovedEvent { - leaf_index, - pubkey_id, - authenticator_address, - authenticator_pubkey: pubkey, - old_offchain_signer_commitment: old_commitment, - new_offchain_signer_commitment: new_commitment, + leafIndex: leaf_index, + pubkeyId: pubkey_id, + authenticatorAddress: authenticator_address, + authenticatorPubkey: pubkey, + oldOffchainSignerCommitment: old_commitment, + newOffchainSignerCommitment: new_commitment, }, ), } @@ -153,11 +153,11 @@ pub fn mock_account_recovered_event( log_index, details: RegistryEvent::AccountRecovered( world_id_indexer::blockchain::AccountRecoveredEvent { - leaf_index, - new_authenticator_address: new_address, - new_authenticator_pubkey: new_pubkey, - old_offchain_signer_commitment: old_commitment, - new_offchain_signer_commitment: new_commitment, + leafIndex: leaf_index, + newAuthenticatorAddress: new_address, + newAuthenticatorPubkey: new_pubkey, + oldOffchainSignerCommitment: old_commitment, + newOffchainSignerCommitment: new_commitment, }, ), } diff --git a/services/indexer/tests/test_blockchain_streams.rs b/services/indexer/tests/test_blockchain_streams.rs index 0b66fd879..dd4e38690 100644 --- a/services/indexer/tests/test_blockchain_streams.rs +++ b/services/indexer/tests/test_blockchain_streams.rs @@ -1,6 +1,6 @@ #![cfg(feature = "integration-tests")] -use std::{sync::atomic::Ordering, time::Duration}; +use std::time::Duration; use alloy::{ network::EthereumWallet, @@ -11,13 +11,20 @@ use alloy::{ }; use eddsa_babyjubjub::EdDSAPrivateKey; use futures_util::{StreamExt, TryStreamExt}; -use world_id_indexer::blockchain::{Blockchain, BlockchainError, BlockchainEvent, RegistryEvent}; +use world_id_indexer::blockchain::{ + Blockchain, BlockchainError, BlockchainEvent, RegistryEvent, RegistryEventExt, +}; use world_id_registries::world_id::WorldIdRegistry; use world_id_services_common::ProviderArgs; use world_id_test_utils::anvil::TestAnvil; const RECOVERY_ADDRESS: Address = address!("0x0000000000000000000000000000000000000001"); +/// Anvil mines instantly on a single node, so there are no reorgs and we can +/// poll right up to the chain head. +const CONFIRMATIONS: u64 = 0; +const POLL_INTERVAL: Duration = Duration::from_millis(200); + fn random_pubkey() -> U256 { let sk = EdDSAPrivateKey::random(&mut rand::thread_rng()); U256::from_le_slice(&sk.public().to_compressed_bytes().unwrap()) @@ -58,23 +65,20 @@ async fn create_accounts( start_index + count } -/// Tests the full `stream_world_tree_events` pipeline that combines backfill, -/// gap-fill, and live WebSocket streaming into a single ordered stream. +/// Tests the full `poll_events` pipeline that backfills historical logs and then +/// keeps polling for newly confirmed logs. /// /// Scenario: /// 1. **Backfill**: 100 accounts are created before the stream starts. The -/// stream is polled until all backfill events are consumed, which -/// deterministically drives the backfill to completion and commits -/// `last_block`. -/// 2. **Gap**: 30 accounts are created *after* the backfill finishes. They -/// land on-chain in blocks after `last_block`, so the WS stage's gap-fill -/// must fetch them via HTTP. -/// 3. **Live**: A spawned task creates 5 accounts after a delay so the WS -/// subscription is already established. These arrive as live WS events. +/// first poll emits all of their logs in one batched range. +/// 2. **Gap**: 30 accounts are created *after* the backfill is consumed. They +/// land on-chain in later blocks and are picked up by a subsequent poll. +/// 3. **Live**: A spawned task creates 5 more accounts after a delay so they +/// arrive on an even later poll. /// 4. **Verification**: All collected events are compared against a ground /// truth `get_logs` query, decoded through `RegistryEvent::decode`. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn test_stream_world_tree_events() { +async fn test_poll_events() { let _ = tracing_subscriber::fmt() .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) .try_init(); @@ -93,9 +97,7 @@ async fn test_stream_world_tree_events() { .await .expect("failed to build provider"); - let blockchain = Blockchain::new(provider.clone(), anvil.ws_endpoint(), registry_address) - .await - .expect("failed to create Blockchain"); + let blockchain = Blockchain::new(provider.clone(), registry_address); let http_provider = ProviderBuilder::new().connect_http(anvil.endpoint().parse::().unwrap()); @@ -133,7 +135,7 @@ async fn test_stream_world_tree_events() { assert!(backfill_log_count > 0, "expected some backfill logs"); let logs_per_account = backfill_log_count as u64 / backfill_count; - let mut stream = blockchain.backfill_and_stream_events(from_block + 1, 2); + let mut stream = blockchain.poll_events(from_block + 1, 2, CONFIRMATIONS, POLL_INTERVAL); // Consume all backfill events let mut stream_events: Vec> = tokio::time::timeout( @@ -206,17 +208,15 @@ async fn test_stream_world_tree_events() { ); } -/// Tests `backfill_events` in isolation (no WebSocket phase). +/// Tests that `poll_events` backfills a pre-existing range of logs correctly. /// /// Scenario: /// 1. 10 accounts are created before the stream starts. -/// 2. `backfill_events` is called, returning a stream and a shared atomic -/// tracking the last fetched block. -/// 3. The stream is fully consumed and compared against a ground truth -/// `get_logs` query. -/// 4. The `last_block` atomic is verified to be at or past the chain head. +/// 2. `poll_events` is polled until all backfill logs are consumed. +/// 3. The collected events are compared against a ground truth `get_logs` +/// query. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn test_backfill_events() { +async fn test_poll_events_backfill() { let _ = tracing_subscriber::fmt() .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) .try_init(); @@ -235,9 +235,7 @@ async fn test_backfill_events() { .await .expect("failed to build provider"); - let blockchain = Blockchain::new(provider.clone(), anvil.ws_endpoint(), registry_address) - .await - .expect("failed to create Blockchain"); + let blockchain = Blockchain::new(provider.clone(), registry_address); let http_provider = ProviderBuilder::new().connect_http(anvil.endpoint().parse::().unwrap()); @@ -259,15 +257,6 @@ async fn test_backfill_events() { ) .await; - // --- Backfill --- - let (stream, last_block) = blockchain.backfill_events(from_block + 1, 2); - - let stream_events: Vec> = - tokio::time::timeout(Duration::from_secs(30), stream.try_collect()) - .await - .expect("timed out waiting for backfill events") - .expect("backfill stream error"); - // --- Ground truth --- let expected_logs = http_provider .get_logs( @@ -284,21 +273,25 @@ async fn test_backfill_events() { .map(|log| RegistryEvent::decode(log).expect("failed to decode ground truth log")) .collect(); + // --- Backfill via polling --- + let mut stream = blockchain.poll_events(from_block + 1, 2, CONFIRMATIONS, POLL_INTERVAL); + + let stream_events: Vec> = tokio::time::timeout( + Duration::from_secs(30), + stream.by_ref().take(expected_events.len()).try_collect(), + ) + .await + .expect("timed out waiting for backfill events") + .expect("backfill stream error"); + assert_eq!( stream_events, expected_events, "backfill stream events do not match ground truth" ); - - // Verify last_block was updated to at least the chain head at creation time - let last = last_block.load(Ordering::Relaxed); - assert!( - last >= from_block, - "last_block ({last}) should be >= from_block ({from_block})" - ); } -/// Tests that the stream produced by `backfill_and_stream_events` yields an -/// `Err` and then terminates when an RPC error occurs during backfill. +/// Tests that the stream produced by `poll_events` yields an `Err` and then +/// terminates when an RPC error occurs. /// /// Scenario: /// 1. 10 accounts are created so there is backfill work to do. @@ -328,9 +321,7 @@ async fn test_stream_stops_on_error() { .await .expect("failed to build provider"); - let blockchain = Blockchain::new(provider.clone(), anvil.ws_endpoint(), registry_address) - .await - .expect("failed to create Blockchain"); + let blockchain = Blockchain::new(provider.clone(), registry_address); let http_provider = ProviderBuilder::new().connect_http(anvil.endpoint().parse::().unwrap()); @@ -352,7 +343,7 @@ async fn test_stream_stops_on_error() { ) .await; - let mut stream = blockchain.backfill_and_stream_events(from_block + 1, 2); + let mut stream = blockchain.poll_events(from_block + 1, 2, CONFIRMATIONS, POLL_INTERVAL); // Consume a small number of events so the backfill is partway through. let _partial: Vec> = tokio::time::timeout( @@ -379,19 +370,10 @@ async fn test_stream_stops_on_error() { .await .expect("timed out waiting for error after anvil kill"); - // After killing Anvil the stream must stop with a connectivity error. - // Which variant surfaces depends on whether the remaining backfill events - // were already buffered before the kill (→ WsSubscriptionClosed, because - // the stream transitioned to the WS phase before the HTTP client had a - // chance to fail) or whether the HTTP client sees the dead process first - // (→ Rpc). Both are correct outcomes: the important invariant is that the - // stream yields exactly one Err and then terminates. + // After killing Anvil the stream must stop with an RPC error. assert!( - matches!( - error_event, - Some(Err(BlockchainError::Rpc(_))) | Some(Err(BlockchainError::WsSubscriptionClosed)) - ), - "expected a connectivity error after killing anvil, got {error_event:?}" + matches!(error_event, Some(Err(BlockchainError::Rpc(_)))), + "expected an RPC error after killing anvil, got {error_event:?}" ); // Stream must terminate immediately after the first error. @@ -404,113 +386,3 @@ async fn test_stream_stops_on_error() { "expected None immediately after first error, got {terminated:?}" ); } - -/// Tests that the stream terminates with a `WsSubscriptionClosed` error when -/// the WebSocket connection is dropped. -/// -/// Scenario: -/// 1. A small backfill (10 accounts) is created and fully consumed. -/// 2. The stream transitions to the WebSocket phase, subscribes, and waits -/// for the first live event. -/// 3. A spawned task kills the Anvil process after a short delay (allowing -/// the WS subscription to be established first). -/// 4. The WS subscription returns `None`, which surfaces as -/// `BlockchainError::WsSubscriptionClosed`. -/// 5. A subsequent poll yields `None` (stream terminated). -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn test_stream_stops_on_ws_drop() { - let _ = tracing_subscriber::fmt() - .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) - .try_init(); - - // --- Setup --- - let anvil = TestAnvil::spawn().expect("failed to spawn anvil"); - let deployer = anvil.signer(0).expect("failed to get deployer"); - let registry_address = anvil - .deploy_world_id_registry_with_depth(deployer, 8) - .await - .expect("failed to deploy registry"); - - let provider = ProviderArgs::new() - .with_http_urls([anvil.endpoint()]) - .http() - .await - .expect("failed to build provider"); - - let blockchain = Blockchain::new(provider.clone(), anvil.ws_endpoint(), registry_address) - .await - .expect("failed to create Blockchain"); - - let http_provider = - ProviderBuilder::new().connect_http(anvil.endpoint().parse::().unwrap()); - - let from_block = http_provider - .get_block_number() - .await - .expect("failed to get block number"); - - // Small backfill so we quickly reach the WS phase. - let backfill_count: u64 = 10; - - create_accounts( - anvil.endpoint(), - anvil.signer(0).unwrap(), - registry_address, - 1, - backfill_count, - ) - .await; - - let backfill_log_count = http_provider - .get_logs( - &Filter::new() - .address(registry_address) - .event_signature(RegistryEvent::signatures()) - .from_block(from_block + 1), - ) - .await - .expect("failed to count backfill logs") - .len(); - assert!(backfill_log_count > 0, "expected some backfill logs"); - - let mut stream = blockchain.backfill_and_stream_events(from_block + 1, 2); - - // Consume all backfill events. - let _backfill_events: Vec> = tokio::time::timeout( - Duration::from_secs(30), - stream.by_ref().take(backfill_log_count).try_collect(), - ) - .await - .expect("timed out waiting for backfill events") - .expect("backfill stream error"); - - tokio::spawn(async move { - tokio::time::sleep(Duration::from_millis(500)).await; - drop(anvil); - }); - - // The stream should yield an error when the WS connection drops. - let next = tokio::time::timeout(Duration::from_secs(30), stream.next()) - .await - .expect("timed out waiting for WS drop error"); - - assert!( - matches!( - next, - Some(Err( - BlockchainError::WsSubscriptionClosed | BlockchainError::Rpc(_) - )) - ), - "expected Some(Err(WsSubscriptionClosed | Rpc(...))) after WS drop, got {next:?}" - ); - - // After the error the stream should be terminated. - let after = tokio::time::timeout(Duration::from_secs(5), stream.next()) - .await - .expect("timed out waiting for stream termination"); - - assert!( - after.is_none(), - "expected None (stream terminated) after WS error, got {after:?}" - ); -} diff --git a/services/indexer/tests/test_get_authenticator_pubkeys.rs b/services/indexer/tests/test_get_authenticator_pubkeys.rs index e6a17ccd1..4a2b4a51b 100644 --- a/services/indexer/tests/test_get_authenticator_pubkeys.rs +++ b/services/indexer/tests/test_get_authenticator_pubkeys.rs @@ -34,6 +34,8 @@ async fn test_get_authenticator_pubkeys_returns_offchain_signer_commitment() { start_block: 0, batch_size: 1000, tree_max_block_age: 1000, + confirmations: 0, + poll_interval_secs: 1, }, http_config: HttpConfig { http_addr: "0.0.0.0:8086".parse().unwrap(), @@ -44,7 +46,6 @@ async fn test_get_authenticator_pubkeys_returns_offchain_signer_commitment() { }, db_url: setup.db_url.clone(), provider: ProviderArgs::new().with_http_urls([setup.rpc_url()]), - ws_rpc_url: setup.ws_url(), registry_address: setup.registry_address, tree_cache: TreeCacheConfig { cache_file_path: temp_cache_path.to_str().unwrap().to_string(), diff --git a/services/indexer/tests/test_get_packed_account.rs b/services/indexer/tests/test_get_packed_account.rs index 41f75d2d8..39198a760 100644 --- a/services/indexer/tests/test_get_packed_account.rs +++ b/services/indexer/tests/test_get_packed_account.rs @@ -35,6 +35,8 @@ async fn test_packed_account_endpoint() { start_block: 0, batch_size: 1000, tree_max_block_age: 1000, + confirmations: 0, + poll_interval_secs: 1, }, http_config: HttpConfig { http_addr: "0.0.0.0:8083".parse().unwrap(), @@ -45,7 +47,6 @@ async fn test_packed_account_endpoint() { }, db_url: setup.db_url.clone(), provider: ProviderArgs::new().with_http_urls([setup.rpc_url()]), - ws_rpc_url: setup.ws_url(), registry_address: setup.registry_address, tree_cache: TreeCacheConfig { cache_file_path: temp_cache_path.to_str().unwrap().to_string(), diff --git a/services/indexer/tests/test_get_pending_recovery_agent.rs b/services/indexer/tests/test_get_pending_recovery_agent.rs index 21b998b81..52428d605 100644 --- a/services/indexer/tests/test_get_pending_recovery_agent.rs +++ b/services/indexer/tests/test_get_pending_recovery_agent.rs @@ -90,6 +90,8 @@ async fn test_get_pending_recovery_agent_endpoint() { start_block: 0, batch_size: 1000, tree_max_block_age: 1000, + confirmations: 0, + poll_interval_secs: 1, }, http_config: HttpConfig { http_addr: "0.0.0.0:8086".parse().unwrap(), @@ -100,7 +102,6 @@ async fn test_get_pending_recovery_agent_endpoint() { }, db_url: setup.db_url.clone(), provider: ProviderArgs::new().with_http_urls([setup.rpc_url()]), - ws_rpc_url: setup.ws_url(), registry_address: setup.registry_address, tree_cache: TreeCacheConfig { cache_file_path: temp_cache_path.to_str().unwrap().to_string(), diff --git a/services/indexer/tests/test_get_recovery_agent.rs b/services/indexer/tests/test_get_recovery_agent.rs index 4262ab7a7..53112b010 100644 --- a/services/indexer/tests/test_get_recovery_agent.rs +++ b/services/indexer/tests/test_get_recovery_agent.rs @@ -33,6 +33,8 @@ async fn test_get_recovery_agent_endpoint() { start_block: 0, batch_size: 1000, tree_max_block_age: 1000, + confirmations: 0, + poll_interval_secs: 1, }, http_config: HttpConfig { http_addr: "0.0.0.0:8085".parse().unwrap(), @@ -43,7 +45,6 @@ async fn test_get_recovery_agent_endpoint() { }, db_url: setup.db_url.clone(), provider: ProviderArgs::new().with_http_urls([setup.rpc_url()]), - ws_rpc_url: setup.ws_url(), registry_address: setup.registry_address, tree_cache: TreeCacheConfig { cache_file_path: temp_cache_path.to_str().unwrap().to_string(), diff --git a/services/indexer/tests/test_get_signature_nonce.rs b/services/indexer/tests/test_get_signature_nonce.rs index 6959b2eb3..0f1f0d1e7 100644 --- a/services/indexer/tests/test_get_signature_nonce.rs +++ b/services/indexer/tests/test_get_signature_nonce.rs @@ -35,6 +35,8 @@ async fn test_signature_nonce_endpoint() { start_block: 0, batch_size: 1000, tree_max_block_age: 1000, + confirmations: 0, + poll_interval_secs: 1, }, http_config: HttpConfig { http_addr: "0.0.0.0:8084".parse().unwrap(), @@ -45,7 +47,6 @@ async fn test_signature_nonce_endpoint() { }, db_url: setup.db_url.clone(), provider: ProviderArgs::new().with_http_urls([setup.rpc_url()]), - ws_rpc_url: setup.ws_url(), registry_address: setup.registry_address, tree_cache: TreeCacheConfig { cache_file_path: temp_cache_path.to_str().unwrap().to_string(), diff --git a/services/indexer/tests/test_inclusion_proof.rs b/services/indexer/tests/test_inclusion_proof.rs index 27d0189b9..3d49bfec2 100644 --- a/services/indexer/tests/test_inclusion_proof.rs +++ b/services/indexer/tests/test_inclusion_proof.rs @@ -49,6 +49,8 @@ async fn test_backfill_and_live_sync() { start_block: 0, batch_size: 1000, tree_max_block_age: 1000, + confirmations: 0, + poll_interval_secs: 1, }, http_config: HttpConfig { http_addr: "0.0.0.0:8080".parse().unwrap(), @@ -59,7 +61,6 @@ async fn test_backfill_and_live_sync() { }, db_url: setup.db_url.clone(), provider: ProviderArgs::new().with_http_urls([setup.rpc_url()]), - ws_rpc_url: setup.ws_url(), registry_address: setup.registry_address, tree_cache: TreeCacheConfig { cache_file_path: temp_cache_path.to_str().unwrap().to_string(), @@ -179,6 +180,8 @@ async fn test_insertion_cycle_and_avoids_race_condition() { start_block: 0, batch_size: 1000, tree_max_block_age: 1000, + confirmations: 0, + poll_interval_secs: 1, }, http_config: HttpConfig { http_addr: "0.0.0.0:8082".parse().unwrap(), @@ -189,7 +192,6 @@ async fn test_insertion_cycle_and_avoids_race_condition() { }, db_url: setup.db_url.clone(), provider: ProviderArgs::new().with_http_urls([setup.rpc_url()]), - ws_rpc_url: setup.ws_url(), registry_address: setup.registry_address, tree_cache: TreeCacheConfig { cache_file_path: temp_cache_path.to_str().unwrap().to_string(), diff --git a/services/indexer/tests/test_tree_cache.rs b/services/indexer/tests/test_tree_cache.rs index 9fb19d1b8..5a24daafe 100644 --- a/services/indexer/tests/test_tree_cache.rs +++ b/services/indexer/tests/test_tree_cache.rs @@ -65,6 +65,8 @@ async fn test_cache_creation_and_restoration() { start_block: 0, batch_size: 1000, tree_max_block_age: 1000, + confirmations: 0, + poll_interval_secs: 1, }, http_config: HttpConfig { http_addr: "0.0.0.0:8090".parse().unwrap(), @@ -75,7 +77,6 @@ async fn test_cache_creation_and_restoration() { }, db_url: setup.db_url.clone(), provider: ProviderArgs::new().with_http_urls([setup.rpc_url()]), - ws_rpc_url: setup.ws_url(), registry_address: setup.registry_address, tree_cache: tree_cache_config.clone(), }; @@ -139,6 +140,8 @@ async fn test_incremental_replay() { start_block: 0, batch_size: 1000, tree_max_block_age: 1000, + confirmations: 0, + poll_interval_secs: 1, }, http_config: HttpConfig { http_addr: "0.0.0.0:8091".parse().unwrap(), @@ -149,7 +152,6 @@ async fn test_incremental_replay() { }, db_url: setup.db_url.clone(), provider: ProviderArgs::new().with_http_urls([setup.rpc_url()]), - ws_rpc_url: setup.ws_url(), registry_address: setup.registry_address, tree_cache: tree_cache_config.clone(), }; @@ -197,6 +199,8 @@ async fn test_incremental_replay() { start_block: 0, batch_size: 1000, tree_max_block_age: 1000, + confirmations: 0, + poll_interval_secs: 1, }, http_config: HttpConfig { http_addr: "0.0.0.0:8092".parse().unwrap(), @@ -207,7 +211,6 @@ async fn test_incremental_replay() { }, db_url: setup.db_url.clone(), provider: ProviderArgs::new().with_http_urls([setup.rpc_url()]), - ws_rpc_url: setup.ws_url(), registry_address: setup.registry_address, tree_cache: tree_cache_config.clone(), }; @@ -270,6 +273,8 @@ async fn test_missing_cache_creates_new() { start_block: 0, batch_size: 1000, tree_max_block_age: 1000, + confirmations: 0, + poll_interval_secs: 1, }, http_config: HttpConfig { http_addr: "0.0.0.0:8093".parse().unwrap(), @@ -280,7 +285,6 @@ async fn test_missing_cache_creates_new() { }, db_url: setup.db_url.clone(), provider: ProviderArgs::new().with_http_urls([setup.rpc_url()]), - ws_rpc_url: setup.ws_url(), registry_address: setup.registry_address, tree_cache: tree_cache_config.clone(), }; @@ -335,6 +339,8 @@ async fn test_http_only_cache_refresh() { start_block: 0, batch_size: 1000, tree_max_block_age: 1000, + confirmations: 0, + poll_interval_secs: 1, }, http_config: HttpConfig { http_addr: "0.0.0.0:8094".parse().unwrap(), @@ -345,7 +351,6 @@ async fn test_http_only_cache_refresh() { }, db_url: setup.db_url.clone(), provider: ProviderArgs::new().with_http_urls([setup.rpc_url()]), - ws_rpc_url: setup.ws_url(), registry_address: setup.registry_address, tree_cache: tree_cache_config.clone(), }; @@ -382,7 +387,6 @@ async fn test_http_only_cache_refresh() { }, db_url: setup.db_url.clone(), provider: ProviderArgs::new().with_http_urls([setup.rpc_url()]), - ws_rpc_url: setup.ws_url(), registry_address: setup.registry_address, tree_cache: tree_cache_config.clone(), }; @@ -457,6 +461,8 @@ async fn test_authenticator_removed_replay() { start_block: 0, batch_size: 1000, tree_max_block_age: 1000, + confirmations: 0, + poll_interval_secs: 1, }, http_config: HttpConfig { http_addr: "0.0.0.0:8098".parse().unwrap(), @@ -467,7 +473,6 @@ async fn test_authenticator_removed_replay() { }, db_url: setup.db_url.clone(), provider: ProviderArgs::new().with_http_urls([setup.rpc_url()]), - ws_rpc_url: setup.ws_url(), registry_address: setup.registry_address, tree_cache: tree_cache_config.clone(), }; @@ -511,12 +516,12 @@ async fn test_authenticator_removed_replay() { tx_hash: U256::from(1234), log_index: 0, details: RegistryEvent::AuthenticatorRemoved(AuthenticatorRemovedEvent { - leaf_index: 1, - pubkey_id: 0, - authenticator_address: Address::ZERO, - authenticator_pubkey: U256::ZERO, - old_offchain_signer_commitment: U256::ZERO, - new_offchain_signer_commitment: new_commitment_after_removal, + leafIndex: 1, + pubkeyId: 0, + authenticatorAddress: Address::ZERO, + authenticatorPubkey: U256::ZERO, + oldOffchainSignerCommitment: U256::ZERO, + newOffchainSignerCommitment: new_commitment_after_removal, }), }; @@ -543,7 +548,6 @@ async fn test_authenticator_removed_replay() { // Restart indexer with replay - it should restore from cache and replay the removal event let db_url = setup.db_url.clone(); let rpc_url = setup.rpc_url(); - let ws_url = setup.ws_url(); let registry_address = setup.registry_address; let cfg2 = GlobalConfig { @@ -553,6 +557,8 @@ async fn test_authenticator_removed_replay() { start_block: 0, batch_size: 1000, tree_max_block_age: 1000, + confirmations: 0, + poll_interval_secs: 1, }, http_config: HttpConfig { http_addr: "0.0.0.0:8096".parse().unwrap(), @@ -563,7 +569,6 @@ async fn test_authenticator_removed_replay() { }, db_url: db_url.clone(), provider: ProviderArgs::new().with_http_urls([rpc_url.clone()]), - ws_rpc_url: ws_url.clone(), registry_address, tree_cache: tree_cache_config.clone(), }; @@ -584,6 +589,8 @@ async fn test_authenticator_removed_replay() { start_block: 0, batch_size: 1000, tree_max_block_age: 1000, + confirmations: 0, + poll_interval_secs: 1, }, http_config: HttpConfig { http_addr: "0.0.0.0:8097".parse().unwrap(), @@ -594,7 +601,6 @@ async fn test_authenticator_removed_replay() { }, db_url: db_url.clone(), provider: ProviderArgs::new().with_http_urls([rpc_url.clone()]), - ws_rpc_url: ws_url.clone(), registry_address, tree_cache: tree_cache_config_fresh.clone(), }; @@ -654,6 +660,8 @@ async fn test_init_root_matches_contract() { start_block: 0, batch_size: 1000, tree_max_block_age: 1000, + confirmations: 0, + poll_interval_secs: 1, }, http_config: HttpConfig { http_addr: "0.0.0.0:8100".parse().unwrap(), @@ -664,7 +672,6 @@ async fn test_init_root_matches_contract() { }, db_url: setup.db_url.clone(), provider: ProviderArgs::new().with_http_urls([setup.rpc_url()]), - ws_rpc_url: setup.ws_url(), registry_address: setup.registry_address, tree_cache: tree_cache_config.clone(), }; @@ -738,6 +745,8 @@ async fn test_replay_root_matches_contract() { start_block: 0, batch_size: 1000, tree_max_block_age: 1000, + confirmations: 0, + poll_interval_secs: 1, }, http_config: HttpConfig { http_addr: "0.0.0.0:8101".parse().unwrap(), @@ -748,7 +757,6 @@ async fn test_replay_root_matches_contract() { }, db_url: setup.db_url.clone(), provider: ProviderArgs::new().with_http_urls([setup.rpc_url()]), - ws_rpc_url: setup.ws_url(), registry_address: setup.registry_address, tree_cache: tree_cache_config.clone(), }; @@ -801,6 +809,8 @@ async fn test_replay_root_matches_contract() { start_block: 0, batch_size: 1000, tree_max_block_age: 1000, + confirmations: 0, + poll_interval_secs: 1, }, http_config: HttpConfig { http_addr: "0.0.0.0:8102".parse().unwrap(), @@ -811,7 +821,6 @@ async fn test_replay_root_matches_contract() { }, db_url: setup.db_url.clone(), provider: ProviderArgs::new().with_http_urls([setup.rpc_url()]), - ws_rpc_url: setup.ws_url(), registry_address: setup.registry_address, tree_cache: tree_cache_config.clone(), }; @@ -893,6 +902,8 @@ async fn test_corrupted_cache_returns_error() { start_block: 0, batch_size: 1000, tree_max_block_age: 1000, + confirmations: 0, + poll_interval_secs: 1, }, http_config: HttpConfig { http_addr: "0.0.0.0:8103".parse().unwrap(), @@ -903,7 +914,6 @@ async fn test_corrupted_cache_returns_error() { }, db_url: setup.db_url.clone(), provider: ProviderArgs::new().with_http_urls([setup.rpc_url()]), - ws_rpc_url: setup.ws_url(), registry_address: setup.registry_address, tree_cache: tree_cache_config.clone(), }; @@ -946,7 +956,6 @@ async fn test_corrupted_cache_returns_error() { }, db_url: setup.db_url.clone(), provider: ProviderArgs::new().with_http_urls([setup.rpc_url()]), - ws_rpc_url: setup.ws_url(), registry_address: setup.registry_address, tree_cache: tree_cache_config.clone(), }; @@ -1004,7 +1013,6 @@ async fn test_sanity_check_exits_on_root_mismatch() { }, db_url: setup.db_url.clone(), provider: ProviderArgs::new().with_http_urls([setup.rpc_url()]), - ws_rpc_url: setup.ws_url(), registry_address: setup.registry_address, tree_cache: tree_cache_config.clone(), }; diff --git a/services/indexer/tests/test_tree_unit.rs b/services/indexer/tests/test_tree_unit.rs index fcf06bb60..45c61a588 100644 --- a/services/indexer/tests/test_tree_unit.rs +++ b/services/indexer/tests/test_tree_unit.rs @@ -102,11 +102,11 @@ async fn test_replay_with_no_new_events() { tx_hash: U256::from(1), log_index: 0, details: RegistryEvent::AccountCreated(AccountCreatedEvent { - leaf_index: 1, - recovery_address: Address::ZERO, - authenticator_addresses: vec![], - authenticator_pubkeys: vec![], - offchain_signer_commitment: U256::from(100), + leafIndex: 1, + recoveryAddress: Address::ZERO, + authenticatorAddresses: vec![], + authenticatorPubkeys: vec![], + offchainSignerCommitment: U256::from(100), }), }, ) @@ -152,11 +152,11 @@ async fn test_replay_deduplication() { tx_hash: U256::from(1), log_index: 0, details: RegistryEvent::AccountCreated(AccountCreatedEvent { - leaf_index: 1, - recovery_address: Address::ZERO, - authenticator_addresses: vec![], - authenticator_pubkeys: vec![], - offchain_signer_commitment: U256::from(100), + leafIndex: 1, + recoveryAddress: Address::ZERO, + authenticatorAddresses: vec![], + authenticatorPubkeys: vec![], + offchainSignerCommitment: U256::from(100), }), }, ) @@ -183,13 +183,13 @@ async fn test_replay_deduplication() { tx_hash: U256::from(block), log_index: 0, details: RegistryEvent::AccountUpdated(AccountUpdatedEvent { - leaf_index: 1, - pubkey_id: 0, - new_authenticator_pubkey: U256::from(commitment), - old_authenticator_address: Address::ZERO, - new_authenticator_address: Address::ZERO, - old_offchain_signer_commitment: U256::ZERO, - new_offchain_signer_commitment: U256::from(commitment), + leafIndex: 1, + pubkeyId: 0, + newAuthenticatorPubkey: U256::from(commitment), + oldAuthenticatorAddress: Address::ZERO, + newAuthenticatorAddress: Address::ZERO, + oldOffchainSignerCommitment: U256::ZERO, + newOffchainSignerCommitment: U256::from(commitment), }), }, ) @@ -242,11 +242,11 @@ async fn test_replay_matches_fresh_build() { tx_hash: U256::from(1), log_index: 0, details: RegistryEvent::AccountCreated(AccountCreatedEvent { - leaf_index: 1, - recovery_address: Address::ZERO, - authenticator_addresses: vec![], - authenticator_pubkeys: vec![], - offchain_signer_commitment: U256::from(100), + leafIndex: 1, + recoveryAddress: Address::ZERO, + authenticatorAddresses: vec![], + authenticatorPubkeys: vec![], + offchainSignerCommitment: U256::from(100), }), }, ) @@ -260,11 +260,11 @@ async fn test_replay_matches_fresh_build() { tx_hash: U256::from(2), log_index: 1, details: RegistryEvent::AccountCreated(AccountCreatedEvent { - leaf_index: 2, - recovery_address: Address::ZERO, - authenticator_addresses: vec![], - authenticator_pubkeys: vec![], - offchain_signer_commitment: U256::from(200), + leafIndex: 2, + recoveryAddress: Address::ZERO, + authenticatorAddresses: vec![], + authenticatorPubkeys: vec![], + offchainSignerCommitment: U256::from(200), }), }, ) @@ -290,13 +290,13 @@ async fn test_replay_matches_fresh_build() { tx_hash: U256::from(11), log_index: 0, details: RegistryEvent::AccountUpdated(AccountUpdatedEvent { - leaf_index: 1, - pubkey_id: 0, - new_authenticator_pubkey: U256::from(300), - old_authenticator_address: Address::ZERO, - new_authenticator_address: Address::ZERO, - old_offchain_signer_commitment: U256::ZERO, - new_offchain_signer_commitment: U256::from(300), + leafIndex: 1, + pubkeyId: 0, + newAuthenticatorPubkey: U256::from(300), + oldAuthenticatorAddress: Address::ZERO, + newAuthenticatorAddress: Address::ZERO, + oldOffchainSignerCommitment: U256::ZERO, + newOffchainSignerCommitment: U256::from(300), }), }, ) @@ -310,13 +310,13 @@ async fn test_replay_matches_fresh_build() { tx_hash: U256::from(12), log_index: 1, details: RegistryEvent::AccountUpdated(AccountUpdatedEvent { - leaf_index: 2, - pubkey_id: 0, - new_authenticator_pubkey: U256::from(400), - old_authenticator_address: Address::ZERO, - new_authenticator_address: Address::ZERO, - old_offchain_signer_commitment: U256::ZERO, - new_offchain_signer_commitment: U256::from(400), + leafIndex: 2, + pubkeyId: 0, + newAuthenticatorPubkey: U256::from(400), + oldAuthenticatorAddress: Address::ZERO, + newAuthenticatorAddress: Address::ZERO, + oldOffchainSignerCommitment: U256::ZERO, + newOffchainSignerCommitment: U256::from(400), }), }, ) @@ -330,11 +330,11 @@ async fn test_replay_matches_fresh_build() { tx_hash: U256::from(13), log_index: 0, details: RegistryEvent::AccountCreated(AccountCreatedEvent { - leaf_index: 3, - recovery_address: Address::ZERO, - authenticator_addresses: vec![], - authenticator_pubkeys: vec![], - offchain_signer_commitment: U256::from(500), + leafIndex: 3, + recoveryAddress: Address::ZERO, + authenticatorAddresses: vec![], + authenticatorPubkeys: vec![], + offchainSignerCommitment: U256::from(500), }), }, ) @@ -419,13 +419,13 @@ async fn test_sync_from_db_deduplication() { tx_hash: U256::from(block), log_index: 0, details: RegistryEvent::AccountUpdated(AccountUpdatedEvent { - leaf_index: 1, - pubkey_id: 0, - new_authenticator_pubkey: U256::from(commitment), - old_authenticator_address: Address::ZERO, - new_authenticator_address: Address::ZERO, - old_offchain_signer_commitment: U256::ZERO, - new_offchain_signer_commitment: U256::from(commitment), + leafIndex: 1, + pubkeyId: 0, + newAuthenticatorPubkey: U256::from(commitment), + oldAuthenticatorAddress: Address::ZERO, + newAuthenticatorAddress: Address::ZERO, + oldOffchainSignerCommitment: U256::ZERO, + newOffchainSignerCommitment: U256::from(commitment), }), }, ) @@ -476,11 +476,11 @@ async fn test_handle_registry_event_root_mismatch() { tx_hash: U256::from(50), log_index: 0, details: RegistryEvent::AccountCreated(AccountCreatedEvent { - leaf_index: 2, - recovery_address: Address::ZERO, - authenticator_addresses: vec![], - authenticator_pubkeys: vec![], - offchain_signer_commitment: U256::from(200), + leafIndex: 2, + recoveryAddress: Address::ZERO, + authenticatorAddresses: vec![], + authenticatorPubkeys: vec![], + offchainSignerCommitment: U256::from(200), }), }; @@ -537,11 +537,11 @@ async fn test_handle_registry_event_root_match() { tx_hash: U256::from(50), log_index: 0, details: RegistryEvent::AccountCreated(AccountCreatedEvent { - leaf_index: 1, - recovery_address: Address::ZERO, - authenticator_addresses: vec![], - authenticator_pubkeys: vec![], - offchain_signer_commitment: U256::from(100), + leafIndex: 1, + recoveryAddress: Address::ZERO, + authenticatorAddresses: vec![], + authenticatorPubkeys: vec![], + offchainSignerCommitment: U256::from(100), }), };