diff --git a/Cargo.lock b/Cargo.lock index fc2da0e40..ef0baa0b7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4612,6 +4612,7 @@ dependencies = [ "risc0-zkvm", "serde", "serde_json", + "serde_with", "storage", "tempfile", "test_methods", diff --git a/lez/indexer/core/Cargo.toml b/lez/indexer/core/Cargo.toml index 5a8f40b77..56f95f97b 100644 --- a/lez/indexer/core/Cargo.toml +++ b/lez/indexer/core/Cargo.toml @@ -16,7 +16,7 @@ chain_state.workspace = true common.workspace = true logos-blockchain-zone-sdk.workspace = true lee.workspace = true -lee_core.workspace = true +lee_core = { workspace = true, features = ["host"] } cross_zone.workspace = true cross_zone_inbox_core.workspace = true programs.workspace = true @@ -24,6 +24,7 @@ storage.workspace = true testnet_initial_state.workspace = true anyhow.workspace = true +serde_with.workspace = true arc-swap.workspace = true log.workspace = true serde.workspace = true diff --git a/lez/indexer/core/src/block_store.rs b/lez/indexer/core/src/block_store.rs index 6c7377cd3..a86375074 100644 --- a/lez/indexer/core/src/block_store.rs +++ b/lez/indexer/core/src/block_store.rs @@ -16,18 +16,23 @@ use logos_blockchain_zone_sdk::Slot; use storage::indexer::RocksDBIO; use tokio::sync::RwLock; -use crate::status::CrossZoneHalt; +use crate::{event_filter::EventFilter, status::CrossZoneHalt}; #[derive(Clone)] pub struct IndexerStore { dbio: Arc, current_state: Arc>, + event_filter: EventFilter, } impl IndexerStore { /// Starting database at the start of new chain. /// Creates files if necessary. - pub fn open_db(location: &Path, genesis_seed: Vec<(AccountId, Account)>) -> Result { + pub fn open_db( + location: &Path, + genesis_seed: Vec<(AccountId, Account)>, + event_filter: EventFilter, + ) -> Result { #[cfg(not(feature = "testnet"))] let base = testnet_initial_state::initial_state(); @@ -46,6 +51,7 @@ impl IndexerStore { Ok(Self { dbio: Arc::new(dbio), current_state: Arc::new(RwLock::new(current_state)), + event_filter, }) } @@ -270,8 +276,6 @@ impl IndexerStore { // TODO: we use scratch state to be atomic, but need to revisit how expensive a clone is let mut scratch = self.current_state.read().await.clone(); - // The events come from the same application that produced `scratch`, and are - // written in the same `put_block` batch as the block and that state. let events = match apply_block_to_state(block, &mut scratch) { Ok(events) => events, Err(err) => { @@ -282,6 +286,9 @@ impl IndexerStore { return Ok(AcceptOutcome::Parked(err)); } }; + // The retained events come from the same application that produced `scratch`, + // and are written in the same `put_block` batch as the block and that state. + let events = self.event_filter.filter_block(events); let mut stored = block.clone(); stored.bedrock_status = BedrockStatus::Finalized; @@ -300,6 +307,16 @@ impl IndexerStore { } } +#[cfg(test)] +fn open_default(home: &Path) -> IndexerStore { + open_with(home, EventFilter::default()) +} + +#[cfg(test)] +fn open_with(home: &Path, filter: EventFilter) -> IndexerStore { + IndexerStore::open_db(home, Vec::new(), filter).expect("open store") +} + #[cfg(test)] mod stall_reason_tests { use common::HashType; @@ -309,7 +326,7 @@ mod stall_reason_tests { #[tokio::test] async fn stall_reason_roundtrips_and_clears() { let dir = tempfile::tempdir().expect("tempdir"); - let store = IndexerStore::open_db(dir.path(), Vec::new()).expect("open store"); + let store = open_default(dir.path()); assert!(store.get_stall_reason().expect("get").is_none()); @@ -346,7 +363,7 @@ mod stall_reason_tests { #[tokio::test] async fn cross_zone_halt_roundtrips_and_clears() { let dir = tempfile::tempdir().expect("tempdir"); - let store = IndexerStore::open_db(dir.path(), Vec::new()).expect("open store"); + let store = open_default(dir.path()); assert!(store.get_cross_zone_halt().expect("get").is_none()); @@ -370,12 +387,15 @@ mod stall_reason_tests { #[cfg(test)] mod tests { + use std::collections::{HashMap, HashSet}; + use common::test_utils::{create_transaction_native_token_transfer, produce_dummy_block}; use lee_core::program::{InstructionData, ProgramEvent, ProgramId}; use tempfile::tempdir; use testnet_initial_state::initial_pub_accounts_private_keys; use super::*; + use crate::event_filter::SelectorFilter; // Host-side mirror of the `event_emitter` test guest's instruction. #[derive(serde::Serialize)] @@ -452,7 +472,7 @@ mod tests { fn correct_startup() { let home = tempdir().unwrap(); - let storage = IndexerStore::open_db(home.as_ref(), Vec::new()).unwrap(); + let storage = open_default(home.as_ref()); let final_id = storage.get_last_block_id().unwrap(); @@ -462,7 +482,7 @@ mod tests { #[tokio::test] async fn accept_block_applies_transfers_and_advances_tip() { let home = tempdir().unwrap(); - let store = IndexerStore::open_db(home.as_ref(), Vec::new()).unwrap(); + let store = open_default(home.as_ref()); let initial_accounts = initial_pub_accounts_private_keys(); let from = initial_accounts[0].account_id; @@ -504,7 +524,7 @@ mod tests { #[tokio::test] async fn account_state_at_block_reflects_history() { let home = tempdir().unwrap(); - let store = IndexerStore::open_db(home.as_ref(), Vec::new()).unwrap(); + let store = open_default(home.as_ref()); let initial_accounts = initial_pub_accounts_private_keys(); let from = initial_accounts[0].account_id; @@ -546,7 +566,7 @@ mod tests { #[tokio::test] async fn accept_block_captures_emitted_events() { let home = tempdir().unwrap(); - let store = IndexerStore::open_db(home.as_ref(), Vec::new()).unwrap(); + let store = open_with(home.as_ref(), EventFilter::Archival); let invoke_hash = seed_emitted_events(&store).await; @@ -586,12 +606,12 @@ mod tests { async fn events_survive_store_reopen() { let home = tempdir().unwrap(); let invoke_hash = { - let store = IndexerStore::open_db(home.as_ref(), Vec::new()).unwrap(); + let store = open_with(home.as_ref(), EventFilter::Archival); seed_emitted_events(&store).await }; // drop releases the RocksDB lock // Reopening replays state from the breakpoints; the events rows must be untouched. - let store = IndexerStore::open_db(home.as_ref(), Vec::new()).unwrap(); + let store = open_with(home.as_ref(), EventFilter::Archival); let groups = store .get_events_for_block(3) .unwrap() @@ -604,7 +624,7 @@ mod tests { #[tokio::test] async fn reaccepting_applied_block_does_not_duplicate_events() { let home = tempdir().unwrap(); - let store = IndexerStore::open_db(home.as_ref(), Vec::new()).unwrap(); + let store = open_with(home.as_ref(), EventFilter::Archival); seed_emitted_events(&store).await; let before = store.get_events_for_block(3).unwrap().unwrap(); @@ -618,10 +638,50 @@ mod tests { assert_eq!(store.get_events_for_block(3).unwrap().unwrap(), before); } + #[tokio::test] + async fn default_filter_stores_no_events() { + let home = tempdir().unwrap(); + let store = open_default(home.as_ref()); + + let invoke_hash = seed_emitted_events(&store).await; + + // Nothing survives the filter, so block 3 gets no row at all — where the + // archival run stores both emitted events. + assert_eq!(store.get_events_for_block(3).unwrap(), None); + assert!(store.get_events_range(1, 3).unwrap().is_empty()); + assert_eq!(store.get_events_by_tx_hash(invoke_hash.0).unwrap(), None); + } + + #[tokio::test] + async fn declared_source_filters_at_ingest() { + let home = tempdir().unwrap(); + let filter = EventFilter::Sources(HashMap::from([( + test_methods::EVENT_EMITTER_ID, + SelectorFilter::Only(HashSet::from([emitted(1).selector])), + )])); + let store = open_with(home.as_ref(), filter); + + let invoke_hash = seed_emitted_events(&store).await; + + // The invoke emits `emitted(0)` and `emitted(1)`; only the declared selector lands. + let groups = store + .get_events_for_block(3) + .unwrap() + .expect("the retained event must still produce a row"); + assert_eq!(groups.len(), 1); + assert_eq!(groups[0].tx_hash, invoke_hash); + assert_eq!(groups[0].events.len(), 1); + assert_eq!( + groups[0].events[0].program_id, + test_methods::EVENT_EMITTER_ID + ); + assert_eq!(groups[0].events[0].event, emitted(1)); + } + #[tokio::test] async fn blocks_without_events_have_no_row() { let home = tempdir().unwrap(); - let store = IndexerStore::open_db(home.as_ref(), Vec::new()).unwrap(); + let store = open_default(home.as_ref()); let initial_accounts = initial_pub_accounts_private_keys(); let from = initial_accounts[0].account_id; @@ -666,7 +726,7 @@ mod accept_tests { #[tokio::test] async fn non_genesis_first_block_parks_with_unexpected_id() { let dir = tempfile::tempdir().expect("tempdir"); - let store = IndexerStore::open_db(dir.path(), Vec::new()).expect("open store"); + let store = open_default(dir.path()); let block = valid_hash_block(2, HashType([0_u8; 32])); let outcome = store @@ -689,7 +749,7 @@ mod accept_tests { #[tokio::test] async fn hash_mismatch_parks() { let dir = tempfile::tempdir().expect("tempdir"); - let store = IndexerStore::open_db(dir.path(), Vec::new()).expect("open store"); + let store = open_default(dir.path()); let mut block = valid_hash_block(1, HashType([0_u8; 32])); block.header.timestamp = 999; // invalidates the stored hash @@ -707,7 +767,7 @@ mod accept_tests { #[tokio::test] async fn second_break_bumps_orphan_count_and_keeps_first() { let dir = tempfile::tempdir().expect("tempdir"); - let store = IndexerStore::open_db(dir.path(), Vec::new()).expect("open store"); + let store = open_default(dir.path()); let first = valid_hash_block(2, HashType([0_u8; 32])); store @@ -728,7 +788,7 @@ mod accept_tests { #[tokio::test] async fn deserialize_break_records_stall_without_header() { let dir = tempfile::tempdir().expect("tempdir"); - let store = IndexerStore::open_db(dir.path(), Vec::new()).expect("open store"); + let store = open_default(dir.path()); store .record_stall( @@ -746,7 +806,7 @@ mod accept_tests { #[tokio::test] async fn parks_then_recovers_on_valid_continuation() { let dir = tempfile::tempdir().expect("tempdir"); - let store = IndexerStore::open_db(dir.path(), Vec::new()).expect("open store"); + let store = open_default(dir.path()); // Genesis (block 1, clock-only) applies and advances the tip. let genesis = produce_dummy_block(1, None, vec![]); @@ -794,7 +854,7 @@ mod accept_tests { #[tokio::test] async fn accept_block_records_tip_inscription_slot() { let dir = tempfile::tempdir().expect("tempdir"); - let store = IndexerStore::open_db(dir.path(), Vec::new()).expect("open store"); + let store = open_default(dir.path()); assert_eq!(store.get_tip_slot().expect("get"), None); @@ -836,7 +896,7 @@ mod accept_tests { use testnet_initial_state::initial_pub_accounts_private_keys; let dir = tempfile::tempdir().expect("tempdir"); - let store = IndexerStore::open_db(dir.path(), Vec::new()).expect("open store"); + let store = open_default(dir.path()); let accounts = initial_pub_accounts_private_keys(); let from = accounts[0].account_id; @@ -886,7 +946,7 @@ mod accept_tests { use testnet_initial_state::initial_pub_accounts_private_keys; let dir = tempfile::tempdir().expect("tempdir"); - let store = IndexerStore::open_db(dir.path(), Vec::new()).expect("open store"); + let store = open_default(dir.path()); let accounts = initial_pub_accounts_private_keys(); let from = accounts[0].account_id; @@ -946,7 +1006,7 @@ mod accept_tests { use testnet_initial_state::initial_pub_accounts_private_keys; let dir = tempfile::tempdir().expect("tempdir"); - let store = IndexerStore::open_db(dir.path(), Vec::new()).expect("open store"); + let store = open_default(dir.path()); let accounts = initial_pub_accounts_private_keys(); let from = accounts[0].account_id; @@ -983,7 +1043,7 @@ mod accept_tests { // The #605 restart: reopening past the boundary must work. drop(store); - let reopened = IndexerStore::open_db(dir.path(), Vec::new()).expect("reopen"); + let reopened = open_default(dir.path()); assert_eq!(reopened.last_block().unwrap(), Some(101)); } @@ -992,7 +1052,7 @@ mod accept_tests { use testnet_initial_state::initial_pub_accounts_private_keys; let dir = tempfile::tempdir().expect("tempdir"); - let store = IndexerStore::open_db(dir.path(), Vec::new()).expect("open store"); + let store = open_default(dir.path()); let accounts = initial_pub_accounts_private_keys(); let from = accounts[0].account_id; diff --git a/lez/indexer/core/src/config.rs b/lez/indexer/core/src/config.rs index 9d9d11305..bcdbfa327 100644 --- a/lez/indexer/core/src/config.rs +++ b/lez/indexer/core/src/config.rs @@ -1,14 +1,20 @@ -use std::{fs::File, io::BufReader, num::NonZeroU32, path::Path, time::Duration}; +use std::{ + collections::HashMap, fmt::Display, fs::File, io::BufReader, num::NonZeroU32, path::Path, + str::FromStr, time::Duration, +}; -use anyhow::{Context as _, Result}; +use anyhow::{Context as _, Result, ensure}; use common::{HashType, config::BasicAuth}; use cross_zone_inbox_core::CrossZoneConfig; use humantime_serde; use lee::AccountId; pub use logos_blockchain_core::mantle::ops::channel::ChannelId; use serde::{Deserialize, Serialize}; +use serde_with::{DeserializeFromStr, SerializeDisplay}; use url::Url; +use crate::event_filter::{EventFilter, SelectorFilter}; + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ClientConfig { pub addr: Url, @@ -54,6 +60,11 @@ pub struct IndexerConfig { /// Defaults to `false`: on mismatch the indexer refuses to start. #[serde(default)] pub allow_chain_reset: bool, + /// Which emitted events this indexer persists at ingest. Omitted, it keeps + /// none: event storage is an explicit operator opt-in, declared per source + /// or wholesale via `archival`. + #[serde(default)] + pub event_filter: EventFilterConfig, } /// A genesis-funded bridge-lock holder balance, configured identically on the @@ -80,7 +91,178 @@ impl IndexerConfig { } } +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum EventFilterConfig { + Archival, + Sources(Vec), +} + +impl Default for EventFilterConfig { + fn default() -> Self { + Self::Sources(Vec::new()) + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct EventSourceConfig { + pub program_id: ProgramId, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub selectors: Option>, +} + +#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash, SerializeDisplay, DeserializeFromStr)] +pub struct ProgramId(pub lee_core::program::ProgramId); + +impl Display for ProgramId { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + AccountId::from(self.0).fmt(f) + } +} + +impl FromStr for ProgramId { + type Err = ::Err; + + fn from_str(s: &str) -> Result { + Ok(Self(s.parse::()?.into())) + } +} + +#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash, SerializeDisplay, DeserializeFromStr)] +pub struct Selector(pub [u8; 8]); + +impl Display for Selector { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", hex::encode(self.0)) + } +} + +impl FromStr for Selector { + type Err = hex::FromHexError; + + fn from_str(s: &str) -> Result { + let mut bytes = [0_u8; 8]; + hex::decode_to_slice(s, &mut bytes)?; + Ok(Self(bytes)) + } +} + +impl EventFilterConfig { + pub fn to_filter(&self) -> Result { + let declared = match self { + Self::Archival => return Ok(EventFilter::Archival), + Self::Sources(sources) => sources, + }; + let mut sources = HashMap::new(); + for source in declared { + let selectors = match &source.selectors { + None => SelectorFilter::All, + Some(selectors) => { + ensure!( + !selectors.is_empty(), + "event_filter declares program {} with no selectors", + source.program_id + ); + SelectorFilter::Only(selectors.iter().map(|selector| selector.0).collect()) + } + }; + ensure!( + sources.insert(source.program_id.0, selectors).is_none(), + "event_filter declares program {} twice", + source.program_id + ); + } + Ok(EventFilter::Sources(sources)) + } +} + /// The window applied when the config omits `peer_block_cache_window`. pub(crate) const fn default_peer_block_cache_window() -> NonZeroU32 { NonZeroU32::new(1024).expect("1024 is nonzero") } + +#[cfg(test)] +mod tests { + use std::collections::HashSet; + + use super::*; + + const PROGRAM_ONES_BASE58: &str = "4uQeVjgVccFGKht1dTy7bqxH3WehditPsgHyN1FSvRM"; + const PROGRAM_TWOS_BASE58: &str = "8opHzUMzEDVXeQm2FvwECguZ62JQGSmnkMawj1Vtqqh"; + + fn parse(json: &str) -> EventFilterConfig { + serde_json::from_str(json).unwrap() + } + + #[test] + fn selector_display_and_parse_are_pinned_to_hex() { + let selector = Selector([1; 8]); + assert_eq!(selector.to_string(), "0101010101010101"); + assert_eq!("0101010101010101".parse::().unwrap(), selector); + } + + #[test] + fn omitted_filter_keeps_nothing() { + assert_eq!( + EventFilterConfig::default().to_filter().unwrap(), + EventFilter::default() + ); + } + + #[test] + fn archival_config_maps_to_archival() { + assert_eq!( + parse(r#""archival""#).to_filter().unwrap(), + EventFilter::Archival + ); + } + + #[test] + fn declared_sources_map_to_selector_filters() { + let config = parse(&format!( + r#"{{ "sources": [ + {{ "program_id": "{PROGRAM_ONES_BASE58}" }}, + {{ "program_id": "{PROGRAM_TWOS_BASE58}", "selectors": ["0303030303030303"] }} + ] }}"# + )); + + let expected = EventFilter::Sources(HashMap::from([ + ([1; 8], SelectorFilter::All), + ([2; 8], SelectorFilter::Only(HashSet::from([[3; 8]]))), + ])); + assert_eq!(config.to_filter().unwrap(), expected); + } + + #[test] + fn duplicate_program_declaration_is_rejected() { + let config = parse(&format!( + r#"{{ "sources": [ + {{ "program_id": "{PROGRAM_ONES_BASE58}" }}, + {{ "program_id": "{PROGRAM_ONES_BASE58}", "selectors": ["0303030303030303"] }} + ] }}"# + )); + assert!(config.to_filter().is_err()); + } + + #[test] + fn empty_selector_list_is_rejected() { + let config = parse(&format!( + r#"{{ "sources": [ + {{ "program_id": "{PROGRAM_ONES_BASE58}", "selectors": [] }} + ] }}"# + )); + let err = config.to_filter().unwrap_err().to_string(); + assert!(err.contains("no selectors")); + } + + #[test] + fn unknown_source_field_is_rejected() { + let json = format!( + r#"{{ "sources": [ + {{ "program_id": "{PROGRAM_ONES_BASE58}", "selector": ["0303030303030303"] }} + ] }}"# + ); + assert!(serde_json::from_str::(&json).is_err()); + } +} diff --git a/lez/indexer/core/src/event_filter.rs b/lez/indexer/core/src/event_filter.rs new file mode 100644 index 000000000..3a5c89090 --- /dev/null +++ b/lez/indexer/core/src/event_filter.rs @@ -0,0 +1,153 @@ +use std::collections::{HashMap, HashSet}; + +use common::transaction::TxEvents; +use lee_core::program::{ProgramId, TransactionEvent}; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum EventFilter { + Archival, + Sources(HashMap), +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum SelectorFilter { + All, + Only(HashSet<[u8; 8]>), +} + +impl Default for EventFilter { + fn default() -> Self { + Self::Sources(HashMap::new()) + } +} + +impl EventFilter { + fn keeps(&self, event: &TransactionEvent) -> bool { + match self { + Self::Archival => true, + Self::Sources(sources) => match sources.get(&event.program_id) { + None => false, + Some(SelectorFilter::All) => true, + Some(SelectorFilter::Only(selectors)) => selectors.contains(&event.event.selector), + }, + } + } + + #[must_use] + pub fn filter_block(&self, block_events: Vec) -> Vec { + if matches!(self, Self::Archival) { + return block_events; + } + block_events + .into_iter() + .filter_map(|mut group| { + group.events.retain(|event| self.keeps(event)); + (!group.events.is_empty()).then_some(group) + }) + .collect() + } +} + +#[cfg(test)] +mod tests { + use common::HashType; + use lee_core::program::ProgramEvent; + + use super::*; + + const PROGRAM_A: ProgramId = [1; 8]; + const PROGRAM_B: ProgramId = [2; 8]; + const SELECTOR_X: [u8; 8] = [1; 8]; + const SELECTOR_Y: [u8; 8] = [2; 8]; + + fn event(program_id: ProgramId, selector: [u8; 8]) -> TransactionEvent { + TransactionEvent { + program_id, + event: ProgramEvent { + selector, + data: selector.to_vec(), + }, + } + } + + fn group(tx_index: u32, events: Vec) -> TxEvents { + TxEvents { + tx_index, + tx_hash: HashType([3; 32]), + events, + } + } + + fn sources(entries: Vec<(ProgramId, SelectorFilter)>) -> EventFilter { + EventFilter::Sources(entries.into_iter().collect()) + } + + #[test] + fn archival_keeps_every_event() { + let blocks = vec![group( + 0, + vec![event(PROGRAM_A, SELECTOR_X), event(PROGRAM_B, SELECTOR_Y)], + )]; + + assert_eq!(EventFilter::Archival.filter_block(blocks.clone()), blocks); + } + + #[test] + fn default_keeps_nothing() { + let blocks = vec![group( + 0, + vec![event(PROGRAM_A, SELECTOR_X), event(PROGRAM_B, SELECTOR_Y)], + )]; + + assert_eq!(EventFilter::default().filter_block(blocks), vec![]); + } + + #[test] + fn program_wide_entry_keeps_all_its_selectors_only() { + let filter = sources(vec![(PROGRAM_A, SelectorFilter::All)]); + let blocks = vec![group( + 0, + vec![ + event(PROGRAM_A, SELECTOR_X), + event(PROGRAM_A, SELECTOR_Y), + event(PROGRAM_B, SELECTOR_X), + ], + )]; + + let expected = vec![group( + 0, + vec![event(PROGRAM_A, SELECTOR_X), event(PROGRAM_A, SELECTOR_Y)], + )]; + assert_eq!(filter.filter_block(blocks), expected); + } + + #[test] + fn selector_entry_keeps_only_listed_selectors() { + let filter = sources(vec![( + PROGRAM_A, + SelectorFilter::Only(HashSet::from([SELECTOR_X])), + )]); + let blocks = vec![group( + 0, + vec![event(PROGRAM_A, SELECTOR_X), event(PROGRAM_A, SELECTOR_Y)], + )]; + + let expected = vec![group(0, vec![event(PROGRAM_A, SELECTOR_X)])]; + assert_eq!(filter.filter_block(blocks), expected); + } + + #[test] + fn fully_filtered_group_is_dropped_while_mixed_group_survives() { + let filter = sources(vec![(PROGRAM_A, SelectorFilter::All)]); + let blocks = vec![ + group(0, vec![event(PROGRAM_B, SELECTOR_X)]), + group( + 1, + vec![event(PROGRAM_B, SELECTOR_X), event(PROGRAM_A, SELECTOR_Y)], + ), + ]; + + let expected = vec![group(1, vec![event(PROGRAM_A, SELECTOR_Y)])]; + assert_eq!(filter.filter_block(blocks), expected); + } +} diff --git a/lez/indexer/core/src/lib.rs b/lez/indexer/core/src/lib.rs index c282159dd..c05608baf 100644 --- a/lez/indexer/core/src/lib.rs +++ b/lez/indexer/core/src/lib.rs @@ -23,6 +23,7 @@ use crate::{ pub mod block_store; pub mod config; pub mod cross_zone_verifier; +pub mod event_filter; mod retry; pub mod status; @@ -131,7 +132,8 @@ impl IndexerCore { // finalized blocks. `None` when cross-zone messaging is disabled. let verifier = CrossZoneVerifier::start(&config); - let store = IndexerStore::open_db(&home, genesis_accounts)?; + let store = + IndexerStore::open_db(&home, genesis_accounts, config.event_filter.to_filter()?)?; // A persisted halt outlives the process: report it from boot with its // stored reason. The ingest loop may still start and re-halt // identically, which refreshes the record. @@ -591,7 +593,7 @@ mod tests { use logos_blockchain_zone_sdk::Slot; use super::*; - use crate::config::{ChannelId, ClientConfig, IndexerConfig}; + use crate::config::{ChannelId, ClientConfig, EventFilterConfig, IndexerConfig}; /// The cursor must not move while more of the same slot may still arrive. /// @@ -657,6 +659,7 @@ mod tests { cross_zone: None, cross_zone_accept_unverified, peer_block_cache_window: NonZeroU32::new(1024).expect("1024 is nonzero"), + event_filter: EventFilterConfig::default(), bridge_lock_holdings: Vec::new(), }; IndexerCore::open(config, dir).expect("open core") diff --git a/test_fixtures/src/config.rs b/test_fixtures/src/config.rs index 34795de95..0ce39ea24 100644 --- a/test_fixtures/src/config.rs +++ b/test_fixtures/src/config.rs @@ -2,7 +2,7 @@ use std::{net::SocketAddr, num::NonZeroU32, path::PathBuf, time::Duration}; use anyhow::{Context as _, Result}; use bytesize::ByteSize; -use indexer_service::{ChannelId, ClientConfig, IndexerConfig}; +use indexer_service::{ChannelId, ClientConfig, EventFilterConfig, IndexerConfig}; use key_protocol::key_management::{KeyChain, secret_holders::SeedHolder}; use lee::{AccountId, PrivateKey, PublicKey}; use lee_core::Identifier; @@ -277,6 +277,7 @@ pub fn indexer_config( peer_block_cache_window: NonZeroU32::new(1024).expect("1024 is nonzero"), bridge_lock_holdings: Vec::new(), allow_chain_reset: false, + event_filter: EventFilterConfig::Archival, }) }