fix(bdk_electrum_streaming): Keep the highest last active index per update

bdk_electrum_streaming/src/state.rs

@@ -467,9 +467,18 @@ impl<PReq: PendingRequest, K: Ord + Clone> State<PReq, K> {
                     Some((spk_hash, tx_update)) => {
                         let update = update.get_or_insert(Update::default());
                         update.tx_update.extend(tx_update);
-                        update
-                            .last_active_indices
-                            .extend(self.spk_tracker.index_of_spk_hash(spk_hash));
+                        if let Some((keychain, index)) =
+                            self.spk_tracker.index_of_spk_hash(spk_hash)
+                        {
+                            // Jobs of the same keychain can finish in the same update (they share
+                            // requests), so keep the highest index. Overwriting would under-report
+                            // the last active index and leave the higher spk unrevealed.
+                            update
+                                .last_active_indices
+                                .entry(keychain)
+                                .and_modify(|last| *last = (*last).max(index))
+                                .or_insert(index);
+                        }
                     }
                     None => {
                         self.spk_jobs.insert(spk_hash, job);

bdk_electrum_streaming/tests/state.rs

@@ -1,4 +1,4 @@
-use std::{str::FromStr, sync::Arc};
+use std::{collections::BTreeMap, str::FromStr, sync::Arc};
 
 use bdk_core::{
     bitcoin::{
@@ -23,9 +23,21 @@ fn raw_msg(v: serde_json::Value) -> RawNotificationOrResponse {
     serde_json::from_value(v).expect("must deserialize raw message")
 }
 
+/// The script status of a history containing a single tx confirmed at `height`.
+fn confirmed_status(txid: Txid, height: u32) -> anyhow::Result<Option<ElectrumScriptStatus>> {
+    Ok(ElectrumScriptStatus::from_history(&[
+        response::Tx::Confirmed(response::ConfirmedTx {
+            txid,
+            height: absolute::Height::from_consensus(height)?,
+        }),
+    ]))
+}
+
 struct Server {
     headers: Vec<block::Header>,
-    spk_hash: ElectrumScriptHash,
+    /// Script hashes that [`Server::tx`] pays to, mapped to the status answered for their
+    /// subscriptions. Every other script hash answers a `null` status and an empty history.
+    active_spks: BTreeMap<ElectrumScriptHash, Option<ElectrumScriptStatus>>,
     tx: Transaction,
 }
 
@@ -34,6 +46,21 @@ impl Server {
         self.headers.len() - 1
     }
 
+    /// Whether the request's script hash is one that [`Server::tx`] pays to.
+    fn is_active(&self, req: &RawRequest) -> bool {
+        self.active_spks
+            .keys()
+            .any(|spk_hash| req.params[0] == json!(spk_hash.to_string()))
+    }
+
+    /// The status to answer a subscription of the request's script hash with.
+    fn status_of(&self, req: &RawRequest) -> Option<ElectrumScriptStatus> {
+        self.active_spks
+            .iter()
+            .find(|(spk_hash, _)| req.params[0] == json!(spk_hash.to_string()))
+            .and_then(|(_, status)| *status)
+    }
+
     fn answer(&self, req: &RawRequest) -> serde_json::Value {
         match req.method.as_ref() {
             "blockchain.headers.subscribe" => {
@@ -49,9 +76,12 @@ impl Server {
                     .collect::<String>();
                 json!({ "count": count, "hex": hex, "max": 2016 })
             }
-            "blockchain.scripthash.subscribe" => json!(null),
+            "blockchain.scripthash.subscribe" => match self.status_of(req) {
+                Some(status) => json!(status.to_string()),
+                None => json!(null),
+            },
             "blockchain.scripthash.get_history" => {
-                if req.params[0] == json!(self.spk_hash.to_string()) && self.tip_height() >= 2 {
+                if self.is_active(req) && self.tip_height() >= 2 {
                     json!([{ "tx_hash": self.tx.compute_txid().to_string(), "height": 2 }])
                 } else {
                     json!([])
@@ -144,7 +174,7 @@ fn anchor_above_local_tip_is_deferred_until_tip_catches_up() -> anyhow::Result<(
     let mut queue = ReqQueue::new();
     let mut server = Server {
         headers: vec![genesis, header_1],
-        spk_hash,
+        active_spks: [(spk_hash, None)].into_iter().collect(),
         tx,
     };
 
@@ -234,3 +264,186 @@ fn descriptor_inserted_mid_connection_is_subscribed() -> anyhow::Result<()> {
     );
     Ok(())
 }
+
+/// `last_active_indices` must report the derivation index of the spk that has history, not the
+/// index after it. Reporting `index + 1` reveals one spk too many on every sync, permanently
+/// skipping an unused address.
+#[test]
+fn last_active_index_is_index_of_active_spk() -> anyhow::Result<()> {
+    const ACTIVE_INDEX: u32 = 3;
+    const LOOKAHEAD: u32 = 5;
+
+    let descriptor = Descriptor::<DescriptorPublicKey>::from_str(&format!("wpkh({XPUB}/0/*)"))?;
+    let spk = descriptor
+        .at_derivation_index(ACTIVE_INDEX)?
+        .script_pubkey();
+    let spk_hash = ElectrumScriptHash::new(&spk);
+
+    let tx = Transaction {
+        version: transaction::Version::ONE,
+        lock_time: absolute::LockTime::ZERO,
+        input: vec![TxIn {
+            previous_output: OutPoint::null(),
+            script_sig: ScriptBuf::new(),
+            sequence: Sequence::MAX,
+            witness: Witness::new(),
+        }],
+        output: vec![TxOut {
+            value: Amount::from_sat(50_000),
+            script_pubkey: spk,
+        }],
+    };
+    let txid = tx.compute_txid();
+
+    let genesis = constants::genesis_block(Network::Regtest).header;
+    let header_1 = block::Header {
+        version: block::Version::ONE,
+        prev_blockhash: genesis.block_hash(),
+        merkle_root: TxMerkleNode::all_zeros(),
+        time: 100,
+        bits: CompactTarget::from_consensus(0x207fffff),
+        nonce: 0,
+    };
+    // With the tx at position 0 of a single-tx block, the merkle root is its txid.
+    let header_2 = block::Header {
+        merkle_root: Txid::to_raw_hash(txid).into(),
+        prev_blockhash: header_1.block_hash(),
+        time: 200,
+        ..header_1
+    };
+
+    let mut spk_tracker = DerivedSpkTracker::new(LOOKAHEAD);
+    spk_tracker.insert_descriptor("external", descriptor, 0);
+
+    let mut state = BlockingState::new(
+        ReqCoord::default(),
+        Cache::default(),
+        spk_tracker,
+        CheckPoint::new(BlockId {
+            height: 0,
+            hash: genesis.block_hash(),
+        }),
+    );
+    let mut queue = ReqQueue::new();
+    let server = Server {
+        headers: vec![genesis, header_1, header_2],
+        active_spks: [(spk_hash, confirmed_status(txid, 2)?)]
+            .into_iter()
+            .collect(),
+        tx,
+    };
+
+    state.init(&mut queue);
+    let updates = drain_requests(&mut state, &mut queue, &server);
+
+    let emitted = updates
+        .iter()
+        .flat_map(|update| &update.last_active_indices)
+        .map(|(&k, &i)| (k, i))
+        .collect::<Vec<_>>();
+    assert!(
+        !emitted.is_empty(),
+        "an spk with history must emit a last active index"
+    );
+    for (keychain, index) in emitted {
+        assert_eq!(
+            (keychain, index),
+            ("external", ACTIVE_INDEX),
+            "only the spk with history is active"
+        );
+    }
+    Ok(())
+}
+
+/// A single tx can pay multiple spks of the same keychain. Their jobs share requests, so they
+/// finish together in one update, and that update must report the *highest* active index.
+/// Reporting a lower one leaves the higher spk unrevealed, so the wallet does not recognise its
+/// txouts as its own.
+#[test]
+fn last_active_index_is_highest_of_batch() -> anyhow::Result<()> {
+    // `4`'s script hash sorts below `3`'s, so a job iteration ordered by script hash reaches the
+    // higher derivation index first.
+    const ACTIVE_INDICES: [u32; 2] = [3, 4];
+    const LOOKAHEAD: u32 = 5;
+
+    let descriptor = Descriptor::<DescriptorPublicKey>::from_str(&format!("wpkh({XPUB}/0/*)"))?;
+    let spks = ACTIVE_INDICES
+        .iter()
+        .map(|&index| Ok(descriptor.at_derivation_index(index)?.script_pubkey()))
+        .collect::<anyhow::Result<Vec<_>>>()?;
+
+    let tx = Transaction {
+        version: transaction::Version::ONE,
+        lock_time: absolute::LockTime::ZERO,
+        input: vec![TxIn {
+            previous_output: OutPoint::null(),
+            script_sig: ScriptBuf::new(),
+            sequence: Sequence::MAX,
+            witness: Witness::new(),
+        }],
+        output: spks
+            .iter()
+            .map(|spk| TxOut {
+                value: Amount::from_sat(50_000),
+                script_pubkey: spk.clone(),
+            })
+            .collect(),
+    };
+    let txid = tx.compute_txid();
+
+    let genesis = constants::genesis_block(Network::Regtest).header;
+    let header_1 = block::Header {
+        version: block::Version::ONE,
+        prev_blockhash: genesis.block_hash(),
+        merkle_root: TxMerkleNode::all_zeros(),
+        time: 100,
+        bits: CompactTarget::from_consensus(0x207fffff),
+        nonce: 0,
+    };
+    // With the tx at position 0 of a single-tx block, the merkle root is its txid.
+    let header_2 = block::Header {
+        merkle_root: Txid::to_raw_hash(txid).into(),
+        prev_blockhash: header_1.block_hash(),
+        time: 200,
+        ..header_1
+    };
+
+    let mut spk_tracker = DerivedSpkTracker::new(LOOKAHEAD);
+    spk_tracker.insert_descriptor("external", descriptor, 0);
+
+    let mut state = BlockingState::new(
+        ReqCoord::default(),
+        Cache::default(),
+        spk_tracker,
+        CheckPoint::new(BlockId {
+            height: 0,
+            hash: genesis.block_hash(),
+        }),
+    );
+    let mut queue = ReqQueue::new();
+    let status = confirmed_status(txid, 2)?;
+    let server = Server {
+        headers: vec![genesis, header_1, header_2],
+        active_spks: spks
+            .iter()
+            .map(|spk| (ElectrumScriptHash::new(spk), status))
+            .collect(),
+        tx,
+    };
+
+    state.init(&mut queue);
+    let updates = drain_requests(&mut state, &mut queue, &server);
+
+    let highest_emitted = updates
+        .iter()
+        .flat_map(|update| &update.last_active_indices)
+        .filter(|(&keychain, _)| keychain == "external")
+        .map(|(_, &index)| index)
+        .max();
+    assert_eq!(
+        highest_emitted,
+        ACTIVE_INDICES.iter().copied().max(),
+        "the highest active index must not be lost to a lower one in the same update"
+    );
+    Ok(())
+}