diff --git a/bdk_electrum_streaming/src/state.rs b/bdk_electrum_streaming/src/state.rs index 49d3e2e..b4bc638 100644 --- a/bdk_electrum_streaming/src/state.rs +++ b/bdk_electrum_streaming/src/state.rs @@ -467,9 +467,18 @@ impl State { 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); diff --git a/bdk_electrum_streaming/tests/state.rs b/bdk_electrum_streaming/tests/state.rs index f95fb27..c96427b 100644 --- a/bdk_electrum_streaming/tests/state.rs +++ b/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> { + Ok(ElectrumScriptStatus::from_history(&[ + response::Tx::Confirmed(response::ConfirmedTx { + txid, + height: absolute::Height::from_consensus(height)?, + }), + ])) +} + struct Server { headers: Vec, - 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>, 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 { + 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::(); 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::::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::>(); + 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::::from_str(&format!("wpkh({XPUB}/0/*)"))?; + let spks = ACTIVE_INDICES + .iter() + .map(|&index| Ok(descriptor.at_derivation_index(index)?.script_pubkey())) + .collect::>>()?; + + 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(()) +}