feat(indexer): filter event capture by declared sources

Cargo.lock

@@ -4612,6 +4612,7 @@ dependencies = [
  "risc0-zkvm",
  "serde",
  "serde_json",
+ "serde_with",
  "storage",
  "tempfile",
  "test_methods",

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

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<RocksDBIO>,
     current_state: Arc<RwLock<V03State>>,
+    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<Self> {
+    pub fn open_db(
+        location: &Path,
+        genesis_seed: Vec<(AccountId, Account)>,
+        event_filter: EventFilter,
+    ) -> Result<Self> {
         #[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;

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<EventSourceConfig>),
+}
+
+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<Vec<Selector>>,
+}
+
+#[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 = <AccountId as FromStr>::Err;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        Ok(Self(s.parse::<AccountId>()?.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<Self, Self::Err> {
+        let mut bytes = [0_u8; 8];
+        hex::decode_to_slice(s, &mut bytes)?;
+        Ok(Self(bytes))
+    }
+}
+
+impl EventFilterConfig {
+    pub fn to_filter(&self) -> Result<EventFilter> {
+        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::<Selector>().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::<EventFilterConfig>(&json).is_err());
+    }
+}

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<ProgramId, SelectorFilter>),
+}
+
+#[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<TxEvents>) -> Vec<TxEvents> {
+        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<TransactionEvent>) -> 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);
+    }
+}

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")

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,
     })
 }