feat: add configurable include set to gate captured tables

node/src/cli.rs

@@ -19,10 +19,12 @@ pub struct Cli {
     #[clap(long)]
     pub event_forwarder_rpc: Option<String>,
 
-    /// 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<url::Url>,
+    /// 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<url::Url>,
+
+    /// 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<sxt_core::prover_db_indexer::TableIdentifierFilter>,
+}
+
 #[derive(Debug, clap::Subcommand)]
 #[allow(clippy::large_enum_variant)]
 pub enum Subcommand {

node/src/service.rs

@@ -77,28 +77,43 @@ pub type TransactionPool = sc_transaction_pool::FullPool<Block, FullClient>;
 /// 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<u8>` 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<N: NetworkBackend<Block, <Block as BlockT>::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),

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<Vec<u8>> = url_ref.get::<Vec<u8>>().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::<ProverDbConsumerConfig>().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<TableIdentifierFilter> = 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,23 +350,37 @@ 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);
             }
 
             crate::offchain_consumer::clear_high_water(block_num);
             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)