fix(spv): match masternode owner/voting key hashes in compact filters

dash-spv/src/sync/filters/manager.rs

@@ -23,12 +23,25 @@ use crate::validation::{FilterValidationInput, FilterValidator, Validator};
 use crate::sync::progress::ProgressPercentage;
 use dashcore::hash_types::FilterHeader;
 use key_wallet_manager::WalletInterface;
-use key_wallet_manager::{check_compact_filters_for_script_pubkeys, FilterMatchKey, WalletId};
+use key_wallet_manager::{check_compact_filters_for_elements, FilterMatchKey, WalletId};
 use tokio::sync::RwLock;
 
 /// Batch size for processing filters.
 const BATCH_PROCESSING_SIZE: u32 = 5000;
 
+/// Snapshot of a behind wallet's compact-filter query inputs for a batch scan.
+struct WalletScanState {
+    /// The wallet these inputs belong to.
+    id: WalletId,
+    /// The wallet's committed sync checkpoint; heights at or below it are skipped.
+    synced: u32,
+    /// Monitored scriptPubKeys.
+    scripts: Vec<ScriptBuf>,
+    /// Bare `hash160` filter elements (owner/voting key hashes) a compact
+    /// filter carries beyond the scriptPubKeys.
+    elements: Vec<Vec<u8>>,
+}
+
 /// Maximum number of batches to scan ahead while waiting for blocks.
 const MAX_LOOKAHEAD_BATCHES: usize = 3;
 
@@ -646,21 +659,33 @@ impl<H: BlockHeaderStorage, FH: FilterHeaderStorage, F: FilterStorage, W: Wallet
         let batch_filters = batch.filters();
 
         // Per-wallet `synced_height` snapshot so heights below the wallet's
-        // own progress are skipped during the rescan.
-        let synced_heights: HashMap<WalletId, u32> = {
+        // own progress are skipped during the rescan, plus the bare
+        // owner/voting filter elements a compact filter carries beyond the
+        // wallet's scriptPubKeys.
+        let mut synced_heights: HashMap<WalletId, u32> = HashMap::new();
+        let mut filter_elements: HashMap<WalletId, Vec<Vec<u8>>> = HashMap::new();
+        {
             let wallet = self.wallet.read().await;
-            new_scripts.keys().map(|id| (*id, wallet.wallet_synced_height(id))).collect()
-        };
+            for id in new_scripts.keys() {
+                synced_heights.insert(*id, wallet.wallet_synced_height(id));
+                filter_elements.insert(*id, wallet.monitored_filter_elements_for(id));
+            }
+        }
 
         let mut block_to_wallets: BTreeMap<FilterMatchKey, BTreeSet<WalletId>> = BTreeMap::new();
         for (wallet_id, scripts) in new_scripts {
-            if scripts.is_empty() {
+            let elements = filter_elements.get(wallet_id).map(Vec::as_slice).unwrap_or(&[]);
+            if scripts.is_empty() && elements.is_empty() {
                 continue;
             }
             let scripts_vec: Vec<ScriptBuf> = scripts.iter().cloned().collect();
             let min_synced = synced_heights.get(wallet_id).copied().unwrap_or(0);
-            let matches =
-                check_compact_filters_for_script_pubkeys(batch_filters, &scripts_vec, min_synced);
+            let matches = check_compact_filters_for_elements(
+                batch_filters,
+                &scripts_vec,
+                elements,
+                min_synced,
+            );
             for key in matches {
                 block_to_wallets.entry(key).or_default().insert(*wallet_id);
             }
@@ -742,12 +767,20 @@ impl<H: BlockHeaderStorage, FH: FilterHeaderStorage, F: FilterStorage, W: Wallet
         // filters.
         let wallet = self.wallet.read().await;
         let behind = wallet.wallets_behind(batch_end);
-        let mut wallet_states: Vec<(WalletId, u32, Vec<ScriptBuf>)> = Vec::new();
+        let mut wallet_states: Vec<WalletScanState> = Vec::new();
         for wallet_id in &behind {
             let synced = wallet.wallet_synced_height(wallet_id);
             let scripts = wallet.monitored_script_pubkeys_for(wallet_id);
-            if !scripts.is_empty() {
-                wallet_states.push((*wallet_id, synced, scripts));
+            // Bare owner/voting key hashes a compact filter carries beyond the
+            // wallet's scriptPubKeys.
+            let elements = wallet.monitored_filter_elements_for(wallet_id);
+            if !scripts.is_empty() || !elements.is_empty() {
+                wallet_states.push(WalletScanState {
+                    id: *wallet_id,
+                    synced,
+                    scripts,
+                    elements,
+                });
             }
         }
         drop(wallet);
@@ -777,17 +810,25 @@ impl<H: BlockHeaderStorage, FH: FilterHeaderStorage, F: FilterStorage, W: Wallet
         }
 
         // Single-pass union-then-attribute: build the union of all scripts
-        // across behind wallets, run the filters once, then for each matched
-        // block re-test per-wallet scripts to attribute the match correctly.
+        // and bare elements across behind wallets, run the filters once, then
+        // for each matched block re-test per-wallet queries to attribute the
+        // match correctly.
         let union_scripts: Vec<ScriptBuf> =
-            wallet_states.iter().flat_map(|(_, _, scripts)| scripts.iter().cloned()).collect();
-        let min_synced = wallet_states.iter().map(|(_, synced, _)| *synced).min().unwrap_or(0);
+            wallet_states.iter().flat_map(|s| s.scripts.iter().cloned()).collect();
+        let union_elements: Vec<Vec<u8>> =
+            wallet_states.iter().flat_map(|s| s.elements.iter().cloned()).collect();
+        let min_synced = wallet_states.iter().map(|s| s.synced).min().unwrap_or(0);
 
-        // Pre-group each wallet's scripts by length once; reused across every matched filter.
+        // Pre-group each wallet's scripts and bare elements by length once;
+        // reused across every matched filter.
         let wallet_queries: Vec<(WalletId, u32, FilterQuery)> = wallet_states
             .iter()
-            .map(|(id, synced, scripts)| {
-                (*id, *synced, scripts.iter().map(|s| s.as_bytes()).collect())
+            .map(|s| {
+                let mut query: FilterQuery = s.scripts.iter().map(|sp| sp.as_bytes()).collect();
+                for element in &s.elements {
+                    query.push(element);
+                }
+                (s.id, s.synced, query)
             })
             .collect();
 
@@ -797,8 +838,12 @@ impl<H: BlockHeaderStorage, FH: FilterHeaderStorage, F: FilterStorage, W: Wallet
             };
             let batch_filters = batch.filters();
 
-            let matches =
-                check_compact_filters_for_script_pubkeys(batch_filters, &union_scripts, min_synced);
+            let matches = check_compact_filters_for_elements(
+                batch_filters,
+                &union_scripts,
+                &union_elements,
+                min_synced,
+            );
             let mut block_to_wallets: BTreeMap<FilterMatchKey, BTreeSet<WalletId>> =
                 BTreeMap::new();
             for key in matches {

key-wallet-manager/src/lib.rs

@@ -21,7 +21,7 @@ mod wallet_interface;
 
 pub use error::WalletError;
 pub use events::{DerivedAddress, WalletEvent};
-pub use matching::{check_compact_filters_for_script_pubkeys, FilterMatchKey};
+pub use matching::{check_compact_filters_for_elements, FilterMatchKey};
 pub use wallet_interface::{BlockProcessingResult, MempoolTransactionResult, WalletInterface};
 
 use dashcore::blockdata::transaction::Transaction;

key-wallet-manager/src/matching.rs

@@ -26,18 +26,28 @@ impl FilterMatchKey {
     }
 }
 
-/// Check compact filters against a set of scriptPubKeys and return the keys
-/// that matched.
+/// Check compact filters against a set of scriptPubKeys plus any bare
+/// filter elements, returning the keys that matched.
+///
+/// `extra_elements` carries arbitrary-length elements a peer inserts into the
+/// compact filter that are not scriptPubKeys, notably the provider owner/voting
+/// key hashes (bare 20-byte `hash160`) from a `ProRegTx`. They fold into the
+/// same [`FilterQuery`] as the scripts, landing in its `other` length bucket.
 ///
 /// Entries with `key.height() <= min_height` are skipped. Pass `0` to test
 /// every filter in the input.
-pub fn check_compact_filters_for_script_pubkeys(
+pub fn check_compact_filters_for_elements(
     input: &HashMap<FilterMatchKey, BlockFilter>,
     script_pubkeys: &[ScriptBuf],
+    extra_elements: &[Vec<u8>],
     min_height: CoreBlockHeight,
 ) -> BTreeSet<FilterMatchKey> {
-    // Group the scripts by length once; the same query is reused across every filter.
-    let query: FilterQuery = script_pubkeys.iter().map(|s| s.as_bytes()).collect();
+    // Group the scripts and bare elements by length once; the same query is
+    // reused across every filter.
+    let mut query: FilterQuery = script_pubkeys.iter().map(|s| s.as_bytes()).collect();
+    for element in extra_elements {
+        query.push(element);
+    }
     let match_filter = |(key, filter): (&FilterMatchKey, &BlockFilter)| {
         if key.height() <= min_height {
             return None;
@@ -70,16 +80,26 @@ pub fn check_compact_filters_for_script_pubkeys(
 #[cfg(test)]
 mod tests {
     use super::*;
+    use dashcore::address::Payload;
+    use dashcore::bip158::BlockFilterWriter;
     use dashcore::{Address, Block, Transaction};
     use key_wallet::Network;
 
     fn scripts_for(addresses: &[Address]) -> Vec<ScriptBuf> {
         addresses.iter().map(|a| a.script_pubkey()).collect()
     }
 
+    /// The bare 20-byte `hash160` a peer inserts for a P2PKH address.
+    fn hash160_of(address: &Address) -> Vec<u8> {
+        match address.payload() {
+            Payload::PubkeyHash(hash) => <[u8; 20]>::from(*hash).to_vec(),
+            other => panic!("expected P2PKH dummy address, got {other:?}"),
+        }
+    }
+
     #[test]
     fn test_empty_input_returns_empty() {
-        let result = check_compact_filters_for_script_pubkeys(&HashMap::new(), &[], 0);
+        let result = check_compact_filters_for_elements(&HashMap::new(), &[], &[], 0);
         assert!(result.is_empty());
     }
 
@@ -94,7 +114,7 @@ mod tests {
         let mut input = HashMap::new();
         input.insert(key.clone(), filter);
 
-        let output = check_compact_filters_for_script_pubkeys(&input, &[], 0);
+        let output = check_compact_filters_for_elements(&input, &[], &[], 0);
         assert!(!output.contains(&key));
     }
 
@@ -110,7 +130,7 @@ mod tests {
         input.insert(key.clone(), filter);
 
         let scripts = scripts_for(&[address]);
-        let output = check_compact_filters_for_script_pubkeys(&input, &scripts, 0);
+        let output = check_compact_filters_for_elements(&input, &scripts, &[], 0);
         assert!(output.contains(&key));
     }
 
@@ -128,7 +148,7 @@ mod tests {
         input.insert(key.clone(), filter);
 
         let scripts = scripts_for(&[address]);
-        let output = check_compact_filters_for_script_pubkeys(&input, &scripts, 0);
+        let output = check_compact_filters_for_elements(&input, &scripts, &[], 0);
         assert!(!output.contains(&key));
     }
 
@@ -159,7 +179,7 @@ mod tests {
         input.insert(key_3.clone(), filter_3);
 
         let scripts = scripts_for(&[address_1, address_2]);
-        let output = check_compact_filters_for_script_pubkeys(&input, &scripts, 0);
+        let output = check_compact_filters_for_elements(&input, &scripts, &[], 0);
         assert_eq!(output.len(), 2);
         assert!(output.contains(&key_1));
         assert!(output.contains(&key_2));
@@ -183,7 +203,7 @@ mod tests {
         }
 
         let scripts = scripts_for(&[address]);
-        let output = check_compact_filters_for_script_pubkeys(&input, &scripts, 0);
+        let output = check_compact_filters_for_elements(&input, &scripts, &[], 0);
 
         // Verify output is sorted by height (ascending)
         let heights_out: Vec<CoreBlockHeight> = output.iter().map(|k| k.height()).collect();
@@ -193,4 +213,47 @@ mod tests {
         assert_eq!(heights_out, sorted_heights);
         assert_eq!(heights_out, vec![100, 200, 300, 400, 500]);
     }
+
+    /// A masternode owner watching only the owner/voting key sees a compact
+    /// filter that carries that key as a bare 20-byte `hash160` (the way Dash
+    /// Core's `ExtractSpecialTxFilterElements` inserts it), not as a P2PKH
+    /// script. The scripts-only query must miss it and the bare-element query
+    /// must hit it, which is exactly the gap this change closes.
+    #[test]
+    fn test_bare_owner_key_hash_requires_extra_element() {
+        // The owner key hash a peer inserts as a bare element.
+        let owner_address = Address::dummy(Network::Regtest, 7);
+        let owner_hash = hash160_of(&owner_address);
+        assert_eq!(owner_hash.len(), 20, "provider owner key is a 20-byte hash160");
+
+        // An unrelated output so the filter is realistic.
+        let unrelated = Address::dummy(Network::Regtest, 99);
+        let tx = Transaction::dummy(&unrelated, 0..0, &[1]);
+        let block = Block::dummy(100, vec![tx]);
+
+        // Build the filter like a Dash Core peer: block output scripts plus the
+        // owner key hash as a bare element.
+        let mut content = Vec::new();
+        {
+            let mut writer = BlockFilterWriter::new(&mut content, &block);
+            writer.add_output_scripts();
+            writer.add_element(&owner_hash);
+            writer.finish().expect("finish filter");
+        }
+        let filter = BlockFilter::new(&content);
+        let key = FilterMatchKey::new(100, block.block_hash());
+
+        let mut input = HashMap::new();
+        input.insert(key.clone(), filter);
+
+        // The P2PKH scriptPubKey form of the owner address does not match the
+        // bare 20-byte element.
+        let owner_script = scripts_for(&[owner_address]);
+        let scripts_only = check_compact_filters_for_elements(&input, &owner_script, &[], 0);
+        assert!(!scripts_only.contains(&key), "scripts-only query must miss the bare hash");
+
+        // Carrying the bare hash in `extra_elements` matches.
+        let with_element = check_compact_filters_for_elements(&input, &[], &[owner_hash], 0);
+        assert!(with_element.contains(&key), "bare-element query must match");
+    }
 }

key-wallet-manager/src/process_block.rs

@@ -228,6 +228,13 @@ impl<T: WalletInfoInterface + Send + Sync + 'static> WalletInterface for WalletM
             .unwrap_or_default()
     }
 
+    fn monitored_filter_elements_for(&self, wallet_id: &WalletId) -> Vec<Vec<u8>> {
+        self.wallet_infos
+            .get(wallet_id)
+            .map(|info| info.monitored_filter_elements())
+            .unwrap_or_default()
+    }
+
     fn watched_outpoints(&self) -> Vec<dashcore::OutPoint> {
         self.watched_outpoints()
     }

key-wallet-manager/src/wallet_interface.rs

@@ -88,6 +88,18 @@ pub trait WalletInterface: Send + Sync + 'static {
     /// Get cached scriptPubKeys for every address monitored by `wallet_id`.
     fn monitored_script_pubkeys_for(&self, wallet_id: &WalletId) -> Vec<ScriptBuf>;
 
+    /// Get the bare `hash160` compact-filter elements monitored by `wallet_id`
+    /// that are not covered by its scriptPubKeys.
+    ///
+    /// These are the provider owner/voting key hashes a Dash Core peer inserts
+    /// into a block's BIP158 compact filter as bare 20-byte elements, so a
+    /// masternode owner watching only those keys can match the block. The
+    /// default returns an empty set for implementations that monitor scripts
+    /// only.
+    fn monitored_filter_elements_for(&self, _wallet_id: &WalletId) -> Vec<Vec<u8>> {
+        Vec::new()
+    }
+
     /// Get all outpoints the wallet is watching (unspent outputs).
     /// Used for bloom filter construction to detect spends of our UTXOs.
     fn watched_outpoints(&self) -> Vec<OutPoint>;