diff --git a/node/src/cli.rs b/node/src/cli.rs index c5438cbd..8d1a49f3 100644 --- a/node/src/cli.rs +++ b/node/src/cli.rs @@ -19,10 +19,12 @@ pub struct Cli { #[clap(long)] pub event_forwarder_rpc: Option, - /// URL of the prover-db indexer to forward indexed events to. Also - /// configurable via the `PROVER_DB_URL` environment variable. - #[clap(long, env)] - pub prover_db_url: Option, + /// Prover-db consumer settings. Setting `--prover-db-url` is what + /// turns this node into a prover-db indexer; the other fields in + /// the group only take effect when the URL is set. See the field + /// docs on [`ProverDbConsumerCli`] for grammar. + #[clap(flatten)] + pub prover_db: ProverDbConsumerCli, #[allow(missing_docs)] #[clap(flatten)] @@ -33,6 +35,48 @@ pub struct Cli { pub proof_of_sql_public_setup_args: ProofOfSqlPublicSetupArgs, } +/// Flattened consumer-side configuration. Lives on the top-level +/// `Cli` via `#[clap(flatten)]`. Validated at service init into +/// `sxt_core::prover_db_indexer::ProverDbConsumerConfig` before being +/// written to offchain local storage; see +/// `service::configure_prover_db_consumer`. +/// +/// On the CLI side the fields are individually optional / repeatable — +/// clap can't express "if URL is set, the group is active" in the type. +/// The resolver in `service.rs` enforces that constraint and produces a +/// concrete `ProverDbConsumerConfig` (URL non-optional) or refuses to +/// start. +#[derive(Debug, clap::Args)] +pub struct ProverDbConsumerCli { + /// URL of the prover-db indexer to forward indexed events to. + /// Also configurable via the `PROVER_DB_URL` env var. + /// + /// Presence of this flag is what enables the consumer. If omitted, + /// the OCW stays dormant regardless of `--prover-db-include`. + #[clap(long, env)] + pub prover_db_url: Option, + + /// Wildcard patterns naming which tables this node should forward + /// to the indexer. Repeat the flag, or pass a single comma- + /// separated value, or set `PROVER_DB_INCLUDE`. + /// + /// Each entry parses as a [`TableIdentifierFilter`]: two + /// dot-separated sides, each either a literal identifier or `*`. + /// Both sides match independently, so: + /// - `*.*` — every captured event (default if omitted). + /// - `NAMESPACE.*` — every table in NAMESPACE. + /// - `*.NAME` — every table called NAME, regardless of namespace. + /// - `NAMESPACE.NAME` — exactly that one table. + /// + /// Identifiers are uppercased before being written to offchain + /// storage so the byte-match against the on-chain canonical form + /// works without per-operator vigilance. + /// + /// [`TableIdentifierFilter`]: sxt_core::prover_db_indexer::TableIdentifierFilter + #[clap(long, env, value_delimiter = ',', default_value = "*.*")] + pub prover_db_include: Vec, +} + #[derive(Debug, clap::Subcommand)] #[allow(clippy::large_enum_variant)] pub enum Subcommand { diff --git a/node/src/service.rs b/node/src/service.rs index 184e9366..58829eb0 100644 --- a/node/src/service.rs +++ b/node/src/service.rs @@ -77,28 +77,43 @@ pub type TransactionPool = sc_transaction_pool::FullPool; /// imported and generated. const GRANDPA_JUSTIFICATION_PERIOD: u32 = 512; -/// Seed the prover-db indexer URL into offchain persistent local storage -/// at startup, before the first block is authored. +/// Validate the consumer CLI group into a concrete +/// `ProverDbConsumerConfig` (URL non-optional) and seed it under +/// `PROVER_DB_CONFIG_KEY` in offchain persistent local storage. /// -/// Stored under `sxt_core::prover_db_indexer::PROVER_DB_URL_KEY` as a -/// SCALE-encoded `Vec` of the URL bytes. +/// `--prover-db-url` is the single switch that enables the consumer: +/// present ⇒ write the config; absent ⇒ the OCW stays dormant and we +/// don't write anything. `--prover-db-include` is allowed (with its +/// `*.*` default) regardless of whether the URL is set; without a URL +/// it just has nothing to filter. #[expect( clippy::result_large_err, reason = "ServiceError is from substrate and cannot be modified" )] -fn configure_prover_db_url(backend: &FullBackend, url: &url::Url) -> Result<(), ServiceError> { +fn configure_prover_db_consumer( + backend: &FullBackend, + cli: &crate::cli::ProverDbConsumerCli, +) -> Result<(), ServiceError> { + let Some(url) = cli.prover_db_url.as_ref() else { + return Ok(()); + }; + + let cfg = sxt_core::prover_db_indexer::ProverDbConsumerConfig { + url: url.as_str().to_string(), + include: cli.prover_db_include.clone(), + }; + let Some(mut storage) = backend.offchain_storage() else { return Err(ServiceError::Other( "backend did not expose an offchain storage handle; \ - cannot apply --prover-db-url" + cannot apply --prover-db-url / --prover-db-include" .into(), )); }; - let encoded = url.as_str().as_bytes().to_vec().encode(); storage.set( STORAGE_PREFIX, - sxt_core::prover_db_indexer::PROVER_DB_URL_KEY, - &encoded, + sxt_core::prover_db_indexer::PROVER_DB_CONFIG_KEY, + &cfg.encode(), ); Ok(()) } @@ -351,22 +366,20 @@ pub fn new_full_base::Hash>>( other: (rpc_builder, import_setup, rpc_setup, mut telemetry, statement_store), } = new_partial(&config)?; - // `--prover-db-url` only takes effect when offchain indexing is - // enabled, since the prover-db-indexer OCW only runs in that case. - // Fail loud if the operator set the URL without enabling indexing, - // rather than silently ignoring it. - if let Some(url) = cli.prover_db_url.as_ref() { - if !config.offchain_worker.indexing_enabled { - return Err(ServiceError::Other( - "--prover-db-url was set but --enable-offchain-indexing is not \ - true; the prover-db-indexer offchain worker would be inactive. \ - Restart with --enable-offchain-indexing=true, or omit \ - --prover-db-url." - .into(), - )); - } - configure_prover_db_url(backend.as_ref(), url)?; + // The prover-db consumer (URL + include set) only takes effect when + // offchain indexing is enabled, since the OCW only runs in that + // case. Fail loud if the operator set --prover-db-url without + // enabling indexing, rather than silently ignoring it. + if cli.prover_db.prover_db_url.is_some() && !config.offchain_worker.indexing_enabled { + return Err(ServiceError::Other( + "--prover-db-url was set but --enable-offchain-indexing is not \ + true; the prover-db-indexer offchain worker would be inactive. \ + Restart with --enable-offchain-indexing=true, or omit \ + --prover-db-url." + .into(), + )); } + configure_prover_db_consumer(backend.as_ref(), &cli.prover_db)?; let metrics = N::register_notification_metrics( config.prometheus_config.as_ref().map(|cfg| &cfg.registry), diff --git a/pallets/prover_db_indexer/src/lib.rs b/pallets/prover_db_indexer/src/lib.rs index ae70b3b9..0f5d089a 100644 --- a/pallets/prover_db_indexer/src/lib.rs +++ b/pallets/prover_db_indexer/src/lib.rs @@ -60,9 +60,8 @@ mod proto { pub use pallet::*; /// Re-export of the canonical offchain local-storage key (defined in `sxt-core`), -/// kept here so existing call sites that reach for `pallet_prover_db_indexer::PROVER_DB_URL_KEY` -/// keep working. -pub use sxt_core::prover_db_indexer::PROVER_DB_URL_KEY; +/// so the node service and any external tooling have a single import. +pub use sxt_core::prover_db_indexer::PROVER_DB_CONFIG_KEY; /// Offchain DB key for the lock that serializes OCW consumer rounds. const OCW_LOCK_KEY: &[u8] = b"prover_db_indexer/ocw_lock"; @@ -78,9 +77,6 @@ const OCW_LOCK_KEY: &[u8] = b"prover_db_indexer/ocw_lock"; /// every variant looks the same. #[derive(Debug, snafu::Snafu)] pub enum ConsumerError { - /// The configured indexer URL was not valid UTF-8. - #[snafu(display("indexer URL was not valid UTF-8"))] - InvalidUrlEncoding, /// The configured indexer URL couldn't be parsed. #[snafu(display("indexer URL is not a valid URL: {error}"))] InvalidUrl { @@ -136,8 +132,11 @@ pub mod pallet { use sxt_core::prover_db_indexer::{ key_for_event, key_for_high_water, + table_matches_filters, BlockEvent, EventCapture, + ProverDbConsumerConfig, + TableIdentifierFilter, }; use crate::ConsumerError; @@ -204,6 +203,12 @@ pub mod pallet { return; } + // Capture every event unconditionally. Per-node filtering is + // applied later by the OCW consumer when forwarding to the + // indexer; the offchain queue mirrors the full block so every + // validator carries the same data, regardless of which subset + // each indexer cares about. + // `block_number()` always fits in u64 in our runtime (BlockNumber // is u32); `extrinsic_index()` is always `Some` while we're inside // an extrinsic, which is the only path that reaches `capture_events`. @@ -255,23 +260,22 @@ pub mod pallet { } }; - // 1. Check if this node is configured as an indexer. - let url_ref = StorageValueRef::persistent(crate::PROVER_DB_URL_KEY); - let url_bytes: Option> = url_ref.get::>().ok().flatten(); - - let Some(url_bytes) = url_bytes else { + // 1. Read the single offchain config. Absence ⇒ this node + // isn't an indexer; OCW is dormant. The node service + // writes the encoded `ProverDbConsumerConfig` at startup + // iff `--prover-db-url` was provided. + let cfg_ref = StorageValueRef::persistent(crate::PROVER_DB_CONFIG_KEY); + let Some(cfg) = cfg_ref.get::().ok().flatten() else { return Ok(()); }; - - let url_str = - core::str::from_utf8(&url_bytes).map_err(|_| ConsumerError::InvalidUrlEncoding)?; let url = - url::Url::parse(url_str).map_err(|error| ConsumerError::InvalidUrl { error })?; + url::Url::parse(&cfg.url).map_err(|error| ConsumerError::InvalidUrl { error })?; + let include_filters: Vec = cfg.include; polkadot_sdk::sp_tracing::debug!( target: "prover_db_indexer", - "consumer round starting; indexer base URL = {}", - url, + "consumer round starting; indexer base URL = {}; {} include filters", + url, include_filters.len(), ); // 2. Current block number — passed in by `offchain_worker`. @@ -313,7 +317,7 @@ pub mod pallet { ); for block_num in start..=end { - Self::forward_block(&url, block_num)?; + Self::forward_block(&url, block_num, &include_filters)?; // Checkpoint on the server (always, even for empty blocks). crate::http_client::checkpoint(&url, block_num) @@ -326,7 +330,11 @@ pub mod pallet { /// Forward a single block's events in extrinsic-index order, then /// clear the offchain entries we consumed. If the block had no /// captures (no high-water-mark key), this is a no-op. - fn forward_block(url: &url::Url, block_num: u64) -> Result<(), ConsumerError> { + fn forward_block( + url: &url::Url, + block_num: u64, + include_filters: &[TableIdentifierFilter], + ) -> Result<(), ConsumerError> { let Some(hwm) = crate::offchain_consumer::read_high_water(block_num) else { return Ok(()); }; @@ -342,7 +350,7 @@ pub mod pallet { let Some(events) = crate::offchain_consumer::read_events(block_num, ext_idx) else { continue; }; - Self::forward_events(url, block_num, &events)?; + Self::forward_events(url, block_num, &events, include_filters)?; crate::offchain_consumer::clear_events(block_num, ext_idx); } @@ -350,15 +358,29 @@ pub mod pallet { Ok(()) } - /// POST one extrinsic's captured events to the indexer in deposit order. + /// POST one extrinsic's captured events to the indexer in deposit + /// order, skipping any whose table doesn't match this node's + /// include set. The capture queue is unfiltered (every validator + /// records the full block), so the filter applies only to the + /// HTTP forwarding done by this node's OCW. fn forward_events( url: &url::Url, block_num: u64, events: &[BlockEvent<'_>], + include_filters: &[TableIdentifierFilter], ) -> Result<(), ConsumerError> { let seq = block_num; for event in events { + let table = match event { + BlockEvent::Create(entry) => entry.ident.as_ref(), + BlockEvent::Drop(ident) => ident.as_ref(), + BlockEvent::Insert(entry) => entry.table.as_ref(), + }; + if !table_matches_filters(table, include_filters) { + continue; + } + match event { BlockEvent::Drop(ident) => { crate::http_client::drop_table(url, seq, ident) diff --git a/pallets/prover_db_indexer/src/tests.rs b/pallets/prover_db_indexer/src/tests.rs index 05a8fd3f..32923ecc 100644 --- a/pallets/prover_db_indexer/src/tests.rs +++ b/pallets/prover_db_indexer/src/tests.rs @@ -21,20 +21,28 @@ use sxt_core::prover_db_indexer::{ key_for_high_water, BlockEvent, CreateEntry, + EventCapture, InsertEntry, + ProverDbConsumerConfig, + TableIdentifierFilter, }; use sxt_core::tables::TableIdentifier; use crate::mock::*; -use crate::{proto, PROVER_DB_URL_KEY}; +use crate::{proto, PROVER_DB_CONFIG_KEY}; type StateArc = std::sync::Arc>; const MOCK_URL: &str = "http://127.0.0.1:9999"; -fn encode_url() -> Vec { - codec::Encode::encode(&MOCK_URL.as_bytes().to_vec()) +/// SCALE-encode the unified consumer config the same way the node +/// service does at startup. +fn encode_config(include: Vec) -> Vec { + codec::Encode::encode(&ProverDbConsumerConfig { + url: MOCK_URL.to_string(), + include, + }) } fn expected_request(path: &str, body: Vec, response: Vec) -> PendingRequest { @@ -65,7 +73,18 @@ fn no_checkpoint_response() -> Vec { .encode_to_vec() } +/// Seed a `*.*` filter — what the operator gets when omitting +/// `--prover-db-include` (clap default). Every captured event should +/// reach the indexer. fn setup_with_url() -> (polkadot_sdk::sp_io::TestExternalities, StateArc) { + setup_with_config(vec!["*.*".parse().unwrap()]) +} + +/// Seed a non-empty include set under the unified config key. Used by +/// the consumer-side filter tests. +fn setup_with_config( + include: Vec, +) -> (polkadot_sdk::sp_io::TestExternalities, StateArc) { let mut ext = new_test_ext(); let (offchain, state) = TestOffchainExt::new(); ext.register_extension(OffchainWorkerExt::new(offchain.clone())); @@ -73,7 +92,7 @@ fn setup_with_url() -> (polkadot_sdk::sp_io::TestExternalities, StateArc) { state .write() .persistent_storage - .set(b"", PROVER_DB_URL_KEY, &encode_url()); + .set(b"", PROVER_DB_CONFIG_KEY, &encode_config(include)); (ext, state) } @@ -395,3 +414,232 @@ fn ocw_walks_multiple_extrinsics_in_one_block() { assert!(s.persistent_storage.get(&key_for_event(1, 3)).is_none()); assert!(s.persistent_storage.get(&key_for_high_water(1)).is_none()); } + +// ─── Include-set tests (consumer-side filter) ────────────────────────── + +/// Helper: build a Drop event for `(name, namespace)`. +fn drop_event(name: &str, namespace: &str) -> BlockEvent<'static> { + BlockEvent::Drop(Cow::Owned(TableIdentifier::from_str_unchecked( + name, namespace, + ))) +} + +/// Helper: build a Create event for `(name, namespace)`. +fn create_event(name: &str, namespace: &str) -> BlockEvent<'static> { + BlockEvent::Create(CreateEntry { + ident: Cow::Owned(TableIdentifier::from_str_unchecked(name, namespace)), + ddl: Cow::Owned(b"CREATE TABLE ...".to_vec()), + }) +} + +/// Helper: build an Insert event for `(name, namespace)`. +fn insert_event(name: &str, namespace: &str) -> BlockEvent<'static> { + BlockEvent::Insert(InsertEntry { + table: Cow::Owned(TableIdentifier::from_str_unchecked(name, namespace)), + data: Cow::Owned(b"rows".to_vec()), + }) +} + +/// `capture_events` is now unconditional — every event reaches the +/// offchain queue regardless of any per-node configuration. This test +/// is the single producer-side regression check. +#[test] +fn capture_events_writes_all_events_unconditionally() { + let mut ext = new_test_ext(); + ext.execute_with(|| { + System::set_block_number(1); + polkadot_sdk::frame_system::Pallet::::set_extrinsic_index(0); + + let events = vec![ + create_event("A", "ALPHA"), + insert_event("B", "BETA"), + drop_event("C", "GAMMA"), + ]; + ::capture_events(events); + }); + ext.persist_offchain_overlay(); + let db = ext.offchain_db(); + let stored = db.get(&key_for_event(1, 0)).unwrap(); + let decoded: Vec> = codec::Decode::decode(&mut &stored[..]).unwrap(); + assert_eq!(decoded.len(), 3); +} + +/// End-to-end consumer test: seed the node's include set into offchain +/// local storage (where the OCW reads it), seed a multi-event block, +/// and confirm only the matching events are POSTed to the indexer. +#[test] +fn ocw_forwards_only_events_matching_include_set() { + // Seed the node-local config with two include rules: any table in + // namespace ALPHA, plus the specific table BETA_NS.BETA_T. The + // service writes both URL + include atomically via a single + // SCALE-encoded `ProverDbConsumerConfig` — same shape this test + // mirrors. + let (mut ext, state) = setup_with_config(vec![ + "ALPHA.*".parse().unwrap(), + "BETA_NS.BETA_T".parse().unwrap(), + ]); + + // Block 1, extrinsic 0: a mix of matching and non-matching events. + seed_block_events( + &state, + 1, + 0, + vec![ + create_event("A", "ALPHA"), // match (namespace) + insert_event("BETA_T", "BETA_NS"), // match (table) + drop_event("OTHER", "GAMMA"), // skip + create_event("X", "GAMMA"), // skip + insert_event("Y", "ALPHA"), // match (namespace) + ], + ); + + // Only the three matching events should reach the indexer. Drops + // for the skipped GAMMA namespace get filtered too — uniform. + { + let mut s = state.write(); + s.expect_request(expected_request( + "/v1/get_last_checkpoint", + vec![], + no_checkpoint_response(), + )); + s.expect_request(expected_request( + "/v1/create_table", + proto::CreateTableRequest { + sequence_number: 1, + table_name: "ALPHA.A".into(), + arrow_schema: b"CREATE TABLE ...".to_vec(), + key: "META_ROW_NUMBER".into(), + } + .encode_to_vec(), + vec![], + )); + s.expect_request(expected_request( + "/v1/put_batches", + proto::PutBatchesRequest { + sequence_number: 1, + batches: vec![proto::TableBatch { + table_name: "BETA_NS.BETA_T".into(), + record_batch: b"rows".to_vec(), + }], + } + .encode_to_vec(), + vec![], + )); + s.expect_request(expected_request( + "/v1/put_batches", + proto::PutBatchesRequest { + sequence_number: 1, + batches: vec![proto::TableBatch { + table_name: "ALPHA.Y".into(), + record_batch: b"rows".to_vec(), + }], + } + .encode_to_vec(), + vec![], + )); + s.expect_request(expected_request( + "/v1/checkpoint", + proto::CheckpointRequest { sequence_number: 1 }.encode_to_vec(), + vec![], + )); + } + + ext.execute_with(|| { + System::set_block_number(1); + ProverDbIndexer::offchain_worker(1); + }); + + let s = state.read(); + assert!(s.persistent_storage.get(&key_for_event(1, 0)).is_none()); + assert!(s.persistent_storage.get(&key_for_high_water(1)).is_none()); +} + +/// Explicit `*.*` filter in offchain storage ⇒ forward every event — +/// this is what the operator gets when omitting `--prover-db-include` +/// because the CLI default is `*.*`. +#[test] +fn ocw_with_wildcard_filter_forwards_everything() { + let (mut ext, state) = setup_with_url(); + + let ddl = b"CREATE TABLE ANY.ANY (ID BIGINT NOT NULL)"; + seed_block_events( + &state, + 1, + 0, + vec![BlockEvent::Create(CreateEntry { + ident: Cow::Owned(TableIdentifier::from_str_unchecked("ANY", "ANY")), + ddl: ddl.to_vec().into(), + })], + ); + + { + let mut s = state.write(); + s.expect_request(expected_request( + "/v1/get_last_checkpoint", + vec![], + no_checkpoint_response(), + )); + s.expect_request(expected_request( + "/v1/create_table", + proto::CreateTableRequest { + sequence_number: 1, + table_name: "ANY.ANY".into(), + arrow_schema: ddl.to_vec(), + key: "META_ROW_NUMBER".into(), + } + .encode_to_vec(), + vec![], + )); + s.expect_request(expected_request( + "/v1/checkpoint", + proto::CheckpointRequest { sequence_number: 1 }.encode_to_vec(), + vec![], + )); + } + + ext.execute_with(|| { + System::set_block_number(1); + ProverDbIndexer::offchain_worker(1); + }); +} + +/// Empty include set in offchain storage ⇒ forward nothing. The +/// checkpoint still advances (so the OCW makes server-side progress +/// past empty-as-far-as-this-node-is-concerned blocks), but no +/// `/v1/create_table` request is emitted. `TestOffchainExt` would +/// panic on an unexpected request, so the absence of that expectation +/// is the assertion. +#[test] +fn ocw_with_empty_include_set_forwards_nothing() { + let (mut ext, state) = setup_with_config(Vec::new()); + + let ddl = b"CREATE TABLE ANY.ANY (ID BIGINT NOT NULL)"; + seed_block_events( + &state, + 1, + 0, + vec![BlockEvent::Create(CreateEntry { + ident: Cow::Owned(TableIdentifier::from_str_unchecked("ANY", "ANY")), + ddl: ddl.to_vec().into(), + })], + ); + + { + let mut s = state.write(); + s.expect_request(expected_request( + "/v1/get_last_checkpoint", + vec![], + no_checkpoint_response(), + )); + s.expect_request(expected_request( + "/v1/checkpoint", + proto::CheckpointRequest { sequence_number: 1 }.encode_to_vec(), + vec![], + )); + } + + ext.execute_with(|| { + System::set_block_number(1); + ProverDbIndexer::offchain_worker(1); + }); +} diff --git a/sxt-core/src/prover_db_indexer.rs b/sxt-core/src/prover_db_indexer.rs index 8989732d..79e650de 100644 --- a/sxt-core/src/prover_db_indexer.rs +++ b/sxt-core/src/prover_db_indexer.rs @@ -3,19 +3,48 @@ //! `pallet-tables` and `pallet-indexing`. use alloc::borrow::Cow; +use alloc::string::{String, ToString}; use alloc::vec::Vec; +use core::str::FromStr; use codec::{Decode, Encode}; +use snafu::Snafu; use crate::tables::TableIdentifier; +use crate::IDENT_LENGTH; -/// Offchain local-storage key holding the prover-db indexer URL. +/// Offchain local-storage key holding the prover-db indexer consumer's +/// configuration. The embedding node writes a SCALE-encoded +/// [`ProverDbConsumerConfig`] under this key at startup; the OCW reads +/// it once per round. Absence of the key means "OCW is dormant". /// -/// The embedding node is expected to write the configured URL to this -/// key in OCW persistent storage before block authoring begins; the -/// prover-db-indexer pallet's offchain worker reads it to know where to -/// forward events. If the key is unset the OCW stays dormant. -pub const PROVER_DB_URL_KEY: &[u8] = b"prover_db_indexer/prover_db_url"; +/// One key for the whole consumer rather than one-per-field so the +/// "consumer is enabled" and "consumer is disabled with stale filter +/// settings" states are unrepresentable. +pub const PROVER_DB_CONFIG_KEY: &[u8] = b"prover_db_indexer/consumer_config"; + +/// Per-node configuration that turns this node into a prover-db +/// indexer: the upstream URL to POST forwarded events to, plus the +/// optional include set that gates which events get forwarded. +/// +/// Lives in offchain local storage under [`PROVER_DB_CONFIG_KEY`]. +/// `url` is non-optional: writing the config _is_ the act of enabling +/// the consumer, and there's no enabled-without-URL state to express. +/// `include` may be empty, which the OCW treats as "match every table" +/// (same default as if the operator hadn't passed any include patterns +/// at all). +#[derive(Encode, Decode, Debug, Clone, Eq, PartialEq)] +pub struct ProverDbConsumerConfig { + /// HTTP base URL of the upstream prover-db indexer. Validated at + /// node start; stored as raw bytes so the pallet doesn't need to + /// pull in the `url` crate's full parsing surface. + pub url: String, + /// Per-node include set. An event is forwarded iff its + /// `TableIdentifier` matches at least one filter. Empty ⇒ forward + /// nothing; callers that want "forward everything" must include an + /// explicit `*.*` filter. + pub include: Vec, +} /// Offchain DB key prefix for per-extrinsic event payloads. SCALE-encoded /// as part of a `(prefix, block, ext_idx)` tuple, so the tuple structure @@ -98,3 +127,312 @@ pub trait EventCapture { impl EventCapture for () { fn capture_events(_events: Vec>) {} } + +/// Filter for one side (namespace or name) of a `TableIdentifier`. +/// Either matches every identifier (`Wildcard`) or matches exactly one +/// byte sequence (`Ident`). `Ident` values are stored uppercased so +/// matches against the on-chain canonical form are byte-exact without +/// per-callsite vigilance. +#[derive(Encode, Decode, Debug, Clone, Eq, PartialEq)] +pub enum IdentFilter { + /// Matches any identifier on this side of the dot. + Wildcard, + /// Matches an identifier whose bytes equal these. + Ident(String), +} + +impl IdentFilter { + /// Returns true if `bytes` (the on-chain canonical, uppercased form) + /// matches this filter. + pub fn matches(&self, bytes: &[u8]) -> bool { + match self { + Self::Wildcard => true, + Self::Ident(s) => s.as_bytes() == bytes, + } + } +} + +/// `FromStr` parse error for [`IdentFilter`]. +#[derive(Debug, Snafu)] +pub enum IdentFilterParseError { + /// The input was empty (neither `*` nor a non-empty identifier). + #[snafu(display("identifier filter is empty"))] + Empty, + /// The identifier exceeded the on-chain identifier length cap and so + /// could never match a captured table identifier. + #[snafu(display("identifier filter exceeds the {max}-byte on-chain length cap"))] + TooLong { + /// The cap, in bytes (`IDENT_LENGTH`). + max: u32, + }, +} + +impl FromStr for IdentFilter { + type Err = IdentFilterParseError; + + fn from_str(s: &str) -> Result { + if s == "*" { + return Ok(Self::Wildcard); + } + if s.is_empty() { + return Err(IdentFilterParseError::Empty); + } + let upper = s.to_uppercase(); + if upper.len() as u32 > IDENT_LENGTH { + return Err(IdentFilterParseError::TooLong { max: IDENT_LENGTH }); + } + Ok(Self::Ident(upper)) + } +} + +/// Filter against a fully-qualified [`TableIdentifier`]. Matches when +/// both sides match — so `*.*` is "everything", `NS.*` is "every table +/// in NS", `NS.NAME` is exact, and `*.NAME` is "any namespace, this +/// name". +/// +/// Stored only in the indexer node's offchain local storage (as the +/// `include` field of [`ProverDbConsumerConfig`]); not part of on-chain +/// state. An empty list matches nothing — see [`table_matches_filters`]. +#[derive(Encode, Decode, Debug, Clone, Eq, PartialEq)] +pub struct TableIdentifierFilter { + /// Filter applied to `ident.namespace`. + pub namespace_filter: IdentFilter, + /// Filter applied to `ident.name`. + pub name_filter: IdentFilter, +} + +impl TableIdentifierFilter { + /// Returns true if `ident` passes both sides of this filter. + pub fn matches(&self, ident: &TableIdentifier) -> bool { + self.namespace_filter.matches(&ident.namespace) && self.name_filter.matches(&ident.name) + } +} + +/// `FromStr` parse error for [`TableIdentifierFilter`]. +#[derive(Debug, Snafu)] +pub enum TableIdentifierFilterParseError { + /// The input didn't contain exactly one `.` separator. + #[snafu(display("expected 'NAMESPACE.NAME' form with exactly one dot, got '{input}'"))] + MalformedShape { + /// The original input string. + input: String, + }, + /// The namespace side failed to parse as an [`IdentFilter`]. + #[snafu(display("invalid namespace filter: {source}"))] + Namespace { + /// Underlying [`IdentFilter`] parse error. + source: IdentFilterParseError, + }, + /// The name side failed to parse as an [`IdentFilter`]. + #[snafu(display("invalid name filter: {source}"))] + Name { + /// Underlying [`IdentFilter`] parse error. + source: IdentFilterParseError, + }, +} + +impl FromStr for TableIdentifierFilter { + type Err = TableIdentifierFilterParseError; + + fn from_str(s: &str) -> Result { + let Some((ns, name)) = s.split_once('.') else { + return Err(TableIdentifierFilterParseError::MalformedShape { + input: s.to_string(), + }); + }; + if s.matches('.').count() != 1 { + return Err(TableIdentifierFilterParseError::MalformedShape { + input: s.to_string(), + }); + } + let namespace_filter = ns + .parse() + .map_err(|source| TableIdentifierFilterParseError::Namespace { source })?; + let name_filter = name + .parse() + .map_err(|source| TableIdentifierFilterParseError::Name { source })?; + Ok(Self { + namespace_filter, + name_filter, + }) + } +} + +/// Returns true if `table` matches at least one filter in `filters`. An +/// empty filter set matches nothing — callers that want "match all" +/// must pass an explicit `*.*` filter. This makes "no filters +/// configured" and "match-all configured" distinct, addressable states. +pub fn table_matches_filters(table: &TableIdentifier, filters: &[TableIdentifierFilter]) -> bool { + filters.iter().any(|f| f.matches(table)) +} + +#[cfg(test)] +mod tests { + use alloc::vec; + + use super::*; + + fn ident(name: &str, namespace: &str) -> TableIdentifier { + TableIdentifier::from_str_unchecked(name, namespace) + } + + // ── IdentFilter::matches ───────────────────────────────────────── + + #[test] + fn ident_filter_wildcard_matches_anything() { + assert!(IdentFilter::Wildcard.matches(b"")); + assert!(IdentFilter::Wildcard.matches(b"ANYTHING")); + } + + #[test] + fn ident_filter_ident_matches_exact_bytes_only() { + let f = IdentFilter::Ident("FOO".into()); + assert!(f.matches(b"FOO")); + assert!(!f.matches(b"foo")); // case-sensitive at match time + assert!(!f.matches(b"FOOBAR")); + assert!(!f.matches(b"FO")); + assert!(!f.matches(b"")); + } + + // ── IdentFilter::FromStr ───────────────────────────────────────── + + #[test] + fn ident_filter_parses_star_as_wildcard() { + assert_eq!("*".parse::().unwrap(), IdentFilter::Wildcard); + } + + #[test] + fn ident_filter_parses_ident_uppercased() { + assert_eq!( + "foo".parse::().unwrap(), + IdentFilter::Ident("FOO".into()), + ); + assert_eq!( + "MIXEDcase".parse::().unwrap(), + IdentFilter::Ident("MIXEDCASE".into()), + ); + } + + #[test] + fn ident_filter_rejects_empty_string() { + assert!(matches!( + "".parse::(), + Err(IdentFilterParseError::Empty), + )); + } + + #[test] + fn ident_filter_rejects_over_length_input() { + // IDENT_LENGTH + 1 ASCII bytes — uppercase is a no-op so the + // post-uppercase length is also IDENT_LENGTH + 1. + let too_long: alloc::string::String = + core::iter::repeat_n('A', IDENT_LENGTH as usize + 1).collect(); + assert!(matches!( + too_long.parse::(), + Err(IdentFilterParseError::TooLong { max }) if max == IDENT_LENGTH, + )); + } + + // ── TableIdentifierFilter::matches ─────────────────────────────── + + #[test] + fn table_filter_matches_only_when_both_sides_match() { + let table = ident("T1", "ALPHA"); + + let alpha_t1: TableIdentifierFilter = "ALPHA.T1".parse().unwrap(); + let alpha_star: TableIdentifierFilter = "ALPHA.*".parse().unwrap(); + let star_t1: TableIdentifierFilter = "*.T1".parse().unwrap(); + let star_star: TableIdentifierFilter = "*.*".parse().unwrap(); + let beta_t1: TableIdentifierFilter = "BETA.T1".parse().unwrap(); + let alpha_other: TableIdentifierFilter = "ALPHA.OTHER".parse().unwrap(); + + assert!(alpha_t1.matches(&table)); + assert!(alpha_star.matches(&table)); + assert!(star_t1.matches(&table)); + assert!(star_star.matches(&table)); + assert!(!beta_t1.matches(&table)); + assert!(!alpha_other.matches(&table)); + } + + // ── TableIdentifierFilter::FromStr ─────────────────────────────── + + #[test] + fn table_filter_parses_all_four_shapes() { + let full: TableIdentifierFilter = "ns.name".parse().unwrap(); + assert!(matches!(full.namespace_filter, IdentFilter::Ident(ref s) if s == "NS")); + assert!(matches!(full.name_filter, IdentFilter::Ident(ref s) if s == "NAME")); + + let ns_star: TableIdentifierFilter = "ns.*".parse().unwrap(); + assert!(matches!(ns_star.namespace_filter, IdentFilter::Ident(ref s) if s == "NS")); + assert!(matches!(ns_star.name_filter, IdentFilter::Wildcard)); + + let star_name: TableIdentifierFilter = "*.name".parse().unwrap(); + assert!(matches!(star_name.namespace_filter, IdentFilter::Wildcard)); + assert!(matches!(star_name.name_filter, IdentFilter::Ident(ref s) if s == "NAME")); + + let star_star: TableIdentifierFilter = "*.*".parse().unwrap(); + assert!(matches!(star_star.namespace_filter, IdentFilter::Wildcard)); + assert!(matches!(star_star.name_filter, IdentFilter::Wildcard)); + } + + #[test] + fn table_filter_rejects_missing_dot() { + assert!(matches!( + "namename".parse::(), + Err(TableIdentifierFilterParseError::MalformedShape { .. }), + )); + } + + #[test] + fn table_filter_rejects_more_than_one_dot() { + assert!(matches!( + "a.b.c".parse::(), + Err(TableIdentifierFilterParseError::MalformedShape { .. }), + )); + } + + #[test] + fn table_filter_propagates_empty_side_error() { + assert!(matches!( + ".name".parse::(), + Err(TableIdentifierFilterParseError::Namespace { + source: IdentFilterParseError::Empty, + }), + )); + assert!(matches!( + "ns.".parse::(), + Err(TableIdentifierFilterParseError::Name { + source: IdentFilterParseError::Empty, + }), + )); + } + + // ── table_matches_filters ──────────────────────────────────────── + + #[test] + fn empty_filter_set_matches_nothing() { + let table = ident("T1", "ALPHA"); + assert!(!table_matches_filters(&table, &[])); + } + + #[test] + fn explicit_wildcard_filter_matches_everything() { + let table = ident("T1", "ALPHA"); + let star: TableIdentifierFilter = "*.*".parse().unwrap(); + assert!(table_matches_filters(&table, core::slice::from_ref(&star))); + } + + #[test] + fn non_empty_filter_set_requires_at_least_one_match() { + let table = ident("T1", "ALPHA"); + let filters = vec![ + "BETA.*".parse().unwrap(), + "GAMMA.T1".parse().unwrap(), + "ALPHA.T1".parse().unwrap(), // this one matches + ]; + assert!(table_matches_filters(&table, &filters)); + + let no_match = vec!["BETA.*".parse().unwrap(), "GAMMA.T1".parse().unwrap()]; + assert!(!table_matches_filters(&table, &no_match)); + } +}