From b3a026a900bb910fab3acac9dbcfab764e273ddf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BF=97=E5=AE=87?= Date: Fri, 21 Aug 2026 01:51:36 +0000 Subject: [PATCH 1/9] fix(bdk_electrum_streaming): Correct two SpkJob logging faults `try_finish` had its two log messages on the wrong branches, reporting "not finished" on completion and vice versa. `elapsed_seconds` subtracted without saturating, so a backwards clock step would panic a log line. --- bdk_electrum_streaming/src/spk_job.rs | 16 +- bdk_electrum_streaming/tests/state.rs | 231 ++++++++++++++++++++------ 2 files changed, 186 insertions(+), 61 deletions(-) diff --git a/bdk_electrum_streaming/src/spk_job.rs b/bdk_electrum_streaming/src/spk_job.rs index af45309..cdfcf50 100644 --- a/bdk_electrum_streaming/src/spk_job.rs +++ b/bdk_electrum_streaming/src/spk_job.rs @@ -107,10 +107,10 @@ impl SpkJob { } pub fn elapsed_seconds(&self) -> String { - let duration = UNIX_EPOCH.elapsed().expect("must get current timestamp") - self.start; - let seconds = duration.as_secs(); - let subsec = duration.subsec_millis(); - format!("{seconds}s {subsec}ms") + let now = UNIX_EPOCH.elapsed().expect("must get current timestamp"); + // The system clock can step backwards, which must not bring a log line down with it. + let duration = now.saturating_sub(self.start); + format!("{}s {}ms", duration.as_secs(), duration.subsec_millis()) } /// Try fullfill all that is missing. @@ -144,17 +144,17 @@ impl SpkJob { pub fn try_finish(&mut self) -> Option<(ElectrumScriptHash, TxUpdate)> { if self.stage.is_done() { - tracing::trace!( + tracing::info!( elapsed_seconds = self.elapsed_seconds(), spk_hash = self.spk_hash.to_string(), - "Spk job not finished" + "Spk job finished" ); Some((self.spk_hash, core::mem::take(&mut self.tx_update))) } else { - tracing::info!( + tracing::trace!( elapsed_seconds = self.elapsed_seconds(), spk_hash = self.spk_hash.to_string(), - "Spk job finished" + "Spk job not finished" ); None } diff --git a/bdk_electrum_streaming/tests/state.rs b/bdk_electrum_streaming/tests/state.rs index f95fb27..fc3d150 100644 --- a/bdk_electrum_streaming/tests/state.rs +++ b/bdk_electrum_streaming/tests/state.rs @@ -2,9 +2,12 @@ use std::{str::FromStr, sync::Arc}; use bdk_core::{ bitcoin::{ - absolute, block, consensus::encode::serialize_hex, constants, hashes::Hash, transaction, - Amount, CompactTarget, Network, OutPoint, ScriptBuf, Sequence, Transaction, TxIn, - TxMerkleNode, TxOut, Txid, Witness, + absolute, block, + consensus::encode::serialize_hex, + constants, + hashes::{sha256d, Hash}, + transaction, Amount, CompactTarget, Network, OutPoint, ScriptBuf, Sequence, Transaction, + TxIn, TxMerkleNode, TxOut, Txid, Witness, }, BlockId, CheckPoint, ConfirmationBlockTime, }; @@ -23,10 +26,14 @@ fn raw_msg(v: serde_json::Value) -> RawNotificationOrResponse { serde_json::from_value(v).expect("must deserialize raw message") } +#[derive(Clone)] struct Server { headers: Vec, spk_hash: ElectrumScriptHash, - tx: Transaction, + /// The transactions in `spk_hash`'s history, and the height each is confirmed at. + txs: Vec<(Transaction, u32)>, + /// The merkle branch and position this server answers every merkle request with. + merkle_proof: (Vec, usize), } impl Server { @@ -34,8 +41,27 @@ impl Server { self.headers.len() - 1 } - fn answer(&self, req: &RawRequest) -> serde_json::Value { - match req.method.as_ref() { + /// The history of `spk_hash`: every tx the chain is long enough to contain. + fn history(&self, spk_hash: &serde_json::Value) -> Vec { + if *spk_hash != json!(self.spk_hash.to_string()) { + return Vec::new(); + } + self.txs + .iter() + .filter(|(_, height)| self.tip_height() >= *height as usize) + .map(|(tx, height)| { + response::Tx::Confirmed(response::ConfirmedTx { + txid: tx.compute_txid(), + height: absolute::Height::from_consensus(*height) + .expect("must be a valid height"), + }) + }) + .collect() + } + + /// Answer a request, or fail it the way a server would. + fn answer(&self, req: &RawRequest) -> Result { + Ok(match req.method.as_ref() { "blockchain.headers.subscribe" => { let tip = self.headers.last().expect("server must have blocks"); json!({ "hex": serialize_hex(tip), "height": self.tip_height() }) @@ -49,20 +75,58 @@ impl Server { .collect::(); json!({ "count": count, "hex": hex, "max": 2016 }) } - "blockchain.scripthash.subscribe" => json!(null), - "blockchain.scripthash.get_history" => { - if req.params[0] == json!(self.spk_hash.to_string()) && self.tip_height() >= 2 { - json!([{ "tx_hash": self.tx.compute_txid().to_string(), "height": 2 }]) - } else { - json!([]) + "blockchain.block.header" => { + let height = req.params[0].as_u64().expect("must have height") as usize; + match self.headers.get(height) { + Some(header) => json!(serialize_hex(header)), + None => return Err(format!("height {height} is above the chain tip")), } } - "blockchain.transaction.get" => json!(serialize_hex(&self.tx)), + "blockchain.scripthash.subscribe" => { + match ElectrumScriptStatus::from_history(&self.history(&req.params[0])) { + Some(status) => json!(status.to_string()), + None => json!(null), + } + } + "blockchain.scripthash.get_history" => json!(self + .history(&req.params[0]) + .iter() + .map(|tx| json!({ + "tx_hash": tx.txid().to_string(), + "height": tx.electrum_height(), + })) + .collect::>()), + "blockchain.transaction.get" => { + let (tx, _) = self + .txs + .iter() + .find(|(tx, _)| req.params[0] == json!(tx.compute_txid().to_string())) + .expect("must be a tx the server knows"); + json!(serialize_hex(tx)) + } "blockchain.transaction.get_merkle" => { - json!({ "block_height": req.params[1], "merkle": [], "pos": 0 }) + // A proof can only be given for a tx the server has in that block. Both + // romanz/electrs and ElectrumX raise an error otherwise. + let height = req.params[1].as_u64().expect("must have height") as u32; + if !self + .txs + .iter() + .any(|(tx, h)| *h == height && req.params[0] == json!(tx.compute_txid())) + { + return Err(format!( + "tx {} not in block at height {height}", + req.params[0] + )); + } + let (branch, pos) = &self.merkle_proof; + json!({ + "block_height": req.params[1], + "merkle": branch.iter().map(|h| h.to_string()).collect::>(), + "pos": pos, + }) } other => panic!("unexpected request: {other}"), - } + }) } } @@ -73,27 +137,43 @@ fn drain_requests( ) -> Vec> { let mut updates = Vec::new(); while let Some(req) = queue.pop_front() { - let resp = - raw_msg(json!({ "jsonrpc": "2.0", "id": req.id, "result": server.answer(&req) })); - if let Some(update) = state.advance(queue, resp).expect("must advance") { + if let Some(update) = state + .advance(queue, response(&req, server)) + .expect("must advance") + { updates.push(update); } } updates } -/// A history response can report a confirmation height above the local tip: on a new block, -/// romanz/electrs notifies the script hash before the header and answers requests in order, so -/// the history arrives while the local tip is still one block behind. Such an anchor must be -/// deferred until the tip catches up — not dropped — and must still be delivered without any -/// further notification for that script. -#[test] -fn anchor_above_local_tip_is_deferred_until_tip_catches_up() -> anyhow::Result<()> { +/// The server's answer to `req`, as a raw JSON-RPC result or error message. +fn response(req: &RawRequest, server: &Server) -> RawNotificationOrResponse { + raw_msg(match server.answer(req) { + Ok(result) => json!({ "jsonrpc": "2.0", "id": req.id, "result": result }), + Err(message) => json!({ + "jsonrpc": "2.0", + "id": req.id, + "error": { "code": 1, "message": message }, + }), + }) +} + +/// A descriptor to track, the script hash of its first spk, and a tx paying to that spk. +fn tracked_descriptor() -> anyhow::Result<( + Descriptor, + ElectrumScriptHash, + ScriptBuf, +)> { let descriptor = Descriptor::::from_str(&format!("wpkh({XPUB}/0/*)"))?; let spk = descriptor.at_derivation_index(0)?.script_pubkey(); let spk_hash = ElectrumScriptHash::new(&spk); + Ok((descriptor, spk_hash, spk)) +} - let tx = Transaction { +/// A coinbase paying `sats` to `spk`, so that varying `sats` gives a distinct tx. +fn tx_paying(spk: &ScriptBuf, sats: u64) -> Transaction { + Transaction { version: transaction::Version::ONE, lock_time: absolute::LockTime::ZERO, input: vec![TxIn { @@ -103,12 +183,14 @@ fn anchor_above_local_tip_is_deferred_until_tip_catches_up() -> anyhow::Result<( witness: Witness::new(), }], output: vec![TxOut { - value: Amount::from_sat(50_000), - script_pubkey: spk, + value: Amount::from_sat(sats), + script_pubkey: spk.clone(), }], - }; - let txid = tx.compute_txid(); + } +} +/// The regtest genesis block and an empty block on top of it. +fn base_headers() -> (block::Header, block::Header) { let genesis = constants::genesis_block(Network::Regtest).header; let header_1 = block::Header { version: block::Version::ONE, @@ -118,21 +200,39 @@ fn anchor_above_local_tip_is_deferred_until_tip_catches_up() -> anyhow::Result<( 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 - }; + (genesis, header_1) +} - let mut cache = Cache::default(); - cache.txs.insert(txid, Arc::new(tx.clone())); +/// A block whose transactions have the given merkle root. +fn block_with_root( + prev: &block::Header, + merkle_root: TxMerkleNode, + time: u32, + nonce: u32, +) -> block::Header { + block::Header { + version: block::Version::ONE, + prev_blockhash: prev.block_hash(), + merkle_root, + time, + bits: CompactTarget::from_consensus(0x207fffff), + nonce, + } +} + +/// A block whose only transaction is `txid`, so that its merkle root is the txid itself. +fn block_with_tx(prev: &block::Header, txid: Txid, time: u32, nonce: u32) -> block::Header { + block_with_root(prev, Txid::to_raw_hash(txid).into(), time, nonce) +} +fn new_state( + cache: Cache, + descriptor: Descriptor, + genesis: block::Header, +) -> BlockingState { let mut spk_tracker = DerivedSpkTracker::new(0); spk_tracker.insert_descriptor("external", descriptor, 0); - - let mut state = BlockingState::new( + BlockingState::new( ReqCoord::default(), cache, spk_tracker, @@ -140,12 +240,43 @@ fn anchor_above_local_tip_is_deferred_until_tip_catches_up() -> anyhow::Result<( height: 0, hash: genesis.block_hash(), }), - ); + ) +} + +/// The anchor a tx confirmed in `header` at `height` must be given. +fn anchor_of(header: &block::Header, height: u32) -> ConfirmationBlockTime { + ConfirmationBlockTime { + block_id: BlockId { + height, + hash: header.block_hash(), + }, + confirmation_time: header.time as u64, + } +} + +/// A history response can report a confirmation height above the local tip: on a new block, +/// romanz/electrs notifies the script hash before the header and answers requests in order, so +/// the history arrives while the local tip is still one block behind. Such an anchor must be +/// deferred until the tip catches up — not dropped — and must still be delivered without any +/// further notification for that script. +#[test] +fn anchor_above_local_tip_is_deferred_until_tip_catches_up() -> anyhow::Result<()> { + let (descriptor, spk_hash, spk) = tracked_descriptor()?; + let tx = tx_paying(&spk, 50_000); + let txid = tx.compute_txid(); + let (genesis, header_1) = base_headers(); + let header_2 = block_with_tx(&header_1, txid, 200, 0); + + let mut cache = Cache::default(); + cache.txs.insert(txid, Arc::new(tx.clone())); + + let mut state = new_state(cache, descriptor, genesis); let mut queue = ReqQueue::new(); let mut server = Server { headers: vec![genesis, header_1], spk_hash, - tx, + txs: vec![(tx, 2)], + merkle_proof: (Vec::new(), 0), }; state.init(&mut queue); @@ -185,17 +316,11 @@ fn anchor_above_local_tip_is_deferred_until_tip_catches_up() -> anyhow::Result<( server.headers.push(header_2); let updates = drain_requests(&mut state, &mut queue, &server); - let expected_anchor = ConfirmationBlockTime { - block_id: BlockId { - height: 2, - hash: header_2.block_hash(), - }, - confirmation_time: header_2.time as u64, - }; assert!( - updates - .iter() - .any(|u| u.tx_update.anchors.contains(&(expected_anchor, txid))), + updates.iter().any(|u| u + .tx_update + .anchors + .contains(&(anchor_of(&header_2, 2), txid))), "anchor must be delivered once the tip catches up" ); Ok(()) From 21cd6a6faa05a103c430f48afd4d9ccb951daa23 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BF=97=E5=AE=87?= Date: Fri, 21 Aug 2026 01:51:38 +0000 Subject: [PATCH 2/9] refactor(bdk_electrum_streaming)!: Resolve a job's anchors as a whole set Anchors were drained from a set and staged into `tx_update` as they resolved, so a reorg landing midway had the job emit anchors from the chain it started on alongside anchors from the chain it ended on. Resolve the whole set afresh each pass and hand it back only once all of it resolved, so an update can only ever describe one chain. The loop moves to `advance_anchors` and now walks anchors and checkpoints together from the top, resolving every height in one pass over the chain. BREAKING CHANGE: `SpkJobStage::ProcessingTxsAndAnchors` gains an `anchors_resolved` field, so exhaustive patterns over it no longer compile. --- bdk_electrum_streaming/src/spk_job.rs | 128 +++++++++++++++++--------- bdk_electrum_streaming/tests/state.rs | 84 +++++++++++++++++ 2 files changed, 170 insertions(+), 42 deletions(-) diff --git a/bdk_electrum_streaming/src/spk_job.rs b/bdk_electrum_streaming/src/spk_job.rs index cdfcf50..e59ba28 100644 --- a/bdk_electrum_streaming/src/spk_job.rs +++ b/bdk_electrum_streaming/src/spk_job.rs @@ -19,7 +19,14 @@ pub enum SpkJobStage { }, ProcessingTxsAndAnchors { txs: Option, + /// The `(height, txid)` pairs to anchor. + /// + /// Held whole for the life of the job: every pass resolves all of them afresh + /// against the chain as it is right then, so a reorg landing mid-job cannot leave the + /// job emitting anchors from two different chains. anchors: BTreeSet<(u32, Txid)>, + /// Whether every anchor resolved on the last pass. + anchors_resolved: bool, }, } @@ -28,12 +35,13 @@ impl SpkJobStage { Self::ProcessingTxsAndAnchors { txs: None, anchors: BTreeSet::new(), + anchors_resolved: true, } } /// Whether it's done. pub fn is_done(&self) -> bool { - matches!(self, SpkJobStage::ProcessingTxsAndAnchors { txs, anchors } if txs.is_none() && anchors.is_empty()) + matches!(self, SpkJobStage::ProcessingTxsAndAnchors { txs, anchors_resolved, .. } if txs.is_none() && *anchors_resolved) } } @@ -117,18 +125,23 @@ impl SpkJob { pub fn advance(mut self, queuer: &mut ReqQueuer, cache: &Cache, cp: &CheckPoint) -> Self { let mut made_progress = true; while made_progress { - (self, made_progress) = self.try_advance_once(queuer, cache, cp.clone()); + (self, made_progress) = self.try_advance_once(queuer, cache, cp); let stage_str = match &self.stage { SpkJobStage::ProcessingHistory { status } => format!("ProcessingHistory({status})"), - SpkJobStage::ProcessingTxsAndAnchors { txs, anchors } => { + SpkJobStage::ProcessingTxsAndAnchors { + txs, + anchors, + anchors_resolved, + } => { let inner_str = match txs { Some(TxsJobStage::Txs(txids)) => format!("txs = {}", txids.len()), Some(TxsJobStage::Prevouts(ops)) => format!("prevouts = {}", ops.len()), None => "tx_done".to_string(), }; format!( - "ProcessingTxsAndAnchors({inner_str}, anchors = {})", - anchors.len() + "ProcessingTxsAndAnchors({inner_str}, anchors = {}, resolved = {})", + anchors.len(), + anchors_resolved, ) } }; @@ -167,7 +180,7 @@ impl SpkJob { mut self, queuer: &mut ReqQueuer, cache: &Cache, - tip: CheckPoint, + tip: &CheckPoint, ) -> (Self, bool) { match self.stage { SpkJobStage::ProcessingHistory { status } => match cache.spk_histories.get(&status) { @@ -196,7 +209,11 @@ impl SpkJob { Some((height, tx.txid())) }) .collect(); - self.stage = SpkJobStage::ProcessingTxsAndAnchors { txs, anchors }; + self.stage = SpkJobStage::ProcessingTxsAndAnchors { + txs, + anchors, + anchors_resolved: false, + }; (self, true) } None => { @@ -206,8 +223,7 @@ impl SpkJob { } }, SpkJobStage::ProcessingTxsAndAnchors { - mut txs, - mut anchors, + mut txs, anchors, .. } => { let mut made_progress = false; txs = match txs { @@ -266,44 +282,72 @@ impl SpkJob { None => None, }; - let anchors_start_count = anchors.len(); - anchors.retain(|&(height, txid)| { - if height > tip.height() { - // Nothing to request for a block we don't know exists yet. The job is - // re-advanced once a chain job advances the tip. - return true; - } + // Anchors are resolved from scratch each pass, so this never leaves a + // partially resolved set staged in the update. + let resolved = advance_anchors(queuer, cache, tip, &anchors); + let anchors_resolved = resolved.is_some(); + self.tx_update.anchors = resolved.unwrap_or_default(); - let blockhash = match tip.get(height) { - Some(cp) if cp.height() == height => cp.hash(), - _ => { - queuer.enqueue(request::Header { height }); - return true; - } - }; + self.stage = SpkJobStage::ProcessingTxsAndAnchors { + txs, + anchors, + anchors_resolved, + }; + (self, made_progress) + } + } + } +} - if !cache.headers.contains_key(&blockhash) { - queuer.enqueue(request::Header { height }); - } +/// Resolve `anchors` against `cache`, queueing whatever is missing. +/// +/// Returns the resolved anchors, but only once every one of them resolved — nothing is held on to +/// between calls. Anchors are therefore always resolved as a set against a single chain: a reorg +/// landing midway through simply has the next call resolve all of them against the chain we moved +/// to, instead of leaving a job to emit anchors from the chain it started on next to anchors from +/// the chain it ended on. +/// +/// This is also why anchors are tracked as `(height, txid)` pairs rather than by block hash: each +/// call resolves them against whichever block `tip` currently has at that height. +fn advance_anchors( + queuer: &mut ReqQueuer, + cache: &Cache, + tip: &CheckPoint, + anchors: &BTreeSet<(u32, Txid)>, +) -> Option> { + let mut resolved = BTreeSet::new(); + let mut all_resolved = true; - if let Some(anchor) = cache.anchors.get(&(txid, blockhash)) { - self.tx_update.anchors.insert((*anchor, txid)); - return false; - }; - if cache.failed_anchors.contains(&(txid, blockhash)) { - return false; - } - - queuer.enqueue(request::GetTxMerkle { txid, height }); - true - }); - if anchors.len() < anchors_start_count { - made_progress = true; - } + for &(height, txid) in anchors { + if height > tip.height() { + // Nothing to request for a block we don't know exists yet. The job is + // re-advanced once a chain job advances the tip. + all_resolved = false; + continue; + } - self.stage = SpkJobStage::ProcessingTxsAndAnchors { txs, anchors }; - (self, made_progress) + let blockhash = match tip.get(height) { + Some(cp) => cp.hash(), + None => { + queuer.enqueue(request::Header { height }); + all_resolved = false; + continue; } + }; + + if let Some(anchor) = cache.anchors.get(&(txid, blockhash)) { + resolved.insert((*anchor, txid)); + continue; + } + if cache.failed_anchors.contains(&(txid, blockhash)) { + continue; } + if !cache.headers.contains_key(&blockhash) { + queuer.enqueue(request::Header { height }); + } + + queuer.enqueue(request::GetTxMerkle { txid, height }); + all_resolved = false; } + all_resolved.then_some(resolved) } diff --git a/bdk_electrum_streaming/tests/state.rs b/bdk_electrum_streaming/tests/state.rs index fc3d150..d52b5f3 100644 --- a/bdk_electrum_streaming/tests/state.rs +++ b/bdk_electrum_streaming/tests/state.rs @@ -359,3 +359,87 @@ fn descriptor_inserted_mid_connection_is_subscribed() -> anyhow::Result<()> { ); Ok(()) } + +/// A reorg can land while a job is midway through fetching anchors. Anchors are staged as they +/// resolve, so the job must give up the ones it staged against the chain it started on — otherwise +/// it goes on to emit them in a single update alongside the ones it resolved against the chain it +/// ended on. +#[test] +fn anchors_staged_before_a_reorg_are_not_emitted_after_it() -> anyhow::Result<()> { + let (descriptor, spk_hash, spk) = tracked_descriptor()?; + let (tx_a, tx_b) = (tx_paying(&spk, 50_000), tx_paying(&spk, 60_000)); + let (txid_a, txid_b) = (tx_a.compute_txid(), tx_b.compute_txid()); + let (genesis, header_1) = base_headers(); + let header_2 = block_with_tx(&header_1, txid_a, 200, 0); + let header_3 = block_with_tx(&header_2, txid_b, 300, 0); + // The reorg keeps both txs at their heights — hence the unchanged script status — but in + // different blocks, and extends the chain by one. + let header_2b = block_with_tx(&header_1, txid_a, 222, 1); + let header_3b = block_with_tx(&header_2b, txid_b, 333, 1); + let header_4b = block_with_root(&header_3b, TxMerkleNode::all_zeros(), 400, 0); + + let mut state = new_state(Cache::default(), descriptor, genesis); + let mut queue = ReqQueue::new(); + let mut server = Server { + headers: vec![genesis, header_1, header_2, header_3], + spk_hash, + txs: vec![(tx_a, 2), (tx_b, 3)], + merkle_proof: (Vec::new(), 0), + }; + + // Sync, but hold back tx_b's proof so that the job has staged tx_a's anchor and is still + // waiting on tx_b's when the reorg lands. + state.init(&mut queue); + let mut in_flight = Vec::new(); + while let Some(req) = queue.pop_front() { + if req.method.as_ref() == "blockchain.transaction.get_merkle" + && req.params[0] == json!(txid_b.to_string()) + { + in_flight.push(req); + continue; + } + state.advance(&mut queue, response(&req, &server))?; + } + let held_req = match in_flight.as_slice() { + [req] => req.clone(), + reqs => panic!("expected one held merkle request, got {}", reqs.len()), + }; + let held_resp = response(&held_req, &server); + + server.headers = vec![genesis, header_1, header_2b, header_3b, header_4b]; + state.advance( + &mut queue, + raw_msg(json!({ + "jsonrpc": "2.0", + "method": "blockchain.headers.subscribe", + "params": [{ "hex": serialize_hex(&header_4b), "height": 4 }], + })), + )?; + let mut updates = drain_requests(&mut state, &mut queue, &server); + updates.extend(state.advance(&mut queue, held_resp)?); + updates.extend(drain_requests(&mut state, &mut queue, &server)); + + let anchors = updates + .iter() + .flat_map(|u| u.tx_update.anchors.iter().copied()) + .collect::>(); + for evicted in [ + (anchor_of(&header_2, 2), txid_a), + (anchor_of(&header_3, 3), txid_b), + ] { + assert!( + !anchors.contains(&evicted), + "an anchor to an evicted block must not be emitted after the reorg: {evicted:?}" + ); + } + for expected in [ + (anchor_of(&header_2b, 2), txid_a), + (anchor_of(&header_3b, 3), txid_b), + ] { + assert!( + anchors.contains(&expected), + "both anchors must be refetched against the new chain: {expected:?}" + ); + } + Ok(()) +} From a5056b57ae7d009bc60ed6b0e773414014b7b8d9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BF=97=E5=AE=87?= Date: Fri, 21 Aug 2026 01:51:39 +0000 Subject: [PATCH 3/9] fix(bdk_electrum_streaming)!: Refetch anchors for blocks evicted by a reorg MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An Electrum script status is a hash over txid-height pairs, so a reorg moving a transaction into a different block of the same height leaves it untouched and the server never notifies the script hash. The anchor keeps pointing at a block no longer in the chain and nothing asks again, so the transaction stops being canonical for good. The chain tip is the one thing that reports it. On eviction, every affected script gets the notification the server will not send: `SpkJob::new` replayed from the status and history already cached, so it costs no round-trip. Not a new job type, because `SpkJob` is already rebuilt by the next real notification if it is ever dropped. A replay never displaces a job already in `spk_jobs` — that one carries a status at least as new as anything replayable from cache. `spk_hashes_by_height` is pruned to `SPK_HASHES_BY_HEIGHT_HORIZON` below the tip, since only heights inside `ChainJob`'s window can ever be evicted. BREAKING CHANGE: `Cache` gains `spk_statuses` and `spk_hashes_by_height`, so struct-literal construction no longer compiles. `Cache::default()` is unaffected. --- bdk_electrum_streaming/src/state.rs | 189 +++++++++++++--- bdk_electrum_streaming/tests/env.rs | 206 +++++++++++++++++- bdk_electrum_streaming/tests/state.rs | 297 ++++++++++++++++++++++++++ 3 files changed, 658 insertions(+), 34 deletions(-) diff --git a/bdk_electrum_streaming/src/state.rs b/bdk_electrum_streaming/src/state.rs index 49d3e2e..fd14e86 100644 --- a/bdk_electrum_streaming/src/state.rs +++ b/bdk_electrum_streaming/src/state.rs @@ -1,5 +1,5 @@ use std::{ - collections::{BTreeMap, BTreeSet, HashMap, HashSet}, + collections::{btree_map, BTreeMap, BTreeSet, HashMap, HashSet}, sync::Arc, }; @@ -161,6 +161,13 @@ impl State { .context("Failed to deserialize notification from server")?; match notification { Notification::Header(header_notification) => { + // A same-height reorg is applied by `ChainJob`'s short-circuit without + // fetching anything, so this notification is the only place the + // replacement header is ever offered to us. Caching it here saves the + // anchor refetch a round-trip on the very path it exists for. + let header = *header_notification.header(); + self.cache.headers.insert(header.block_hash(), header); + // Always replace prev job since a new notification means a new tip. self.chain_job = ChainJob::new( self.coord.queuer(req_queue, JobId::Chain), @@ -168,17 +175,7 @@ impl State { *header_notification.header(), header_notification.height(), ); - if let Some(job) = self.chain_job.take() { - match job.try_finish(&mut self.cp) { - Ok(cp) => Ok(Some(self.on_chain_job_completed(req_queue, cp))), - Err(job) => { - self.chain_job = Some(job); - Ok(None) - } - } - } else { - Ok(None) - } + Ok(self.try_finish_chain_job(req_queue)) } Notification::ScriptHash(script_hash_notification) => { let spk_hash = script_hash_notification.script_hash(); @@ -194,6 +191,10 @@ impl State { let mut last_active_indices = BTreeMap::new(); + if spk_status.is_none() { + self.forget_spk_history(spk_hash); + } + if spk_status.is_some() || self.cache.spk_txids.contains_key(&spk_hash) { for script_hash in self.spk_tracker.mark_script_hash_used(&k, i) { self.coord @@ -252,16 +253,9 @@ impl State { if let Some(job) = self.chain_job.take() { let new_blocks = (req.start_height..) .zip(resp.headers.into_iter().map(|h| h.block_hash())); - match job.process_blocks(new_blocks).try_finish(&mut self.cp) { - Ok(cp) => Ok(Some(self.on_chain_job_completed(req_queue, cp))), - Err(job) => { - self.chain_job = Some(job); - Ok(None) - } - } - } else { - Ok(None) + self.chain_job = Some(job.process_blocks(new_blocks)); } + Ok(self.try_finish_chain_job(req_queue)) } JobRequest::GetHeader(req) => { let resp = from_raw(&req, raw)?; @@ -301,6 +295,18 @@ impl State { .entry(req.script_hash) .or_default() .extend(resp.iter().map(|tx| tx.txid())); + // Recorded together with the history, so replaying this script's + // job always finds the history its status stands for. + self.cache.spk_statuses.insert(req.script_hash, spk_status); + for tx in &resp { + if let Some(height) = tx.confirmation_height() { + self.cache + .spk_hashes_by_height + .entry(height.to_consensus_u32()) + .or_default() + .insert(req.script_hash); + } + } } Ok(self.advance_spk_jobs(req_queue, job_ids)) } @@ -311,6 +317,7 @@ impl State { } JobRequest::GetTxMerkle(req) => { let resp = from_raw(&req, raw)?; + let cp = match self.cp.get(req.height) { Some(cp) if cp.height() == req.height => cp, _ => { @@ -379,6 +386,10 @@ impl State { let mut last_active_indices = BTreeMap::new(); + if spk_status.is_none() { + self.forget_spk_history(spk_hash); + } + if spk_status.is_some() || self.cache.spk_txids.contains_key(&spk_hash) { for script_hash in self.spk_tracker.mark_script_hash_used(&k, i) { self.coord @@ -408,6 +419,9 @@ impl State { } JobRequest::HeadersSubscribe(req) => { let resp = from_raw(&req, raw)?; + self.cache + .headers + .insert(resp.header.block_hash(), resp.header); // Always replace prev job since a new notification means a new tip. self.chain_job = ChainJob::new( @@ -416,27 +430,90 @@ impl State { resp.header, resp.height, ); - if let Some(job) = self.chain_job.take() { - match job.try_finish(&mut self.cp) { - Ok(cp) => Ok(Some(self.on_chain_job_completed(req_queue, cp))), - Err(job) => { - self.chain_job = Some(job); - Ok(None) - } - } - } else { - Ok(None) - } + Ok(self.try_finish_chain_job(req_queue)) } } } } } + /// Forget the history the server no longer reports for `spk_hash`. + /// + /// A replay is built from the last status and the heights that status was seen at, so + /// leaving them behind would have a later eviction rebuild this script's job from a history + /// it no longer has. + fn forget_spk_history(&mut self, spk_hash: ElectrumScriptHash) { + self.cache.spk_statuses.remove(&spk_hash); + for spk_hashes in self.cache.spk_hashes_by_height.values_mut() { + spk_hashes.remove(&spk_hash); + } + } + + /// Apply the pending chain job to the local chain, if it has everything it needs. + /// + /// Returns the resulting update, if the job completed. + fn try_finish_chain_job(&mut self, req_queue: &mut ReqQueue) -> Option> { + let job = self.chain_job.take()?; + let prev_cp = self.cp.clone(); + match job.try_finish(&mut self.cp) { + Ok(cp) => Some(self.on_chain_job_completed(req_queue, &prev_cp, cp)), + Err(job) => { + self.chain_job = Some(job); + None + } + } + } + + /// React to the local chain having moved from `prev_cp` to `cp`. + /// /// Spk jobs cannot extend the local chain, so a job whose anchor is above the local tip /// waits with no request in flight. Advancing the tip is what makes such an anchor /// resolvable, so every completed chain job must re-advance the stashed spk jobs. - fn on_chain_job_completed(&mut self, req_queue: &mut ReqQueue, cp: CheckPoint) -> Update { + /// + /// A chain job may also drop blocks. Any transaction we have seen at an evicted height is + /// anchored to a block which is no longer ours, so its anchor has to be refetched — and a + /// spk notification will not tell us to, since a transaction which moved to a different + /// block of the same height leaves the spk status untouched. + fn on_chain_job_completed( + &mut self, + req_queue: &mut ReqQueue, + prev_cp: &CheckPoint, + cp: CheckPoint, + ) -> Update { + let evicted = evicted_heights(prev_cp, &cp); + if !evicted.is_empty() { + tracing::info!( + heights = ?evicted, + "Blocks evicted from the local chain. Refetching anchors." + ); + let affected = evicted + .iter() + .flat_map(|height| self.cache.spk_hashes_by_height.get(height)) + .flatten() + .copied(); + + // Start each affected script the job its own notification would have started, from + // the status and history already cached: a script hash notification we raise + // ourselves, because the server will not raise one. A transaction which moved to a + // different block of the same height leaves the status untouched. + for spk_hash in affected { + // A vacant entry is the whole rule: never displace a job the server's own + // notification built, since that one carries a status at least as new as + // anything we could replay, and every stashed job is re-advanced below anyway. + if let btree_map::Entry::Vacant(e) = self.spk_jobs.entry(spk_hash) { + if let Some(&status) = self.cache.spk_statuses.get(&spk_hash) { + e.insert(SpkJob::new(&self.cache, spk_hash, Some(status))); + } + } + } + } + + // Only heights inside `ChainJob`'s reorg window can ever be evicted, and only evicted + // heights are ever read back, so entries far below the tip can never be consulted + // again. Pruning keeps this bounded with a wide margin over that horizon. + let prune_below = cp.height().saturating_sub(SPK_HASHES_BY_HEIGHT_HORIZON); + self.cache.spk_hashes_by_height = self.cache.spk_hashes_by_height.split_off(&prune_below); + let stashed_jobs = self .spk_jobs .keys() @@ -497,6 +574,23 @@ impl State { } } +/// The heights whose block was dropped from the local chain when it went from `prev` to `next`. +/// +/// Checkpoint chains share everything below the point at which they diverge, so walking `prev` +/// down from its tip until `next` agrees is enough. +fn evicted_heights(prev: &CheckPoint, next: &CheckPoint) -> BTreeSet { + let mut evicted = BTreeSet::new(); + for cp in prev.iter() { + match next.get(cp.height()) { + Some(next_cp) if next_cp.hash() == cp.hash() => break, + _ => { + evicted.insert(cp.height()); + } + } + } + evicted +} + pub fn from_raw(_req: &R, raw: serde_json::Value) -> Result where R: Request, @@ -513,4 +607,33 @@ pub struct Cache { pub anchors: HashMap<(Txid, BlockHash), ConfirmationBlockTime>, pub failed_anchors: HashSet<(Txid, BlockHash)>, pub headers: HashMap, + /// The last status the server reported for each script hash. + /// + /// Replaying a script's job needs its status, since that is the key its history is cached + /// under. Only recorded together with that history, so a replay always finds one. + pub spk_statuses: HashMap, + /// Script hashes whose history reported a transaction at each height. + /// + /// This is what makes a reorg actionable: when the local chain drops a block, the scripts + /// recorded at that height are the ones whose anchors need refetching. + /// + /// Unlike the rest of the cache this is pruned. Its only reader is the eviction path, and + /// evictions can only come from a conflict inside the window [`ChainJob`] rewrites, so + /// entries more than [`SPK_HASHES_BY_HEIGHT_HORIZON`] below the tip can never be read + /// again. + /// + /// That window is also the reorg horizon the anchor refetch inherits: a fork deeper than + /// [`ChainJob`]'s suffix length leaves the checkpoint chain claiming blocks the server does + /// not have, and no eviction is reported for them. + /// + /// [`ChainJob`]: crate::chain_job::ChainJob + pub spk_hashes_by_height: BTreeMap>, } + +/// How far below the tip [`Cache::spk_hashes_by_height`] is retained. +/// +/// Comfortably above [`ChainJob`]'s 21-block suffix, which bounds how deep an eviction — the +/// only thing that reads the map — can ever reach. +/// +/// [`ChainJob`]: crate::chain_job::ChainJob +pub const SPK_HASHES_BY_HEIGHT_HORIZON: u32 = 100; diff --git a/bdk_electrum_streaming/tests/env.rs b/bdk_electrum_streaming/tests/env.rs index 1662d9f..40bf639 100644 --- a/bdk_electrum_streaming/tests/env.rs +++ b/bdk_electrum_streaming/tests/env.rs @@ -5,7 +5,7 @@ use bdk_chain::{ ChainPosition, IndexedTxGraph, }; use bdk_core::{ - bitcoin::{key::Secp256k1, params::REGTEST, Address, Amount}, + bitcoin::{key::Secp256k1, params::REGTEST, Address, Amount, BlockHash, Txid}, ConfirmationBlockTime, }; use bdk_electrum_streaming::{ @@ -359,3 +359,207 @@ async fn new_block_confirmation_is_anchored_live() -> anyhow::Result<()> { Ok(()) } + +type Graph = IndexedTxGraph>; + +/// The anchor `txid` is canonically confirmed at, if it is confirmed at all. +fn canonical_anchor( + chain: &LocalChain, + graph: &Graph, + txid: Txid, +) -> Option { + graph + .graph() + .list_canonical_txs( + chain, + chain.tip().block_id(), + CanonicalizationParams::default(), + ) + .find(|ctx| ctx.tx_node.txid == txid) + .and_then(|ctx| match ctx.chain_position { + ChainPosition::Confirmed { anchor, .. } => Some(anchor), + ChainPosition::Unconfirmed { .. } => None, + }) +} + +/// A live client against a fresh `electrsd`, with the machinery the reorg tests share. +struct LiveWallet { + env: TestEnv, + chain: LocalChain, + graph: Graph, + update_rx: mpsc::UnboundedReceiver>, + client: AsyncClient<&'static str>, + run_handle: tokio::task::JoinHandle>, +} + +impl LiveWallet { + /// Connect to a fresh test environment and apply the first (genesis) update. + async fn new() -> anyhow::Result { + init(); + + let secp = Secp256k1::new(); + let env = TestEnv::new()?; + let electrum_url = env.electrsd.electrum_url.clone(); + + let (external, _) = Descriptor::parse_descriptor(&secp, DESCRIPTORS[0])?; + let (internal, _) = Descriptor::parse_descriptor(&secp, DESCRIPTORS[1])?; + + let mut graph = IndexedTxGraph::::new({ + let mut indexer = KeychainTxOutIndex::<&'static str>::new(LOOKAHEAD, false); + indexer.insert_descriptor(EXTERNAL, external.clone())?; + indexer.insert_descriptor(INTERNAL, internal.clone())?; + indexer + }); + let (mut chain, _cs) = LocalChain::from_genesis_hash(env.genesis_hash()?); + + let mut spk_tracker = DerivedSpkTracker::<&'static str>::new(LOOKAHEAD); + spk_tracker.insert_descriptor(EXTERNAL, external, 0); + spk_tracker.insert_descriptor(INTERNAL, internal, 0); + + let mut state = AsyncState::new( + ReqCoord::default(), + Cache::default(), + spk_tracker, + chain.tip(), + ); + + let (mut update_tx, mut update_rx) = mpsc::unbounded::>(); + let (client, mut client_rx) = AsyncClient::new(); + + let run_handle = tokio::spawn(async move { + let mut conn = TcpStream::connect(&electrum_url).await?; + let (read, write) = conn.split(); + run_async( + &mut state, + &mut update_tx, + &mut client_rx, + read.compat(), + write.compat_write(), + ) + .await?; + anyhow::Ok(()) + }); + + let update = update_rx.next().await.expect("Must have next update"); + apply_update(&mut chain, &mut graph, update)?; + + Ok(Self { + env, + chain, + graph, + update_rx, + client, + run_handle, + }) + } + + /// Apply updates until `f` holds. + /// + /// Errors if the client stops — which is what a connection torn down by an expected server + /// error looks like from here — or if `f` has not held within the timeout. + async fn wait_until( + &mut self, + what: &str, + mut f: impl FnMut(&LocalChain, &Graph) -> bool, + ) -> anyhow::Result<()> { + let timeout = tokio::time::sleep(Duration::from_secs(150)).fuse(); + pin_mut!(timeout); + loop { + if f(&self.chain, &self.graph) { + return Ok(()); + } + futures::select! { + _ = timeout => return Err(anyhow::anyhow!("timed out waiting for {what}")), + update = self.update_rx.next() => { + let update = update.ok_or_else(|| { + anyhow::anyhow!("the client stopped while waiting for {what}") + })?; + apply_update(&mut self.chain, &mut self.graph, update)?; + }, + } + } + } + + /// Mine past coinbase maturity, then send `Amount::ONE_BTC` to a tracked spk and mine it in. + /// + /// Returns the txid and the hash of the block confirming it. + async fn confirm_tracked_tx(&mut self) -> anyhow::Result<(Txid, BlockHash)> { + self.env.mine_blocks(101, None)?; + let premine_height = self.env.rpc_client().get_block_count()? as u32; + self.wait_until("the premined chain", |chain, _| { + chain.tip().height() >= premine_height + }) + .await?; + + let ((_, spk), _) = self + .graph + .index + .next_unused_spk(EXTERNAL) + .expect("must derive spk"); + let txid = self + .env + .send(&Address::from_script(&spk, ®TEST)?, Amount::ONE_BTC)?; + self.wait_until("the unconfirmed tx", |_, graph| { + graph.graph().get_tx(txid).is_some() + }) + .await?; + + self.env.mine_blocks(1, None)?; + self.wait_until("the tx to be anchored", |chain, graph| { + canonical_anchor(chain, graph, txid).is_some() + }) + .await?; + + let anchor = canonical_anchor(&self.chain, &self.graph, txid).expect("just waited for it"); + Ok((txid, anchor.block_id.hash)) + } + + async fn stop(self) -> anyhow::Result<()> { + self.client.stop().await?; + self.run_handle.await??; + Ok(()) + } +} + +/// Issue #12, end to end: a reorg re-mines a confirmed tx into a *different* block at the *same* +/// height. The Electrum script status is a hash over txid-height pairs, so it is unchanged and no +/// script hash notification is sent. The anchor must still be refetched off the tip alone. +#[tokio::test] +async fn reorg_to_same_height_block_refetches_anchor_live() -> anyhow::Result<()> { + let mut w = LiveWallet::new().await?; + let (txid, first_block) = w.confirm_tracked_tx().await?; + let confirm_height = w.env.rpc_client().get_block_count()? as u32; + + // Invalidate the confirming block and re-mine at the same height. The tx is back in the + // mempool, so it goes into the replacement block too. + w.env.reorg(1)?; + let second_block = w.env.rpc_client().get_best_block_hash()?; + assert_ne!( + first_block, second_block, + "the reorg must actually replace the block" + ); + assert_eq!( + w.env.rpc_client().get_block_count()? as u32, + confirm_height, + "the replacement block must be at the same height" + ); + assert!( + w.env + .rpc_client() + .get_block(&second_block)? + .txdata + .iter() + .any(|tx| tx.compute_txid() == txid), + "the replacement block must still contain the tx" + ); + + w.wait_until("the refetched anchor", |chain, graph| { + canonical_anchor(chain, graph, txid).is_some_and(|a| a.block_id.hash == second_block) + }) + .await?; + + let anchor = canonical_anchor(&w.chain, &w.graph, txid).expect("just waited for it"); + assert_eq!(anchor.block_id.height, confirm_height); + + w.stop().await +} diff --git a/bdk_electrum_streaming/tests/state.rs b/bdk_electrum_streaming/tests/state.rs index d52b5f3..86c07d5 100644 --- a/bdk_electrum_streaming/tests/state.rs +++ b/bdk_electrum_streaming/tests/state.rs @@ -360,6 +360,75 @@ fn descriptor_inserted_mid_connection_is_subscribed() -> anyhow::Result<()> { Ok(()) } +/// A reorg can move a transaction into a different block at the *same* height. An Electrum +/// script status is a hash over txid-height pairs, so it does not change and the server has no +/// reason to send a script hash notification. The anchor we already delivered now points at a +/// block that is no longer in the chain, so it must be refetched off the tip update alone. +#[test] +fn anchor_is_refetched_when_tx_moves_to_another_block_of_same_height() -> anyhow::Result<()> { + let (descriptor, spk_hash, spk) = tracked_descriptor()?; + let tx = tx_paying(&spk, 50_000); + let txid = tx.compute_txid(); + let (genesis, header_1) = base_headers(); + let header_2 = block_with_tx(&header_1, txid, 200, 0); + // The block that replaces height 2 contains the tx too, hence the identical script status. + let header_2b = block_with_tx(&header_1, txid, 222, 1); + let header_3b = block::Header { + prev_blockhash: header_2b.block_hash(), + time: 300, + ..header_1 + }; + assert_ne!(header_2.block_hash(), header_2b.block_hash()); + + let mut state = new_state(Cache::default(), descriptor, genesis); + let mut queue = ReqQueue::new(); + let mut server = Server { + headers: vec![genesis, header_1, header_2], + spk_hash, + txs: vec![(tx, 2)], + merkle_proof: (Vec::new(), 0), + }; + + state.init(&mut queue); + let updates = drain_requests(&mut state, &mut queue, &server); + assert!( + updates.iter().any(|u| u + .tx_update + .anchors + .contains(&(anchor_of(&header_2, 2), txid))), + "tx must first be anchored to the original block" + ); + + // Reorg. Only a header notification is sent: the script status is unchanged, so a server + // has no reason to notify the script hash. + server.headers = vec![genesis, header_1, header_2b, header_3b]; + state.advance( + &mut queue, + raw_msg(json!({ + "jsonrpc": "2.0", + "method": "blockchain.headers.subscribe", + "params": [{ "hex": serialize_hex(&header_3b), "height": 3 }], + })), + )?; + let updates = drain_requests(&mut state, &mut queue, &server); + + assert!( + updates.iter().any(|u| u + .chain_update + .as_ref() + .is_some_and(|cp| cp.block_id() == anchor_of(&header_3b, 3).block_id)), + "chain update must follow the reorg" + ); + assert!( + updates.iter().any(|u| u + .tx_update + .anchors + .contains(&(anchor_of(&header_2b, 2), txid))), + "anchor must be refetched for the block that replaced the evicted one" + ); + Ok(()) +} + /// A reorg can land while a job is midway through fetching anchors. Anchors are staged as they /// resolve, so the job must give up the ones it staged against the chain it started on — otherwise /// it goes on to emit them in a single update alongside the ones it resolved against the chain it @@ -443,3 +512,231 @@ fn anchors_staged_before_a_reorg_are_not_emitted_after_it() -> anyhow::Result<() } Ok(()) } + +/// Issue #12's literal case: a reorg to a block of the *same* height, with no growth at all. +/// `ChainJob` applies this by short-circuit straight from the header notification, so it is the +/// one reorg shape which never fetches anything — and every other reorg test here also grows the +/// chain, which takes a different path. +#[test] +fn anchor_is_refetched_after_a_same_height_reorg() -> anyhow::Result<()> { + let (descriptor, spk_hash, spk) = tracked_descriptor()?; + let tx = tx_paying(&spk, 50_000); + let txid = tx.compute_txid(); + let (genesis, header_1) = base_headers(); + let header_2 = block_with_tx(&header_1, txid, 200, 0); + let header_2b = block_with_tx(&header_1, txid, 222, 1); + assert_ne!(header_2.block_hash(), header_2b.block_hash()); + + let mut state = new_state(Cache::default(), descriptor, genesis); + let mut queue = ReqQueue::new(); + let mut server = Server { + headers: vec![genesis, header_1, header_2], + spk_hash, + txs: vec![(tx, 2)], + merkle_proof: (Vec::new(), 0), + }; + + state.init(&mut queue); + let updates = drain_requests(&mut state, &mut queue, &server); + assert!( + updates.iter().any(|u| u + .tx_update + .anchors + .contains(&(anchor_of(&header_2, 2), txid))), + "tx must first be anchored to the original block" + ); + + // The tip does not move: same height, different block, unchanged script status. + server.headers = vec![genesis, header_1, header_2b]; + state.advance( + &mut queue, + raw_msg(json!({ + "jsonrpc": "2.0", + "method": "blockchain.headers.subscribe", + "params": [{ "hex": serialize_hex(&header_2b), "height": 2 }], + })), + )?; + let updates = drain_requests(&mut state, &mut queue, &server); + + assert!( + updates.iter().any(|u| u + .tx_update + .anchors + .contains(&(anchor_of(&header_2b, 2), txid))), + "anchor must be refetched for the block that replaced the evicted one" + ); + Ok(()) +} + +/// The refetch is a script hash notification we raise ourselves, so it must not displace one the +/// server actually sent. A real notification carries a status at least as new as anything we +/// could replay from cache; replacing its job would resolve the script against a stale history +/// and drop whatever the new status was reporting. +#[test] +fn a_replayed_job_does_not_displace_one_the_server_started() -> anyhow::Result<()> { + let (descriptor, spk_hash, spk) = tracked_descriptor()?; + let (tx_a, tx_b) = (tx_paying(&spk, 50_000), tx_paying(&spk, 60_000)); + let (txid_a, txid_b) = (tx_a.compute_txid(), tx_b.compute_txid()); + let (genesis, _) = base_headers(); + let header_1 = block_with_tx(&genesis, txid_b, 100, 0); + let header_2 = block_with_tx(&header_1, txid_a, 200, 0); + // The reorg replaces height 2 with another block holding tx_a, and extends the chain. + let header_2b = block_with_tx(&header_1, txid_a, 222, 1); + let header_3b = block_with_root(&header_2b, TxMerkleNode::all_zeros(), 300, 0); + + let mut state = new_state(Cache::default(), descriptor, genesis); + let mut queue = ReqQueue::new(); + let mut server = Server { + headers: vec![genesis, header_1, header_2], + spk_hash, + // The server only reports tx_a to begin with. + txs: vec![(tx_a, 2)], + merkle_proof: (Vec::new(), 0), + }; + + state.init(&mut queue); + let updates = drain_requests(&mut state, &mut queue, &server); + assert!( + updates.iter().any(|u| u + .tx_update + .anchors + .contains(&(anchor_of(&header_2, 2), txid_a))), + "tx_a must first be anchored to the original block" + ); + + // The server now reports tx_b as well, and notifies the new status. The job that starts is + // the only thing which knows about tx_b. + server.txs.insert(0, (tx_b, 1)); + let new_status = + ElectrumScriptStatus::from_history(&server.history(&json!(spk_hash.to_string()))) + .expect("history must be non-empty"); + state.advance( + &mut queue, + raw_msg(json!({ + "jsonrpc": "2.0", + "method": "blockchain.scripthash.subscribe", + "params": [spk_hash.to_string(), new_status.to_string()], + })), + )?; + + // Hold back that job's history, so it is still in flight when the reorg lands. + let mut in_flight = Vec::new(); + while let Some(req) = queue.pop_front() { + if req.method.as_ref() == "blockchain.scripthash.get_history" { + in_flight.push(req); + continue; + } + state.advance(&mut queue, response(&req, &server))?; + } + let held_req = match in_flight.as_slice() { + [req] => req.clone(), + reqs => panic!("expected one held history request, got {}", reqs.len()), + }; + + // The reorg evicts height 2, where this script is recorded — so the refetch wants to replay + // its job, and must decline because the server's own job is already there. + server.headers = vec![genesis, header_1, header_2b, header_3b]; + state.advance( + &mut queue, + raw_msg(json!({ + "jsonrpc": "2.0", + "method": "blockchain.headers.subscribe", + "params": [{ "hex": serialize_hex(&header_3b), "height": 3 }], + })), + )?; + let mut updates = drain_requests(&mut state, &mut queue, &server); + updates.extend(state.advance(&mut queue, response(&held_req, &server))?); + updates.extend(drain_requests(&mut state, &mut queue, &server)); + + assert!( + updates + .iter() + .any(|u| u.tx_update.txs.iter().any(|tx| tx.compute_txid() == txid_b)), + "the server's own job must survive and deliver what its status was reporting" + ); + assert!( + updates.iter().any(|u| u + .tx_update + .anchors + .contains(&(anchor_of(&header_2b, 2), txid_a))), + "and must still refetch the anchor the reorg invalidated" + ); + Ok(()) +} + +/// A replay is built from the last status and the heights it was seen at, so both have to go +/// when the server stops reporting a history for the script — an RBF'd transaction, say. Left +/// behind, a later eviction at that height would rebuild the job from a history the script no +/// longer has and go asking for proofs of a transaction that is gone. +#[test] +fn a_script_whose_history_goes_away_is_not_replayed() -> anyhow::Result<()> { + let (descriptor, spk_hash, spk) = tracked_descriptor()?; + let tx = tx_paying(&spk, 50_000); + let txid = tx.compute_txid(); + let (genesis, header_1) = base_headers(); + let header_2 = block_with_tx(&header_1, txid, 200, 0); + let header_2b = block_with_root(&header_1, TxMerkleNode::all_zeros(), 222, 1); + let header_3b = block_with_root(&header_2b, TxMerkleNode::all_zeros(), 300, 0); + + let mut state = new_state(Cache::default(), descriptor, genesis); + let mut queue = ReqQueue::new(); + let mut server = Server { + headers: vec![genesis, header_1, header_2], + spk_hash, + txs: vec![(tx, 2)], + merkle_proof: (Vec::new(), 0), + }; + + state.init(&mut queue); + let updates = drain_requests(&mut state, &mut queue, &server); + assert!( + updates.iter().any(|u| u + .tx_update + .anchors + .contains(&(anchor_of(&header_2, 2), txid))), + "tx must first be anchored" + ); + assert!( + state.cache().spk_statuses.contains_key(&spk_hash), + "the status must be recorded while the script has a history" + ); + + // The transaction is gone, so the script's status goes to null. + server.txs = Vec::new(); + state.advance( + &mut queue, + raw_msg(json!({ + "jsonrpc": "2.0", + "method": "blockchain.scripthash.subscribe", + "params": [spk_hash.to_string(), null], + })), + )?; + drain_requests(&mut state, &mut queue, &server); + assert!( + !state.cache().spk_statuses.contains_key(&spk_hash), + "a null status must drop the recorded status" + ); + + // A reorg evicting the height it used to be seen at must not replay anything. + server.headers = vec![genesis, header_1, header_2b, header_3b]; + state.advance( + &mut queue, + raw_msg(json!({ + "jsonrpc": "2.0", + "method": "blockchain.headers.subscribe", + "params": [{ "hex": serialize_hex(&header_3b), "height": 3 }], + })), + )?; + let mut asked = Vec::new(); + while let Some(req) = queue.pop_front() { + asked.push(req.method.to_string()); + state.advance(&mut queue, response(&req, &server))?; + } + assert!( + !asked + .iter() + .any(|m| m == "blockchain.transaction.get_merkle"), + "no proof may be asked for a transaction the script no longer has, got: {asked:?}" + ); + Ok(()) +} From 7ae51695ff87303b8d6adf9f7511864c5ae5f7f4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BF=97=E5=AE=87?= Date: Fri, 21 Aug 2026 01:51:41 +0000 Subject: [PATCH 4/9] fix(bdk_electrum_streaming): Fetch a block's header before its merkle proof MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A proof is verified against the merkle root of the block we have at that height, so asking for both in one pass only worked if the server answered in request order — which the protocol does not promise. If the proof won, the handler could not find the header and cancelled the job; the header arriving second returned `Ok(None)` and advanced nothing. Withhold the proof until the header is cached, and always advance waiting jobs from a `GetHeader` response, leaving only the checkpoint insert conditional. `blockchain.block.header` is keyed by height, so if our chain claims a different block there than the server, no request can ever satisfy the waiting job. That happens below `ChainJob`'s suffix, which nothing rewrites. Cancel those jobs rather than re-asking forever. --- bdk_electrum_streaming/src/spk_job.rs | 5 + bdk_electrum_streaming/src/state.rs | 50 ++++-- bdk_electrum_streaming/tests/state.rs | 219 +++++++++++++++++++++++++- 3 files changed, 253 insertions(+), 21 deletions(-) diff --git a/bdk_electrum_streaming/src/spk_job.rs b/bdk_electrum_streaming/src/spk_job.rs index e59ba28..351fa84 100644 --- a/bdk_electrum_streaming/src/spk_job.rs +++ b/bdk_electrum_streaming/src/spk_job.rs @@ -342,8 +342,13 @@ fn advance_anchors( if cache.failed_anchors.contains(&(txid, blockhash)) { continue; } + // A proof is verified against this block's merkle root, so asking for the proof before + // the header is cached would make the outcome depend on the server answering in request + // order — which the protocol's request ids exist precisely because it does not promise. if !cache.headers.contains_key(&blockhash) { queuer.enqueue(request::Header { height }); + all_resolved = false; + continue; } queuer.enqueue(request::GetTxMerkle { txid, height }); diff --git a/bdk_electrum_streaming/src/state.rs b/bdk_electrum_streaming/src/state.rs index fd14e86..037a5e8 100644 --- a/bdk_electrum_streaming/src/state.rs +++ b/bdk_electrum_streaming/src/state.rs @@ -259,27 +259,45 @@ impl State { } JobRequest::GetHeader(req) => { let resp = from_raw(&req, raw)?; + let hash = resp.header.block_hash(); + self.cache.headers.insert(hash, resp.header); - self.cache - .headers - .insert(resp.header.block_hash(), resp.header); - - // Do not extend checkpoints. - if req.height > self.cp.height() { - return Ok(None); - } - // Do not replace blocks. - if self + // The block our own chain has at this height, if it has one. + let ours = self .cp .get(req.height) - .is_some_and(|cp| cp.height() == req.height) - { + .filter(|cp| cp.height() == req.height) + .map(|cp| cp.hash()); + + // `blockchain.block.header` is keyed by height, so this is the only + // block the server will ever hand us there. If our chain claims a + // different one it is stale at a height no chain job will rewrite — + // below `ChainJob`'s suffix nothing reports the difference — and the + // header a job is waiting on can never arrive. Asking again would loop + // for as long as the connection is up, so drop the jobs instead and + // let the next notification rebuild them. + if ours.is_some_and(|ours| ours != hash) { + tracing::warn!( + height = req.height, + ours = ours.expect("just matched").to_string(), + theirs = hash.to_string(), + "Local chain disagrees with the server below the reorg horizon", + ); + self.cancel_jobs(job_ids); return Ok(None); } - self.cp = self - .cp - .clone() - .insert(BlockId::from((req.height, resp.header.block_hash()))); + + // Whether or not the checkpoint chain wants this header, the header + // now being cached is what a waiting anchor may need, and nothing else + // will wake it. So only the insert below is conditional; the jobs are + // advanced either way. + // + // The insert is skipped when it would extend the checkpoints, and + // when we already have this block. + let extends = req.height > self.cp.height(); + if !extends && ours.is_none() { + self.cp = self.cp.clone().insert(BlockId::from((req.height, hash))); + } Ok(self.advance_spk_jobs(req_queue, job_ids)) } JobRequest::GetHistory(req) => { diff --git a/bdk_electrum_streaming/tests/state.rs b/bdk_electrum_streaming/tests/state.rs index 86c07d5..c01474c 100644 --- a/bdk_electrum_streaming/tests/state.rs +++ b/bdk_electrum_streaming/tests/state.rs @@ -147,6 +147,34 @@ fn drain_requests( updates } +/// Drain like [`drain_requests`], but answer every merkle proof in the queue ahead of everything +/// else in each round. +/// +/// The Electrum protocol carries a request id precisely because responses need not come back in +/// the order they were asked for: romanz/electrs happens to answer in order, Fulcrum processes +/// requests concurrently. Nothing may depend on the ordering. +fn drain_requests_proofs_first( + state: &mut BlockingState, + queue: &mut ReqQueue, + server: &Server, +) -> Vec> { + let mut updates = Vec::new(); + while !queue.is_empty() { + let (proofs, rest): (Vec<_>, Vec<_>) = queue + .drain(..) + .partition(|req| req.method.as_ref() == "blockchain.transaction.get_merkle"); + for req in proofs.into_iter().chain(rest) { + if let Some(update) = state + .advance(queue, response(&req, server)) + .expect("must advance") + { + updates.push(update); + } + } + } + updates +} + /// The server's answer to `req`, as a raw JSON-RPC result or error message. fn response(req: &RawRequest, server: &Server) -> RawNotificationOrResponse { raw_msg(match server.answer(req) { @@ -230,12 +258,9 @@ fn new_state( descriptor: Descriptor, genesis: block::Header, ) -> BlockingState { - let mut spk_tracker = DerivedSpkTracker::new(0); - spk_tracker.insert_descriptor("external", descriptor, 0); - BlockingState::new( - ReqCoord::default(), + new_state_with_cp( cache, - spk_tracker, + descriptor, CheckPoint::new(BlockId { height: 0, hash: genesis.block_hash(), @@ -243,6 +268,16 @@ fn new_state( ) } +fn new_state_with_cp( + cache: Cache, + descriptor: Descriptor, + cp: CheckPoint, +) -> BlockingState { + let mut spk_tracker = DerivedSpkTracker::new(0); + spk_tracker.insert_descriptor("external", descriptor, 0); + BlockingState::new(ReqCoord::default(), cache, spk_tracker, cp) +} + /// The anchor a tx confirmed in `header` at `height` must be given. fn anchor_of(header: &block::Header, height: u32) -> ConfirmationBlockTime { ConfirmationBlockTime { @@ -568,6 +603,116 @@ fn anchor_is_refetched_after_a_same_height_reorg() -> anyhow::Result<()> { Ok(()) } +/// A proof is verified against the merkle root of the block we have at that height, so a job +/// which asks for the proof and that block's header together only works if the server answers in +/// request order. It need not: the protocol carries request ids for that reason. +#[test] +fn anchor_is_refetched_whatever_order_the_server_answers_in() -> anyhow::Result<()> { + let (descriptor, spk_hash, spk) = tracked_descriptor()?; + let tx = tx_paying(&spk, 50_000); + let txid = tx.compute_txid(); + let (genesis, header_1) = base_headers(); + let header_2 = block_with_tx(&header_1, txid, 200, 0); + // Height 2 is replaced and the chain grows, so the notified header is height 3's — the + // replacement block's header has to be fetched before its proof can be verified. + let header_2b = block_with_tx(&header_1, txid, 222, 1); + let header_3b = block_with_root(&header_2b, TxMerkleNode::all_zeros(), 300, 0); + + let mut state = new_state(Cache::default(), descriptor, genesis); + let mut queue = ReqQueue::new(); + let mut server = Server { + headers: vec![genesis, header_1, header_2], + spk_hash, + txs: vec![(tx, 2)], + merkle_proof: (Vec::new(), 0), + }; + + state.init(&mut queue); + let updates = drain_requests(&mut state, &mut queue, &server); + assert!( + updates.iter().any(|u| u + .tx_update + .anchors + .contains(&(anchor_of(&header_2, 2), txid))), + "tx must first be anchored to the original block" + ); + + server.headers = vec![genesis, header_1, header_2b, header_3b]; + state.advance( + &mut queue, + raw_msg(json!({ + "jsonrpc": "2.0", + "method": "blockchain.headers.subscribe", + "params": [{ "hex": serialize_hex(&header_3b), "height": 3 }], + })), + )?; + let updates = drain_requests_proofs_first(&mut state, &mut queue, &server); + + assert!( + updates.iter().any(|u| u + .tx_update + .anchors + .contains(&(anchor_of(&header_2b, 2), txid))), + "the anchor must be refetched even when the proof overtakes the header" + ); + Ok(()) +} + +/// A client restored from a persisted checkpoint chain starts with blocks in its chain whose +/// headers are not in its cache. Resolving an anchor at such a height needs both the header and +/// the proof, and nothing else will fetch that header — `ChainJob` short-circuits, since the tip +/// is already correct. +/// +/// So this is the case where the two halves of the ordering fix are load-bearing: the proof must +/// not be asked for before the header is cached, and the header response must advance the waiting +/// job even though the chain itself has nothing to learn from it. +#[test] +fn anchor_resolves_when_the_chain_is_restored_without_its_headers() -> anyhow::Result<()> { + let (descriptor, spk_hash, spk) = tracked_descriptor()?; + let tx = tx_paying(&spk, 50_000); + let txid = tx.compute_txid(); + let (genesis, header_1) = base_headers(); + let header_2 = block_with_tx(&header_1, txid, 200, 0); + let header_3 = block_with_root(&header_2, TxMerkleNode::all_zeros(), 300, 0); + + // The restored chain knows the blocks, the fresh cache knows none of their headers. The tx + // is one block below the tip, so the header it needs is not the one `headers.subscribe` + // hands back and nothing else fetches it either — `ChainJob` short-circuits, the tip being + // already correct. + let cp = CheckPoint::new(BlockId { + height: 0, + hash: genesis.block_hash(), + }) + .insert(BlockId { + height: 2, + hash: header_2.block_hash(), + }) + .insert(BlockId { + height: 3, + hash: header_3.block_hash(), + }); + let mut state = new_state_with_cp(Cache::default(), descriptor, cp); + let mut queue = ReqQueue::new(); + let server = Server { + headers: vec![genesis, header_1, header_2, header_3], + spk_hash, + txs: vec![(tx, 2)], + merkle_proof: (Vec::new(), 0), + }; + + state.init(&mut queue); + let updates = drain_requests_proofs_first(&mut state, &mut queue, &server); + + assert!( + updates.iter().any(|u| u + .tx_update + .anchors + .contains(&(anchor_of(&header_2, 2), txid))), + "the anchor must resolve even when the proof overtakes the header it is verified against" + ); + Ok(()) +} + /// The refetch is a script hash notification we raise ourselves, so it must not displace one the /// server actually sent. A real notification carries a status at least as new as anything we /// could replay from cache; replacing its job would resolve the script against a stale history @@ -664,6 +809,70 @@ fn a_replayed_job_does_not_displace_one_the_server_started() -> anyhow::Result<( Ok(()) } +/// A persisted checkpoint chain can be stale at a height below the window [`ChainJob`] rewrites +/// — an offline reorg, say. Then the block *we* have at that height is one the server does not +/// have, and `blockchain.block.header` is keyed by height, so no request can ever fetch it. +/// +/// Withholding the proof until that header is cached must not turn into an endless request loop. +#[test] +fn a_header_the_server_does_not_have_does_not_loop() -> anyhow::Result<()> { + let (descriptor, spk_hash, spk) = tracked_descriptor()?; + let tx = tx_paying(&spk, 50_000); + let txid = tx.compute_txid(); + let (genesis, header_1) = base_headers(); + // The server's height 2, and the stale one our persisted chain still claims. + let server_2 = block_with_tx(&header_1, txid, 200, 0); + let stale_2 = block_with_tx(&header_1, txid, 999, 7); + assert_ne!(server_2.block_hash(), stale_2.block_hash()); + + let mut chain = vec![genesis, header_1, server_2]; + for height in 3..=30u32 { + let prev = *chain.last().expect("non-empty"); + chain.push(block_with_root( + &prev, + TxMerkleNode::all_zeros(), + 1000 + height, + 0, + )); + } + + // Tip agrees with the server, so `ChainJob` short-circuits and never rewrites height 2. + let cp = CheckPoint::new(BlockId { + height: 0, + hash: genesis.block_hash(), + }) + .insert(BlockId { + height: 2, + hash: stale_2.block_hash(), + }) + .insert(BlockId { + height: 30, + hash: chain[30].block_hash(), + }); + let mut state = new_state_with_cp(Cache::default(), descriptor, cp); + let mut queue = ReqQueue::new(); + let server = Server { + headers: chain, + spk_hash, + txs: vec![(tx, 2)], + merkle_proof: (Vec::new(), 0), + }; + + state.init(&mut queue); + let mut served = 0; + while let Some(req) = queue.pop_front() { + served += 1; + assert!( + served < 200, + "the client must not loop: {served} requests, last was {} {:?}", + req.method, + req.params + ); + state.advance(&mut queue, response(&req, &server))?; + } + Ok(()) +} + /// A replay is built from the last status and the heights it was seen at, so both have to go /// when the server stops reporting a history for the script — an RBF'd transaction, say. Left /// behind, a later eviction at that height would rebuild the job from a history the script no From 002f228ceaed6d31b8a57b8247a82bef8d105964 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BF=97=E5=AE=87?= Date: Fri, 21 Aug 2026 01:51:42 +0000 Subject: [PATCH 5/9] fix(bdk_electrum_streaming)!: Never record a disproof the server cannot give MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two answers to a proof request were being read as evidence about our own block, and neither is. An error means the server has no proof at that height — the everyday reorg, a transaction back in the mempool. That took the whole connection down. It is answered in place now, and the job is left stashed rather than cancelled: not advancing it is what stops a re-ask loop, keeping it is what makes recovery automatic, since the same-height reorg this all exists for sends no notification to rebuild it with. A mismatching proof means the server proved inclusion in whichever block *it* has at that height, so our chain and the server's disagree there. It says nothing about whether the transaction is in *our* block. That is the same conclusion `GetHeader` draws from the same kind of evidence, so it gets the same treatment — drop the jobs, record nothing. `Cache::failed_anchors` therefore goes. It could only ever be written from a mismatch, and a mismatch only happens when the two chains disagree, so it never held a true disproof — only an artifact, permanently, since nothing pruned it. A chain that later came back to that block would consult a verdict that had never been established and skip the anchor for good. BREAKING CHANGE: `Cache::failed_anchors` is removed. --- bdk_electrum_streaming/src/spk_job.rs | 3 - bdk_electrum_streaming/src/state.rs | 44 ++++- bdk_electrum_streaming/tests/env.rs | 35 ++++ bdk_electrum_streaming/tests/state.rs | 224 ++++++++++++++++++++++++++ 4 files changed, 296 insertions(+), 10 deletions(-) diff --git a/bdk_electrum_streaming/src/spk_job.rs b/bdk_electrum_streaming/src/spk_job.rs index 351fa84..70ae6f9 100644 --- a/bdk_electrum_streaming/src/spk_job.rs +++ b/bdk_electrum_streaming/src/spk_job.rs @@ -339,9 +339,6 @@ fn advance_anchors( resolved.insert((*anchor, txid)); continue; } - if cache.failed_anchors.contains(&(txid, blockhash)) { - continue; - } // A proof is verified against this block's merkle root, so asking for the proof before // the header is cached would make the outcome depend on the server answering in request // order — which the protocol's request ids exist precisely because it does not promise. diff --git a/bdk_electrum_streaming/src/state.rs b/bdk_electrum_streaming/src/state.rs index 037a5e8..26bbeda 100644 --- a/bdk_electrum_streaming/src/state.rs +++ b/bdk_electrum_streaming/src/state.rs @@ -1,5 +1,5 @@ use std::{ - collections::{btree_map, BTreeMap, BTreeSet, HashMap, HashSet}, + collections::{btree_map, BTreeMap, BTreeSet, HashMap}, sync::Arc, }; @@ -237,6 +237,32 @@ impl State { let raw = match raw_response.result { Ok(raw) => raw, Err(err) => { + // An anchor fetch is speculative: it asks for a proof of inclusion at + // the height a transaction was last reported at, and a reorg may have + // taken the transaction out of that block, or out of the chain + // altogether. A server answers that with an error, which is an answer, + // not a reason to bring the connection down. + if let JobRequest::GetTxMerkle(req) = &orig_req { + tracing::warn!( + txid = req.txid.to_string(), + block_height = req.height, + ?err, + "Server gave no merkle proof at this height. Reorg?", + ); + // An error proves nothing — it is equally a rate limit, an index + // still catching up or a daemon hiccup — so nothing durable is + // recorded, and there is nowhere to record it: a proof + // request that comes back with nothing leaves no trace. + // + // The job is left stashed rather than cancelled. Returning without + // advancing it is what stops the re-ask loop; keeping it is what + // makes the recovery automatic, since the next chain update advances + // it again and the answer may be different once our chain has caught + // up with the server's. Cancelling would need a notification to + // revive it, and the same-height reorg this all exists for is + // precisely the case that sends none. + return Ok(None); + } // Cancel jobs that resulted in error. self.cancel_jobs(job_ids); return Err(anyhow::anyhow!(err).context("Server responded with error")); @@ -349,7 +375,7 @@ impl State { } }; let header = match self.cache.headers.get(&cp.hash()) { - Some(header) => header, + Some(header) => *header, None => { tracing::warn!( ?req, @@ -382,11 +408,16 @@ impl State { block_hash = header.block_hash().to_string(), header_root = header.merkle_root.to_string(), expected_root = exp_root.to_string(), - "Failed to verify anchor." + "Proof does not match the block we have at this height", ); - self.cache - .failed_anchors - .insert((req.txid, header.block_hash())); + // The server proves inclusion in whichever block *it* has at this + // height, so a mismatch says our chain and the server's disagree + // there — not that the transaction is absent from our block. That is + // the conclusion `GetHeader` draws from the same kind of evidence, + // so it gets the same treatment: drop the jobs and let the next + // notification rebuild them. + self.cancel_jobs(job_ids); + return Ok(None); } Ok(self.advance_spk_jobs(req_queue, job_ids)) } @@ -623,7 +654,6 @@ pub struct Cache { pub spk_txids: HashMap>, pub txs: HashMap>, pub anchors: HashMap<(Txid, BlockHash), ConfirmationBlockTime>, - pub failed_anchors: HashSet<(Txid, BlockHash)>, pub headers: HashMap, /// The last status the server reported for each script hash. /// diff --git a/bdk_electrum_streaming/tests/env.rs b/bdk_electrum_streaming/tests/env.rs index 40bf639..3bedfa7 100644 --- a/bdk_electrum_streaming/tests/env.rs +++ b/bdk_electrum_streaming/tests/env.rs @@ -563,3 +563,38 @@ async fn reorg_to_same_height_block_refetches_anchor_live() -> anyhow::Result<() w.stop().await } + +/// The everyday reorg: one which takes a tx out of its block and back to the mempool. The anchor +/// refetch then asks for a proof the server cannot give, and that error must not take the +/// connection down with it. +#[tokio::test] +async fn reorg_unconfirming_a_tx_keeps_the_connection_alive() -> anyhow::Result<()> { + let mut w = LiveWallet::new().await?; + let (txid, _) = w.confirm_tracked_tx().await?; + let confirm_height = w.env.rpc_client().get_block_count()? as u32; + + // Invalidate the confirming block and replace it with empty ones, so the tx cannot be + // re-mined and the server has no proof to give at that height. + w.env.invalidate_blocks(1)?; + w.env.mine_empty_block()?; + w.env.mine_empty_block()?; + let tip_height = w.env.rpc_client().get_block_count()? as u32; + assert_eq!(tip_height, confirm_height + 1); + assert!( + w.env.rpc_client().get_raw_mempool()?.contains(&txid), + "the tx must be back in the mempool, so the server really has no proof for it" + ); + + // The connection has to keep serving: `wait_until` fails if the client stops. + w.wait_until("the chain tip after the reorg", |chain, _| { + chain.tip().height() >= tip_height + }) + .await?; + + assert!( + canonical_anchor(&w.chain, &w.graph, txid).is_none(), + "the tx must no longer be canonically confirmed" + ); + + w.stop().await +} diff --git a/bdk_electrum_streaming/tests/state.rs b/bdk_electrum_streaming/tests/state.rs index c01474c..d60b5e7 100644 --- a/bdk_electrum_streaming/tests/state.rs +++ b/bdk_electrum_streaming/tests/state.rs @@ -548,6 +548,136 @@ fn anchors_staged_before_a_reorg_are_not_emitted_after_it() -> anyhow::Result<() Ok(()) } +/// The everyday reorg: one which takes a transaction out of its block and back to the mempool. +/// The refetch is speculative — we ask for a proof of inclusion at a height the transaction was +/// *last seen* at — so the server answering "not in that block" is expected, and must not take +/// the connection down with it. +#[test] +fn a_tx_unconfirmed_by_a_reorg_does_not_error_the_connection() -> anyhow::Result<()> { + let (descriptor, spk_hash, spk) = tracked_descriptor()?; + let tx = tx_paying(&spk, 50_000); + let txid = tx.compute_txid(); + let (genesis, header_1) = base_headers(); + let header_2 = block_with_tx(&header_1, txid, 200, 0); + // Height 2 is replaced by a block without the tx, and the chain grows by one. + let header_2b = block_with_root(&header_1, TxMerkleNode::all_zeros(), 222, 1); + let header_3b = block_with_root(&header_2b, TxMerkleNode::all_zeros(), 333, 0); + + let mut state = new_state(Cache::default(), descriptor, genesis); + let mut queue = ReqQueue::new(); + let mut server = Server { + headers: vec![genesis, header_1, header_2], + spk_hash, + txs: vec![(tx, 2)], + merkle_proof: (Vec::new(), 0), + }; + + state.init(&mut queue); + let updates = drain_requests(&mut state, &mut queue, &server); + assert!( + updates.iter().any(|u| u + .tx_update + .anchors + .contains(&(anchor_of(&header_2, 2), txid))), + "tx must first be anchored" + ); + + // The reorg leaves the tx in the mempool, so the server no longer has it in any block. + server.headers = vec![genesis, header_1, header_2b, header_3b]; + server.txs = Vec::new(); + state.advance( + &mut queue, + raw_msg(json!({ + "jsonrpc": "2.0", + "method": "blockchain.headers.subscribe", + "params": [{ "hex": serialize_hex(&header_3b), "height": 3 }], + })), + )?; + + while let Some(req) = queue.pop_front() { + state + .advance(&mut queue, response(&req, &server)) + .map_err(|e| anyhow::anyhow!("{e:#}"))?; + } + Ok(()) +} + +/// A server error is not a disproof — it is equally a rate limit, an index still catching up or +/// a daemon hiccup — so it must not be recorded as one. The transaction stays in the record of +/// what was seen at that height, so the next reorg of that height asks again; and the job it +/// blocked must still finish rather than re-ask in a loop. +#[test] +fn a_merkle_error_is_not_recorded_as_a_failed_anchor() -> anyhow::Result<()> { + let (descriptor, spk_hash, spk) = tracked_descriptor()?; + let tx = tx_paying(&spk, 50_000); + let txid = tx.compute_txid(); + let (genesis, header_1) = base_headers(); + let header_2 = block_with_tx(&header_1, txid, 200, 0); + // Height 2 is replaced by a block containing the tx, and the chain grows by one. + let header_2b = block_with_tx(&header_1, txid, 222, 1); + let header_3b = block_with_root(&header_2b, TxMerkleNode::all_zeros(), 300, 0); + + let mut state = new_state(Cache::default(), descriptor, genesis); + let mut queue = ReqQueue::new(); + let mut server = Server { + headers: vec![genesis, header_1, header_2], + spk_hash, + txs: vec![(tx.clone(), 2)], + merkle_proof: (Vec::new(), 0), + }; + + // Sync, but fail every merkle request the way a busy server would. + state.init(&mut queue); + let mut merkle_requests = 0; + while let Some(req) = queue.pop_front() { + let resp = if req.method.as_ref() == "blockchain.transaction.get_merkle" { + merkle_requests += 1; + assert!( + merkle_requests < 10, + "a server error must not put the job in a re-ask loop" + ); + raw_msg(json!({ + "jsonrpc": "2.0", + "id": req.id, + "error": { "code": 1, "message": "server busy" }, + })) + } else { + response(&req, &server) + }; + state.advance(&mut queue, resp)?; + } + assert_eq!( + merkle_requests, 1, + "the job must give up on the pair rather than re-ask" + ); + assert!( + state.cache().anchors.is_empty(), + "an error proves nothing, so no anchor may be recorded from it" + ); + + // The reorg replaces the block, so the anchor is asked for again — which it could not be + // had the error dropped this script from `spk_hashes_by_height`. + server.headers = vec![genesis, header_1, header_2b, header_3b]; + state.advance( + &mut queue, + raw_msg(json!({ + "jsonrpc": "2.0", + "method": "blockchain.headers.subscribe", + "params": [{ "hex": serialize_hex(&header_3b), "height": 3 }], + })), + )?; + let updates = drain_requests(&mut state, &mut queue, &server); + + assert!( + updates.iter().any(|u| u + .tx_update + .anchors + .contains(&(anchor_of(&header_2b, 2), txid))), + "the tx must still be asked about at this height after a server error" + ); + Ok(()) +} + /// Issue #12's literal case: a reorg to a block of the *same* height, with no growth at all. /// `ChainJob` applies this by short-circuit straight from the header notification, so it is the /// one reorg shape which never fetches anything — and every other reorg test here also grows the @@ -949,3 +1079,97 @@ fn a_script_whose_history_goes_away_is_not_replayed() -> anyhow::Result<()> { ); Ok(()) } + +/// A server proves inclusion in whichever block *it* has at a height, so a proof whose root does +/// not match ours says the two chains disagree there — not that the transaction is absent from +/// our block. Remembering that against our block would be a verdict the proof cannot support, and +/// a chain that came back to that block would consult it and skip the anchor for good. +#[test] +fn a_proof_for_another_block_is_not_a_verdict_on_ours() -> anyhow::Result<()> { + let (descriptor, spk_hash, spk) = tracked_descriptor()?; + let tx = tx_paying(&spk, 50_000); + let txid = tx.compute_txid(); + let (genesis, header_1) = base_headers(); + let ours = block_with_tx(&header_1, txid, 200, 0); + // Their block holds the tx alongside another, so it needs a different proof — the root the + // server's proof expands to cannot match the root of our block. + let proof_theirs = response::TxMerkle { + block_height: absolute::Height::from_consensus(2)?, + merkle: vec![sha256d::Hash::hash(b"the other tx")], + pos: 1, + }; + let theirs = block_with_root(&header_1, proof_theirs.expected_merkle_root(txid), 222, 1); + assert_ne!(ours.merkle_root, theirs.merkle_root); + + let mut chain = vec![genesis, header_1, theirs]; + for height in 3..=30u32 { + let prev = *chain.last().expect("non-empty"); + chain.push(block_with_root( + &prev, + TxMerkleNode::all_zeros(), + 1000 + height, + 0, + )); + } + + // A persisted chain holding our block at height 2, and its header already cached — otherwise + // `GetHeader` catches the disagreement before any proof is asked for. The tip agrees, so no + // chain job runs and nothing rewrites height 2: the disagreement is below the window. + let mut cache = Cache::default(); + cache.headers.insert(ours.block_hash(), ours); + let cp = CheckPoint::new(BlockId { + height: 0, + hash: genesis.block_hash(), + }) + .insert(BlockId { + height: 2, + hash: ours.block_hash(), + }) + .insert(BlockId { + height: 30, + hash: chain[30].block_hash(), + }); + let mut state = new_state_with_cp(cache, descriptor, cp); + let mut queue = ReqQueue::new(); + let mut server = Server { + headers: chain, + spk_hash, + txs: vec![(tx, 2)], + merkle_proof: (proof_theirs.merkle.clone(), proof_theirs.pos), + }; + + state.init(&mut queue); + let mut served = 0; + while let Some(req) = queue.pop_front() { + served += 1; + assert!(served < 200, "a mismatch must not become a request loop"); + state.advance(&mut queue, response(&req, &server))?; + } + assert!( + state.cache().anchors.is_empty(), + "a proof for a block we do not have must not anchor anything" + ); + + // The server comes back to our block at that height. Nothing durable was written against it, + // so the job a notification rebuilds must be able to anchor there. + server.headers[2] = ours; + server.merkle_proof = (Vec::new(), 0); + let status = ElectrumScriptStatus::from_history(&server.history(&json!(spk_hash.to_string()))) + .expect("history must be non-empty"); + state.advance( + &mut queue, + raw_msg(json!({ + "jsonrpc": "2.0", + "method": "blockchain.scripthash.subscribe", + "params": [spk_hash.to_string(), status.to_string()], + })), + )?; + let updates = drain_requests(&mut state, &mut queue, &server); + assert!( + updates + .iter() + .any(|u| u.tx_update.anchors.contains(&(anchor_of(&ours, 2), txid))), + "the anchor must still be reachable once the chains agree again" + ); + Ok(()) +} From 5f6f8c4371136ba612df9b357c9eb09aff52888b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BF=97=E5=AE=87?= Date: Fri, 21 Aug 2026 01:51:44 +0000 Subject: [PATCH 6/9] fix(bdk_electrum_streaming)!: Discard responses that predate a reorg MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A server answers from the chain it had when it received the request, so a response arriving after a reorg may describe a block we have left behind. A stale proof was verified against the replacement block and its valid anchor written off permanently; a stale error was blamed on the replacement; a stale header could be spliced into the checkpoint chain. `ReqCoord` stamps each request with a chain generation, bumped whenever the local chain drops blocks, so `pop` can report whether a reorg landed since. All three handlers now discard such a response and ask again. This closes the stale-chain case only; a genuine fault on the current chain is handled in the previous commit by recording nothing. `ReqCoord::pop` returns `PoppedRequest` rather than a tuple. Drops `ReqCoord::clear`, which had no callers — `State::reset` deliberately resends `pending_requests()` instead. BREAKING CHANGE: `ReqCoord::pop` returns `Option` rather than `Option<(JobRequest, BTreeSet)>`, and `ReqCoord::clear` is removed. --- bdk_electrum_streaming/src/req.rs | 49 +++-- bdk_electrum_streaming/src/state.rs | 35 +++- bdk_electrum_streaming/tests/state.rs | 266 ++++++++++++++++++++++++++ 3 files changed, 332 insertions(+), 18 deletions(-) diff --git a/bdk_electrum_streaming/src/req.rs b/bdk_electrum_streaming/src/req.rs index b1b70db..85745f7 100644 --- a/bdk_electrum_streaming/src/req.rs +++ b/bdk_electrum_streaming/src/req.rs @@ -87,6 +87,20 @@ impl From for JobRequest { } } +/// A response's originating request, alongside what is needed to route it. +#[derive(Debug, Clone)] +pub struct PoppedRequest { + /// The request this response is answering. + pub request: JobRequest, + /// Jobs awaiting this response. + pub job_ids: BTreeSet, + /// Whether the local chain dropped blocks after this request was sent. + /// + /// The server answers from its own chain, so a response which predates a reorg may describe + /// a block which is no longer the one we have at that height. + pub reorged_since_sent: bool, +} + /// Request coordinator. /// /// Associates responses to their requests and requests to their jobs. @@ -94,10 +108,12 @@ impl From for JobRequest { pub struct ReqCoord { /// Next request id. next_id: u32, - /// Req id -> Req. - awaiting_responses: HashMap, + /// Req id -> the request, and the chain generation it was enqueued at. + awaiting_responses: HashMap, /// So we won't have duplicate requests. req_to_job: HashMap>, + /// Bumped every time the local chain drops blocks. + chain_generation: u64, } impl ReqCoord { @@ -112,16 +128,22 @@ impl ReqCoord { &mut self.next_id } - pub fn pop(&mut self, req_id: u32) -> Option<(JobRequest, BTreeSet)> { - let any_req = self.awaiting_responses.remove(&req_id)?; - let job_ids = self.req_to_job.remove(&any_req).unwrap_or_default(); - Some((any_req, job_ids)) + pub fn pop(&mut self, req_id: u32) -> Option { + let (request, generation) = self.awaiting_responses.remove(&req_id)?; + let job_ids = self.req_to_job.remove(&request).unwrap_or_default(); + Some(PoppedRequest { + request, + job_ids, + reorged_since_sent: generation < self.chain_generation, + }) } - /// To be called when the network resets. - pub fn clear(&mut self) { - self.awaiting_responses.clear(); - self.req_to_job.clear(); + /// To be called when the local chain drops blocks. + /// + /// Requests already in flight were made against the old chain, so their responses can no + /// longer be trusted to describe the blocks we now have. + pub fn bump_chain_generation(&mut self) { + self.chain_generation += 1; } pub fn queuer<'q>(&'q mut self, queue: &'q mut ReqQueue, job_id: JobId) -> ReqQueuer<'q> { @@ -136,7 +158,7 @@ impl ReqCoord { pub fn pending_requests(&self) -> impl ExactSizeIterator + '_ { self.awaiting_responses .iter() - .map(|(&req_id, req)| req.to_raw(req_id)) + .map(|(&req_id, (req, _))| req.to_raw(req_id)) } } @@ -163,7 +185,10 @@ impl<'q> ReqQueuer<'q> { e.insert(BTreeSet::new()).insert(self.job_id); let req_id = self.coord.next_id; self.coord.next_id = self.coord.next_id.wrapping_add(1); - self.coord.awaiting_responses.insert(req_id, req.clone()); + let generation = self.coord.chain_generation; + self.coord + .awaiting_responses + .insert(req_id, (req.clone(), generation)); self.queue.push_back(req.into_raw(req_id)); } } diff --git a/bdk_electrum_streaming/src/state.rs b/bdk_electrum_streaming/src/state.rs index 26bbeda..8065f24 100644 --- a/bdk_electrum_streaming/src/state.rs +++ b/bdk_electrum_streaming/src/state.rs @@ -18,7 +18,7 @@ use serde_json::from_value; use crate::{ chain_job::ChainJob, - req::{JobRequest, ReqCoord, ReqQueue}, + req::{JobRequest, PoppedRequest, ReqCoord, ReqQueue}, spk_job::SpkJob, DerivedSpkTracker, Update, }; @@ -228,7 +228,11 @@ impl State { } } RawNotificationOrResponse::Response(raw_response) => { - let (orig_req, job_ids) = match self.coord.pop(raw_response.id) { + let PoppedRequest { + request: orig_req, + job_ids, + reorged_since_sent, + } = match self.coord.pop(raw_response.id) { Some(req) => req, None => return Ok(None), }; @@ -249,6 +253,12 @@ impl State { ?err, "Server gave no merkle proof at this height. Reorg?", ); + // An error about the chain we have since left says nothing about + // the block we now have at this height. Discard it and let the + // jobs ask again. + if reorged_since_sent { + return Ok(self.advance_spk_jobs(req_queue, job_ids)); + } // An error proves nothing — it is equally a rate limit, an index // still catching up or a daemon hiccup — so nothing durable is // recorded, and there is nowhere to record it: a proof @@ -302,7 +312,7 @@ impl State { // header a job is waiting on can never arrive. Asking again would loop // for as long as the connection is up, so drop the jobs instead and // let the next notification rebuild them. - if ours.is_some_and(|ours| ours != hash) { + if ours.is_some_and(|ours| ours != hash) && !reorged_since_sent { tracing::warn!( height = req.height, ours = ours.expect("just matched").to_string(), @@ -318,10 +328,11 @@ impl State { // will wake it. So only the insert below is conditional; the jobs are // advanced either way. // - // The insert is skipped when it would extend the checkpoints, and - // when we already have this block. + // The insert is skipped when the block at this height may have been + // replaced since we asked (`reorged_since_sent`), when it would extend + // the checkpoints, and when we already have this block. let extends = req.height > self.cp.height(); - if !extends && ours.is_none() { + if !reorged_since_sent && !extends && ours.is_none() { self.cp = self.cp.clone().insert(BlockId::from((req.height, hash))); } Ok(self.advance_spk_jobs(req_queue, job_ids)) @@ -362,6 +373,14 @@ impl State { JobRequest::GetTxMerkle(req) => { let resp = from_raw(&req, raw)?; + // The proof was built against the server's chain at the time we asked, + // which may not be the block we now have at this height. Checking it + // against that block would read a disagreement between two chains as a + // verdict on this one, so discard it and let the job ask again. + if reorged_since_sent { + return Ok(self.advance_spk_jobs(req_queue, job_ids)); + } + let cp = match self.cp.get(req.height) { Some(cp) if cp.height() == req.height => cp, _ => { @@ -535,6 +554,10 @@ impl State { heights = ?evicted, "Blocks evicted from the local chain. Refetching anchors." ); + // Responses to requests which are still in flight describe the chain we just left + // behind. + self.coord.bump_chain_generation(); + let affected = evicted .iter() .flat_map(|height| self.cache.spk_hashes_by_height.get(height)) diff --git a/bdk_electrum_streaming/tests/state.rs b/bdk_electrum_streaming/tests/state.rs index d60b5e7..63e4293 100644 --- a/bdk_electrum_streaming/tests/state.rs +++ b/bdk_electrum_streaming/tests/state.rs @@ -464,6 +464,83 @@ fn anchor_is_refetched_when_tx_moves_to_another_block_of_same_height() -> anyhow Ok(()) } +/// A reorg can land while an anchor fetch is in flight. The merkle proof we get back was built +/// against the chain the server had when it received the request, so it may not prove inclusion +/// in the block we now have at that height. Verifying it against that block would record a +/// permanent "not in this block" verdict for an anchor which is in fact valid. +#[test] +fn merkle_proof_predating_a_reorg_is_not_taken_as_a_failed_anchor() -> anyhow::Result<()> { + let (descriptor, spk_hash, spk) = tracked_descriptor()?; + let tx = tx_paying(&spk, 50_000); + let txid = tx.compute_txid(); + let (genesis, header_1) = base_headers(); + let header_2 = block_with_tx(&header_1, txid, 200, 0); + // The block which replaces height 2 contains the tx alongside another one, so the tx keeps + // its height — and with it its script status — but needs a different merkle proof. + let proof_2b = response::TxMerkle { + block_height: absolute::Height::from_consensus(2)?, + merkle: vec![sha256d::Hash::hash(b"the other tx")], + pos: 1, + }; + let header_2b = block_with_root(&header_1, proof_2b.expected_merkle_root(txid), 222, 1); + let header_3b = block_with_root(&header_2b, TxMerkleNode::all_zeros(), 300, 0); + + let mut state = new_state(Cache::default(), descriptor, genesis); + let mut queue = ReqQueue::new(); + let mut server = Server { + headers: vec![genesis, header_1, header_2], + spk_hash, + txs: vec![(tx, 2)], + merkle_proof: (Vec::new(), 0), + }; + + // Sync, but hold back the merkle proof so the anchor fetch is still in flight. + state.init(&mut queue); + let mut in_flight = Vec::new(); + while let Some(req) = queue.pop_front() { + if req.method.as_ref() == "blockchain.transaction.get_merkle" { + in_flight.push(req); + continue; + } + state.advance(&mut queue, response(&req, &server))?; + } + let stale_req = match in_flight.as_slice() { + [req] => req.clone(), + reqs => panic!( + "expected exactly one merkle request in flight, got {}", + reqs.len() + ), + }; + let stale_resp = response(&stale_req, &server); + + // The reorg lands before the server answers. + server.headers = vec![genesis, header_1, header_2b, header_3b]; + server.merkle_proof = (proof_2b.merkle.clone(), proof_2b.pos); + state.advance( + &mut queue, + raw_msg(json!({ + "jsonrpc": "2.0", + "method": "blockchain.headers.subscribe", + "params": [{ "hex": serialize_hex(&header_3b), "height": 3 }], + })), + )?; + drain_requests(&mut state, &mut queue, &server); + + // The held answer proves inclusion in the block which was evicted, not in the one which + // replaced it. + state.advance(&mut queue, stale_resp)?; + let updates = drain_requests(&mut state, &mut queue, &server); + + assert!( + updates.iter().any(|u| u + .tx_update + .anchors + .contains(&(anchor_of(&header_2b, 2), txid))), + "the anchor must be refetched rather than written off from a proof of the evicted block" + ); + Ok(()) +} + /// A reorg can land while a job is midway through fetching anchors. Anchors are staged as they /// resolve, so the job must give up the ones it staged against the chain it started on — otherwise /// it goes on to emit them in a single update alongside the ones it resolved against the chain it @@ -602,6 +679,84 @@ fn a_tx_unconfirmed_by_a_reorg_does_not_error_the_connection() -> anyhow::Result Ok(()) } +/// The other half of `merkle_proof_predating_a_reorg_is_not_taken_as_a_failed_anchor`: a server +/// answers a proof request from the chain it had when it *received* it, so if the tx was out of +/// its block at that moment it answers with an error — an error about the chain we have since +/// left. Blaming that on whichever block the reorg put at the height would write off an anchor +/// which is in fact valid. +#[test] +fn merkle_error_predating_a_reorg_is_not_taken_as_a_failed_anchor() -> anyhow::Result<()> { + let (descriptor, spk_hash, spk) = tracked_descriptor()?; + let tx = tx_paying(&spk, 50_000); + let txid = tx.compute_txid(); + let (genesis, header_1) = base_headers(); + let header_2 = block_with_tx(&header_1, txid, 200, 0); + // Height 2 is replaced by a block which contains the tx too, so the anchor is still valid. + let header_2b = block_with_tx(&header_1, txid, 222, 1); + let header_3b = block_with_root(&header_2b, TxMerkleNode::all_zeros(), 300, 0); + + let mut state = new_state(Cache::default(), descriptor, genesis); + let mut queue = ReqQueue::new(); + let mut server = Server { + headers: vec![genesis, header_1, header_2], + spk_hash, + txs: vec![(tx, 2)], + merkle_proof: (Vec::new(), 0), + }; + + // Sync, but hold back the merkle request so the anchor fetch is still in flight. + state.init(&mut queue); + let mut in_flight = Vec::new(); + while let Some(req) = queue.pop_front() { + if req.method.as_ref() == "blockchain.transaction.get_merkle" { + in_flight.push(req); + continue; + } + state.advance(&mut queue, response(&req, &server))?; + } + let stale_req = match in_flight.as_slice() { + [req] => req.clone(), + reqs => panic!( + "expected exactly one merkle request in flight, got {}", + reqs.len() + ), + }; + + // The reorg lands before the server answers. + server.headers = vec![genesis, header_1, header_2b, header_3b]; + state.advance( + &mut queue, + raw_msg(json!({ + "jsonrpc": "2.0", + "method": "blockchain.headers.subscribe", + "params": [{ "hex": serialize_hex(&header_3b), "height": 3 }], + })), + )?; + drain_requests(&mut state, &mut queue, &server); + + // The held request was received while the tx was out of its block, so it is answered with + // an error — the wording is the one romanz/electrs really sends, a bare JSON string which + // conflates a genuine fault with the everyday reorg. + state.advance( + &mut queue, + raw_msg(json!({ + "jsonrpc": "2.0", + "id": stale_req.id, + "error": "tx not found or is unconfirmed", + })), + )?; + let updates = drain_requests(&mut state, &mut queue, &server); + + assert!( + updates.iter().any(|u| u + .tx_update + .anchors + .contains(&(anchor_of(&header_2b, 2), txid))), + "the anchor must be refetched rather than written off from an error about the evicted block" + ); + Ok(()) +} + /// A server error is not a disproof — it is equally a rate limit, an index still catching up or /// a daemon hiccup — so it must not be recorded as one. The transaction stays in the record of /// what was seen at that height, so the next reorg of that height asks again; and the job it @@ -843,6 +998,117 @@ fn anchor_resolves_when_the_chain_is_restored_without_its_headers() -> anyhow::R Ok(()) } +/// A header fetched before a reorg describes the chain we have since left behind, and inserting +/// it would splice a purged block into the checkpoint chain. +/// +/// `extends`/`replaces` do not catch this on their own: they only decline a height the chain +/// already has. The gap they leave open is a *sparse* chain — a restored one, or one whose +/// missing heights sit below the 21-block suffix `ChainJob` rewrites — reorged deeper than that +/// suffix, so the refetch never learns the low block changed too. +#[test] +fn header_fetched_before_a_reorg_is_not_spliced_into_the_chain() -> anyhow::Result<()> { + let (descriptor, spk_hash, spk) = tracked_descriptor()?; + let tx = tx_paying(&spk, 50_000); + let txid = tx.compute_txid(); + let (genesis, header_1) = base_headers(); + + // Two chains which differ at height 2 as well as near the tip. Both contain the tx at + // height 2, so the anchor stays valid throughout — only the block it belongs to changes. + let build = |second: block::Header, tip: u32, nonce: u32| { + let mut chain = vec![genesis, header_1, second]; + for height in 3..=tip { + let prev = *chain.last().expect("non-empty"); + chain.push(block_with_root( + &prev, + TxMerkleNode::all_zeros(), + 1000 + height, + nonce, + )); + } + chain + }; + let chain_a = build(block_with_tx(&header_1, txid, 200, 0), 30, 0); + let chain_b = build(block_with_tx(&header_1, txid, 222, 1), 31, 1); + let (a2, b2) = (chain_a[2], chain_b[2]); + assert_ne!(a2.block_hash(), b2.block_hash()); + + // A restored chain sparse enough that height 2 is a gap — so the anchor has to fetch that + // header, and `replaces` will not decline it when it comes back. + let cp = CheckPoint::new(BlockId { + height: 0, + hash: genesis.block_hash(), + }) + .insert(BlockId { + height: 30, + hash: chain_a[30].block_hash(), + }); + let mut state = new_state_with_cp(Cache::default(), descriptor, cp); + let mut queue = ReqQueue::new(); + let mut server = Server { + headers: chain_a.clone(), + spk_hash, + txs: vec![(tx, 2)], + merkle_proof: (Vec::new(), 0), + }; + + // Sync, but hold back the height-2 header so the fetch is still in flight. + state.init(&mut queue); + let mut in_flight = Vec::new(); + while let Some(req) = queue.pop_front() { + if req.method.as_ref() == "blockchain.block.header" && req.params[0] == json!(2) { + in_flight.push(req); + continue; + } + state.advance(&mut queue, response(&req, &server))?; + } + let stale_req = match in_flight.as_slice() { + [req] => req.clone(), + reqs => panic!("expected one held header request, got {}", reqs.len()), + }; + let stale_resp = response(&stale_req, &server); + + // The reorg lands. It runs deeper than `ChainJob`'s suffix, so the refetch rewrites the + // top 21 blocks and never learns that height 2 changed too. + server.headers = chain_b.clone(); + state.advance( + &mut queue, + raw_msg(json!({ + "jsonrpc": "2.0", + "method": "blockchain.headers.subscribe", + "params": [{ "hex": serialize_hex(&chain_b[31]), "height": 31 }], + })), + )?; + drain_requests(&mut state, &mut queue, &server); + + // The held answer describes the chain we have left behind. + let mut updates = Vec::from_iter(state.advance(&mut queue, stale_resp)?); + updates.extend(drain_requests(&mut state, &mut queue, &server)); + + let tip = updates + .iter() + .rev() + .find_map(|u| u.chain_update.clone()) + .expect("must get a chain update"); + let at_2 = tip.iter().find(|cp| cp.height() == 2); + assert_ne!( + at_2.as_ref().map(|cp| cp.hash()), + Some(a2.block_hash()), + "a header from the chain we left must not be spliced into the checkpoint chain" + ); + assert_eq!( + at_2.map(|cp| cp.hash()), + Some(b2.block_hash()), + "the height must be refetched against the chain we are actually on" + ); + assert!( + updates + .iter() + .any(|u| u.tx_update.anchors.contains(&(anchor_of(&b2, 2), txid))), + "and the anchor must resolve against that block" + ); + Ok(()) +} + /// The refetch is a script hash notification we raise ourselves, so it must not displace one the /// server actually sent. A real notification carries a status at least as new as anything we /// could replay from cache; replacing its job would resolve the script against a stale history From 2d462528031638085fdf07dbf766c171d01e8939 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BF=97=E5=AE=87?= Date: Fri, 21 Aug 2026 10:32:12 +0000 Subject: [PATCH 7/9] feat(bdk_electrum_streaming)!: Make the cache persistable Group the server's per-script histories behind `SpkHistories`, holding each status with the history it stands for so the two cannot desync, and keep the height index private since it is derived from them. Move it and `Cache` to `cache.rs` and give both serde impls. The height index is rebuilt on load rather than stored, and `anchors` is written as a sequence because its tuple key cannot be a JSON map key. BREAKING CHANGE: `Cache::spk_histories` keeps its name but changes type from `HashMap>` to `SpkHistories`, and `Cache` loses `spk_statuses` and `spk_hashes_by_height` to it. The free `SPK_HASHES_BY_HEIGHT_HORIZON` is now `SpkHistories::HEIGHT_INDEX_HORIZON`. Co-Authored-By: Claude Opus 5 --- Cargo.lock | 25 ++ bdk_electrum_streaming/Cargo.toml | 3 +- bdk_electrum_streaming/src/cache.rs | 346 ++++++++++++++++++++++++++ bdk_electrum_streaming/src/lib.rs | 2 + bdk_electrum_streaming/src/spk_job.rs | 74 +++--- bdk_electrum_streaming/src/state.rs | 105 ++------ bdk_electrum_streaming/tests/state.rs | 4 +- 7 files changed, 431 insertions(+), 128 deletions(-) create mode 100644 bdk_electrum_streaming/src/cache.rs diff --git a/Cargo.lock b/Cargo.lock index edac76b..4f540dd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -19,6 +19,18 @@ dependencies = [ "cpufeatures", ] +[[package]] +name = "ahash" +version = "0.8.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75" +dependencies = [ + "cfg-if", + "once_cell", + "version_check", + "zerocopy", +] + [[package]] name = "anyhow" version = "1.0.104" @@ -76,6 +88,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cb3028782f6bf14a6df987244333d34e6b272b5a40a53e4879ec2dfd82275a3a" dependencies = [ "bitcoin", + "hashbrown", + "serde", ] [[package]] @@ -90,6 +104,7 @@ dependencies = [ "futures", "futures-timer", "miniscript", + "serde", "serde_json", "tokio", "tokio-util", @@ -573,6 +588,16 @@ dependencies = [ "r-efi", ] +[[package]] +name = "hashbrown" +version = "0.14.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1" +dependencies = [ + "ahash", + "serde", +] + [[package]] name = "hex-conservative" version = "0.2.2" diff --git a/bdk_electrum_streaming/Cargo.toml b/bdk_electrum_streaming/Cargo.toml index 646d1ab..15e23fa 100644 --- a/bdk_electrum_streaming/Cargo.toml +++ b/bdk_electrum_streaming/Cargo.toml @@ -13,9 +13,10 @@ readme = "README.md" futures = "0.3" futures-timer = "3" anyhow = "1" -bdk_core = "0.6" +bdk_core = { version = "0.6", features = ["serde"] } miniscript = { version = "12.0.0" } electrum_streaming_client = { version = "0.4" } +serde = { version = "1", features = ["derive", "rc"] } serde_json = "1" tracing = "0.1" diff --git a/bdk_electrum_streaming/src/cache.rs b/bdk_electrum_streaming/src/cache.rs new file mode 100644 index 0000000..7de9857 --- /dev/null +++ b/bdk_electrum_streaming/src/cache.rs @@ -0,0 +1,346 @@ +use std::{ + collections::{BTreeMap, BTreeSet, HashMap, HashSet}, + sync::Arc, +}; + +use bdk_core::{ + bitcoin::{self, BlockHash, Transaction, Txid}, + ConfirmationBlockTime, +}; +use electrum_streaming_client::{response, ElectrumScriptHash, ElectrumScriptStatus}; + +/// Everything learned from the server, kept so a reconnect need not ask again. +#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)] +pub struct Cache { + /// The server's per-script histories. + pub spk_histories: SpkHistories, + /// Every txid ever seen for each script hash. + /// + /// Stays here rather than in [`SpkHistories`] because it is the one spk-keyed map that must + /// survive [`SpkHistories::remove`]. Two things read it after the server has stopped + /// reporting a history: evictions are the difference between this set and the history now + /// in hand, so replacing it with the latest history would make that difference empty and + /// no transaction would ever be reported as evicted; and it is the record that a script + /// was *once* active, which keeps its derivation index revealed and the lookahead + /// extended past it. + pub spk_txids: HashMap>, + pub txs: HashMap>, + /// Written as a sequence: a `(Txid, BlockHash)` key is not a string, so a map would be + /// unserializable in JSON and every other format that requires string keys. + #[serde(with = "persist::anchors_as_seq")] + pub anchors: HashMap<(Txid, BlockHash), ConfirmationBlockTime>, + pub headers: HashMap, +} + +/// The last history the server reported for each script hash. +/// +/// The only part of [`Cache`] a caller cannot rebuild from wallet data, since the server reports +/// a script's history as it stands now and will never again mention a transaction it has dropped. +/// +/// Fields are private: the height index is derived from the histories, and letting it be set +/// independently would reintroduce the desync the type exists to prevent. +#[derive(Debug, Clone, Default)] +pub struct SpkHistories { + /// The last history reported for each script hash, with the status it stands for. + /// + /// The status is stored with the history rather than beside it so the two cannot desync: a + /// replay needs the last status, and [`Self::get`] needs to know which status the + /// history it hands back is an answer to. + spk_hash_to_history: HashMap)>, + /// Script hashes whose history reported a transaction at each height. + /// + /// This is what makes a reorg actionable: when the local chain drops a block, the scripts + /// recorded at that height are the ones whose anchors need refetching. Derived from + /// `spk_hash_to_history`, so it is rebuilt on deserialization rather than stored. + height_to_spk_hashes: BTreeMap>, +} + +impl SpkHistories { + /// How far below the tip the height index is retained by [`Self::prune`]. + /// + /// Comfortably above [`ChainJob`]'s 21-block suffix, which bounds how deep an eviction — the + /// only thing that reads the index — can ever reach. + /// + /// [`ChainJob`]: crate::chain_job::ChainJob + pub const HEIGHT_INDEX_HORIZON: u32 = 100; + + /// Drop the history for `spk_hash`, for when the server stops reporting one. + /// + /// Does not touch [`Cache::spk_txids`], which has to outlive this to report the evictions. + pub fn remove(&mut self, spk_hash: ElectrumScriptHash) { + self.spk_hash_to_history.remove(&spk_hash); + for spk_hashes in self.height_to_spk_hashes.values_mut() { + spk_hashes.remove(&spk_hash); + } + } + + /// Record `history` as the answer to `spk_status`, replacing whatever `spk_hash` had before. + pub fn insert( + &mut self, + spk_hash: ElectrumScriptHash, + spk_status: ElectrumScriptStatus, + history: Vec, + ) { + for tx in &history { + if let Some(height) = tx.confirmation_height() { + self.height_to_spk_hashes + .entry(height.to_consensus_u32()) + .or_default() + .insert(spk_hash); + } + } + self.spk_hash_to_history + .insert(spk_hash, (spk_status, history)); + } + + /// The history for `spk_hash`, but only if it is the one `spk_status` stands for. + /// + /// A job fetches for the status its notification carried, so handing it a history that + /// answers an older status would have it finish on stale data. `None` sends it to the + /// server instead, which is why the status is effectively part of the key. + pub fn get( + &self, + spk_hash: ElectrumScriptHash, + spk_status: ElectrumScriptStatus, + ) -> Option<&[response::Tx]> { + match self.spk_hash_to_history.get(&spk_hash)? { + (status, history) if *status == spk_status => Some(history), + _ => None, + } + } + + /// Every script hash whose history reported a transaction at one of `heights`, deduplicated. + /// + /// Given the heights a reorg evicted, these are the scripts whose anchors need refetching. + pub fn spk_hashes_at_heights<'a>( + &'a self, + heights: impl IntoIterator + 'a, + ) -> impl Iterator + 'a { + heights + .into_iter() + .filter_map(|height| self.height_to_spk_hashes.get(&height)) + .flatten() + .copied() + .filter({ + let mut dedup = HashSet::new(); + move |&spk_hash| dedup.insert(spk_hash) + }) + } + + /// The last status the server reported for `spk_hash`, if it still has a history. + pub fn status(&self, spk_hash: ElectrumScriptHash) -> Option { + self.spk_hash_to_history + .get(&spk_hash) + .map(|&(status, _)| status) + } + + /// Drop height index entries too far below `tip_height` for any reorg to reach. + /// + /// The histories themselves are untouched; only the index is bounded. + pub fn prune(&mut self, tip_height: u32) { + if let Some(height) = tip_height.checked_sub(Self::HEIGHT_INDEX_HORIZON) { + self.height_to_spk_hashes = self.height_to_spk_hashes.split_off(&height); + } + } +} + +/// Types and impls that exist only so [`Cache`] can be stored and loaded. +/// +/// Kept apart from the cache itself because [`HistoryTx`] mirrors [`response::Tx`] and the two +/// are easy to mistake for each other at a glance. +mod persist { + use super::*; + + /// A history entry in the shape we can write back out. + /// + /// [`response::Tx`] derives `Deserialize` only, so histories round-trip through this instead. + #[derive(serde::Serialize, serde::Deserialize)] + enum HistoryTx { + Mempool { + txid: Txid, + fee_sats: u64, + confirmed_inputs: bool, + }, + Confirmed { + txid: Txid, + height: u32, + }, + } + + impl From<&response::Tx> for HistoryTx { + fn from(tx: &response::Tx) -> Self { + match tx { + response::Tx::Mempool(tx) => Self::Mempool { + txid: tx.txid, + fee_sats: tx.fee.to_sat(), + confirmed_inputs: tx.confirmed_inputs, + }, + response::Tx::Confirmed(tx) => Self::Confirmed { + txid: tx.txid, + height: tx.height.to_consensus_u32(), + }, + } + } + } + + impl TryFrom for response::Tx { + type Error = bitcoin::absolute::ConversionError; + + fn try_from(tx: HistoryTx) -> Result { + Ok(match tx { + HistoryTx::Mempool { + txid, + fee_sats, + confirmed_inputs, + } => Self::Mempool(response::MempoolTx { + txid, + fee: bitcoin::Amount::from_sat(fee_sats), + confirmed_inputs, + }), + HistoryTx::Confirmed { txid, height } => Self::Confirmed(response::ConfirmedTx { + txid, + height: bitcoin::absolute::Height::from_consensus(height)?, + }), + }) + } + } + + /// Only the histories are written; the height index is rebuilt from them on the way back in. + impl serde::Serialize for SpkHistories { + fn serialize(&self, serializer: S) -> Result { + serializer.collect_map(self.spk_hash_to_history.iter().map( + |(&spk_hash, (status, history))| { + ( + spk_hash, + ( + status, + history.iter().map(HistoryTx::from).collect::>(), + ), + ) + }, + )) + } + } + + impl<'de> serde::Deserialize<'de> for SpkHistories { + fn deserialize>(deserializer: D) -> Result { + use serde::de::Error; + let stored = + HashMap::)>::deserialize( + deserializer, + )?; + let mut spk_histories = Self::default(); + for (spk_hash, (status, history)) in stored { + let history = history + .into_iter() + .map(response::Tx::try_from) + .collect::, _>>() + .map_err(D::Error::custom)?; + spk_histories.insert(spk_hash, status, history); + } + Ok(spk_histories) + } + } + + pub(super) mod anchors_as_seq { + use super::*; + use serde::{Deserialize, Deserializer, Serializer}; + + type Anchors = HashMap<(Txid, BlockHash), ConfirmationBlockTime>; + + pub fn serialize( + anchors: &Anchors, + serializer: S, + ) -> Result { + serializer.collect_seq( + anchors + .iter() + .map(|(&(txid, block_hash), anchor)| (txid, block_hash, anchor)), + ) + } + + pub fn deserialize<'de, D: Deserializer<'de>>( + deserializer: D, + ) -> Result { + Ok( + Vec::<(Txid, BlockHash, ConfirmationBlockTime)>::deserialize(deserializer)? + .into_iter() + .map(|(txid, block_hash, anchor)| ((txid, block_hash), anchor)) + .collect(), + ) + } + } +} + +#[cfg(test)] +mod test { + use super::*; + use bitcoin::hashes::Hash; + + fn txid(byte: u8) -> Txid { + Txid::from_byte_array([byte; 32]) + } + + fn spk_hash(byte: u8) -> ElectrumScriptHash { + ElectrumScriptHash::from_byte_array([byte; 32]) + } + + /// The height index is not stored, so it has to come back from the histories themselves. + #[test] + fn spk_histories_round_trip_rebuilds_the_height_index() { + let history = vec![ + response::Tx::Confirmed(response::ConfirmedTx { + txid: txid(1), + height: bitcoin::absolute::Height::from_consensus(700_000).unwrap(), + }), + response::Tx::Mempool(response::MempoolTx { + txid: txid(2), + fee: bitcoin::Amount::from_sat(1234), + confirmed_inputs: false, + }), + ]; + let status = ElectrumScriptStatus::from_history(&history).expect("history is not empty"); + + let mut before = SpkHistories::default(); + before.insert(spk_hash(9), status, history); + + let json = serde_json::to_string(&before).expect("must serialize"); + assert!( + !json.contains("height_to_spk_hashes"), + "the derived index must not be stored" + ); + let after: SpkHistories = serde_json::from_str(&json).expect("must deserialize"); + + assert_eq!( + after.get(spk_hash(9), status).map(|h| h.len()), + Some(2), + "the history must survive, and still answer to its status" + ); + assert_eq!( + after.spk_hashes_at_heights([700_000]).collect::>(), + vec![spk_hash(9)], + "the confirmed entry must be indexed by height again" + ); + assert_eq!( + after.spk_hashes_at_heights([700_001]).count(), + 0, + "only heights the history actually reported" + ); + assert_eq!(after.status(spk_hash(9)), Some(status)); + } + + /// `anchors` is keyed by a tuple, which JSON cannot use as a map key. + #[test] + fn cache_round_trips_through_json() { + let anchor = (txid(1), bitcoin::BlockHash::from_byte_array([2; 32])); + let mut before = Cache::default(); + before + .anchors + .insert(anchor, ConfirmationBlockTime::default()); + + let json = serde_json::to_string(&before).expect("must serialize"); + let after: Cache = serde_json::from_str(&json).expect("must deserialize"); + + assert_eq!(after.anchors.get(&anchor), before.anchors.get(&anchor)); + } +} diff --git a/bdk_electrum_streaming/src/lib.rs b/bdk_electrum_streaming/src/lib.rs index 99ae8df..9f00fa7 100644 --- a/bdk_electrum_streaming/src/lib.rs +++ b/bdk_electrum_streaming/src/lib.rs @@ -5,6 +5,8 @@ use bdk_core::spk_client::FullScanResponse; pub use electrum_streaming_client; use bdk_core::ConfirmationBlockTime; +mod cache; +pub use cache::*; mod state; use electrum_streaming_client::{ AsyncPendingRequest, BlockingPendingRequest, MaybeBatch, PendingRequest, diff --git a/bdk_electrum_streaming/src/spk_job.rs b/bdk_electrum_streaming/src/spk_job.rs index 70ae6f9..ae7bcc5 100644 --- a/bdk_electrum_streaming/src/spk_job.rs +++ b/bdk_electrum_streaming/src/spk_job.rs @@ -183,45 +183,47 @@ impl SpkJob { tip: &CheckPoint, ) -> (Self, bool) { match self.stage { - SpkJobStage::ProcessingHistory { status } => match cache.spk_histories.get(&status) { - Some(history) => { - if let Some(prev_txids) = cache.spk_txids.get(&self.spk_hash) { - let these_txids = - history.iter().map(|tx| tx.txid()).collect::>(); - let to_evict = prev_txids - .difference(&these_txids) - .map(|&txid| (txid, self.start.as_secs())); - self.tx_update.evicted_ats.extend(to_evict); - } - for tx in history { - if let response::Tx::Mempool(tx) = tx { - self.tx_update - .seen_ats - .insert((tx.txid, self.start.as_secs())); + SpkJobStage::ProcessingHistory { status } => { + match cache.spk_histories.get(self.spk_hash, status) { + Some(history) => { + if let Some(prev_txids) = cache.spk_txids.get(&self.spk_hash) { + let these_txids = + history.iter().map(|tx| tx.txid()).collect::>(); + let to_evict = prev_txids + .difference(&these_txids) + .map(|&txid| (txid, self.start.as_secs())); + self.tx_update.evicted_ats.extend(to_evict); + } + for tx in history { + if let response::Tx::Mempool(tx) = tx { + self.tx_update + .seen_ats + .insert((tx.txid, self.start.as_secs())); + } } - } - let txs = TxsJobStage::from_missing_txs(history.iter().map(|tx| tx.txid())); - let anchors = history - .iter() - .filter_map(|tx| { - let height = tx.confirmation_height()?.to_consensus_u32(); - Some((height, tx.txid())) - }) - .collect(); - self.stage = SpkJobStage::ProcessingTxsAndAnchors { - txs, - anchors, - anchors_resolved: false, - }; - (self, true) - } - None => { - let script_hash = self.spk_hash; - queuer.enqueue(request::GetHistory { script_hash }); - (self, false) + let txs = TxsJobStage::from_missing_txs(history.iter().map(|tx| tx.txid())); + let anchors = history + .iter() + .filter_map(|tx| { + let height = tx.confirmation_height()?.to_consensus_u32(); + Some((height, tx.txid())) + }) + .collect(); + self.stage = SpkJobStage::ProcessingTxsAndAnchors { + txs, + anchors, + anchors_resolved: false, + }; + (self, true) + } + None => { + let script_hash = self.spk_hash; + queuer.enqueue(request::GetHistory { script_hash }); + (self, false) + } } - }, + } SpkJobStage::ProcessingTxsAndAnchors { mut txs, anchors, .. } => { diff --git a/bdk_electrum_streaming/src/state.rs b/bdk_electrum_streaming/src/state.rs index 8065f24..30b052f 100644 --- a/bdk_electrum_streaming/src/state.rs +++ b/bdk_electrum_streaming/src/state.rs @@ -1,15 +1,9 @@ -use std::{ - collections::{btree_map, BTreeMap, BTreeSet, HashMap}, - sync::Arc, -}; +use std::collections::{btree_map, BTreeMap, BTreeSet}; use anyhow::Context; -use bdk_core::{ - bitcoin::{self, BlockHash, Transaction, Txid}, - BlockId, CheckPoint, ConfirmationBlockTime, -}; +use bdk_core::{BlockId, CheckPoint, ConfirmationBlockTime}; use electrum_streaming_client::{ - notification::Notification, request, response, AsyncPendingRequest, BlockingPendingRequest, + notification::Notification, request, AsyncPendingRequest, BlockingPendingRequest, ElectrumScriptHash, ElectrumScriptStatus, MaybeBatch, PendingRequest, RawNotificationOrResponse, Request, }; @@ -17,6 +11,7 @@ use miniscript::{Descriptor, DescriptorPublicKey}; use serde_json::from_value; use crate::{ + cache::{Cache, SpkHistories}, chain_job::ChainJob, req::{JobRequest, PoppedRequest, ReqCoord, ReqQueue}, spk_job::SpkJob, @@ -82,6 +77,10 @@ impl State { &self.cache } + pub fn spk_histories(&self) -> &SpkHistories { + &self.cache.spk_histories + } + /// Reset the state to be not initialized. /// /// Call this after disconnection otherwise pending requests will not be resent and no @@ -192,7 +191,7 @@ impl State { let mut last_active_indices = BTreeMap::new(); if spk_status.is_none() { - self.forget_spk_history(spk_hash); + self.cache.spk_histories.remove(spk_hash); } if spk_status.is_some() || self.cache.spk_txids.contains_key(&spk_hash) { @@ -340,28 +339,14 @@ impl State { JobRequest::GetHistory(req) => { let resp = from_raw(&req, raw)?; if let Some(spk_status) = ElectrumScriptStatus::from_history(&resp) { - self.cache - .spk_histories - .entry(spk_status) - .or_default() - .extend(resp.clone()); self.cache .spk_txids .entry(req.script_hash) .or_default() .extend(resp.iter().map(|tx| tx.txid())); - // Recorded together with the history, so replaying this script's - // job always finds the history its status stands for. - self.cache.spk_statuses.insert(req.script_hash, spk_status); - for tx in &resp { - if let Some(height) = tx.confirmation_height() { - self.cache - .spk_hashes_by_height - .entry(height.to_consensus_u32()) - .or_default() - .insert(req.script_hash); - } - } + self.cache + .spk_histories + .insert(req.script_hash, spk_status, resp); } Ok(self.advance_spk_jobs(req_queue, job_ids)) } @@ -455,7 +440,7 @@ impl State { let mut last_active_indices = BTreeMap::new(); if spk_status.is_none() { - self.forget_spk_history(spk_hash); + self.cache.spk_histories.remove(spk_hash); } if spk_status.is_some() || self.cache.spk_txids.contains_key(&spk_hash) { @@ -505,18 +490,6 @@ impl State { } } - /// Forget the history the server no longer reports for `spk_hash`. - /// - /// A replay is built from the last status and the heights that status was seen at, so - /// leaving them behind would have a later eviction rebuild this script's job from a history - /// it no longer has. - fn forget_spk_history(&mut self, spk_hash: ElectrumScriptHash) { - self.cache.spk_statuses.remove(&spk_hash); - for spk_hashes in self.cache.spk_hashes_by_height.values_mut() { - spk_hashes.remove(&spk_hash); - } - } - /// Apply the pending chain job to the local chain, if it has everything it needs. /// /// Returns the resulting update, if the job completed. @@ -558,22 +531,16 @@ impl State { // behind. self.coord.bump_chain_generation(); - let affected = evicted - .iter() - .flat_map(|height| self.cache.spk_hashes_by_height.get(height)) - .flatten() - .copied(); - // Start each affected script the job its own notification would have started, from // the status and history already cached: a script hash notification we raise // ourselves, because the server will not raise one. A transaction which moved to a // different block of the same height leaves the status untouched. - for spk_hash in affected { + for spk_hash in self.cache.spk_histories.spk_hashes_at_heights(evicted) { // A vacant entry is the whole rule: never displace a job the server's own // notification built, since that one carries a status at least as new as // anything we could replay, and every stashed job is re-advanced below anyway. if let btree_map::Entry::Vacant(e) = self.spk_jobs.entry(spk_hash) { - if let Some(&status) = self.cache.spk_statuses.get(&spk_hash) { + if let Some(status) = self.cache.spk_histories.status(spk_hash) { e.insert(SpkJob::new(&self.cache, spk_hash, Some(status))); } } @@ -583,8 +550,7 @@ impl State { // Only heights inside `ChainJob`'s reorg window can ever be evicted, and only evicted // heights are ever read back, so entries far below the tip can never be consulted // again. Pruning keeps this bounded with a wide margin over that horizon. - let prune_below = cp.height().saturating_sub(SPK_HASHES_BY_HEIGHT_HORIZON); - self.cache.spk_hashes_by_height = self.cache.spk_hashes_by_height.split_off(&prune_below); + self.cache.spk_histories.prune(cp.height()); let stashed_jobs = self .spk_jobs @@ -669,42 +635,3 @@ where { from_value(raw) } - -/// A monotonically growing cache. -#[derive(Debug, Clone, Default)] -pub struct Cache { - pub spk_histories: HashMap>, - pub spk_txids: HashMap>, - pub txs: HashMap>, - pub anchors: HashMap<(Txid, BlockHash), ConfirmationBlockTime>, - pub headers: HashMap, - /// The last status the server reported for each script hash. - /// - /// Replaying a script's job needs its status, since that is the key its history is cached - /// under. Only recorded together with that history, so a replay always finds one. - pub spk_statuses: HashMap, - /// Script hashes whose history reported a transaction at each height. - /// - /// This is what makes a reorg actionable: when the local chain drops a block, the scripts - /// recorded at that height are the ones whose anchors need refetching. - /// - /// Unlike the rest of the cache this is pruned. Its only reader is the eviction path, and - /// evictions can only come from a conflict inside the window [`ChainJob`] rewrites, so - /// entries more than [`SPK_HASHES_BY_HEIGHT_HORIZON`] below the tip can never be read - /// again. - /// - /// That window is also the reorg horizon the anchor refetch inherits: a fork deeper than - /// [`ChainJob`]'s suffix length leaves the checkpoint chain claiming blocks the server does - /// not have, and no eviction is reported for them. - /// - /// [`ChainJob`]: crate::chain_job::ChainJob - pub spk_hashes_by_height: BTreeMap>, -} - -/// How far below the tip [`Cache::spk_hashes_by_height`] is retained. -/// -/// Comfortably above [`ChainJob`]'s 21-block suffix, which bounds how deep an eviction — the -/// only thing that reads the map — can ever reach. -/// -/// [`ChainJob`]: crate::chain_job::ChainJob -pub const SPK_HASHES_BY_HEIGHT_HORIZON: u32 = 100; diff --git a/bdk_electrum_streaming/tests/state.rs b/bdk_electrum_streaming/tests/state.rs index 63e4293..3ec1fa1 100644 --- a/bdk_electrum_streaming/tests/state.rs +++ b/bdk_electrum_streaming/tests/state.rs @@ -1302,7 +1302,7 @@ fn a_script_whose_history_goes_away_is_not_replayed() -> anyhow::Result<()> { "tx must first be anchored" ); assert!( - state.cache().spk_statuses.contains_key(&spk_hash), + state.spk_histories().status(spk_hash).is_some(), "the status must be recorded while the script has a history" ); @@ -1318,7 +1318,7 @@ fn a_script_whose_history_goes_away_is_not_replayed() -> anyhow::Result<()> { )?; drain_requests(&mut state, &mut queue, &server); assert!( - !state.cache().spk_statuses.contains_key(&spk_hash), + !state.spk_histories().status(spk_hash).is_some(), "a null status must drop the recorded status" ); From a11cc719e92991ac9e32125a42d235e967244604 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BF=97=E5=AE=87?= Date: Fri, 21 Aug 2026 10:43:30 +0000 Subject: [PATCH 8/9] test(bdk_electrum_streaming): Cover an anchor resolved before its tx Every existing test has the server answer `GetTx` before the merkle proof, so the anchors resolve on the job's final pass and nothing runs after them. Pin the other order, where a later pass re-resolves anchors already in hand. Co-Authored-By: Claude Opus 5 --- bdk_electrum_streaming/tests/state.rs | 38 +++++++++++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/bdk_electrum_streaming/tests/state.rs b/bdk_electrum_streaming/tests/state.rs index 3ec1fa1..c2c1050 100644 --- a/bdk_electrum_streaming/tests/state.rs +++ b/bdk_electrum_streaming/tests/state.rs @@ -1439,3 +1439,41 @@ fn a_proof_for_another_block_is_not_a_verdict_on_ours() -> anyhow::Result<()> { ); Ok(()) } + +/// A job needs both the transactions and their anchors, and the server answers in any order. +/// +/// Every other test lets the `GetTx` land first, which puts the anchors on the job's final pass +/// with nothing running after them. This forces the other order: the merkle proof arrives first, +/// so the anchor resolves early and the job runs again when the transaction finally lands. +/// +/// Anchors are re-resolved from scratch on every pass, so that later pass must not lose the +/// anchor already in hand — the set of what to anchor is the question, not the answer, and +/// clearing it once answered would have the next pass ask an empty question and stage nothing. +#[test] +fn anchor_survives_a_pass_that_happens_after_it_resolved() -> anyhow::Result<()> { + let (descriptor, spk_hash, spk) = tracked_descriptor()?; + let tx = tx_paying(&spk, 50_000); + let txid = tx.compute_txid(); + let (genesis, header_1) = base_headers(); + let header_2 = block_with_tx(&header_1, txid, 200, 0); + + let mut state = new_state(Cache::default(), descriptor, genesis); + let mut queue = ReqQueue::new(); + let server = Server { + headers: vec![genesis, header_1, header_2], + spk_hash, + txs: vec![(tx, 2)], + merkle_proof: (Vec::new(), 0), + }; + + state.init(&mut queue); + let updates = drain_requests_proofs_first(&mut state, &mut queue, &server); + assert!( + updates.iter().any(|u| u + .tx_update + .anchors + .contains(&(anchor_of(&header_2, 2), txid))), + "the anchor must still be emitted after a later pass" + ); + Ok(()) +} From 674a2a3040dea318ecb1b52eb62b0850619e28f8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BF=97=E5=AE=87?= Date: Fri, 21 Aug 2026 11:40:39 +0000 Subject: [PATCH 9/9] fix(bdk_electrum_streaming)!: Never settle on a chain the server has left Two ways a chain job could write a chain the server is not on into the local checkpoint chain. Neither recovers on its own: the server has already announced the tip it is on, and will not announce it again until the next block. `ChainJob::new` replaces `chain_job` on every tip notification, but the new job asks for the same heights as the one it replaced, so its `request::Headers` is byte-identical and `ReqQueuer` deduplicates it against the request already in flight. Nothing is sent, and the answer already on its way fills the new job with the chain the server has just left. The chain job's in-flight requests are now forgotten when it is replaced. A request another job still wants stays in flight and merely loses `JobId::Chain` as an owner; one that nothing else wants is dropped outright, so the stale answer is ignored on arrival and the replacement's request is actually sent. A batch is also answered from whichever chain the server holds when it replies, not the one it announced when the request went out. A reorg in between returns blocks for a tip no notification ever mentioned, and that batch is the reorg's only appearance. `ChainJob` now keeps the block it was created to reach and checks the assembled update against it. A mismatch abandons the job rather than retrying: the server only answers from a chain it has moved to, and moving is what makes it announce a new tip, so the notification that rebuilds the job is already on its way. Neither half subsumes the other. Forgetting the replaced job's requests is what gets a request back on the wire; the target check is what stops an unannounced chain being believed once one is answered. Fixes #19. BREAKING CHANGE: `ChainJob::try_finish` returns `ChainJobOutcome` rather than `Result`. Co-Authored-By: Claude Opus 5 --- bdk_electrum_streaming/src/chain_job.rs | 55 +++++++++- bdk_electrum_streaming/src/req.rs | 18 +++ bdk_electrum_streaming/src/state.rs | 16 ++- bdk_electrum_streaming/tests/state.rs | 139 ++++++++++++++++++++++++ 4 files changed, 222 insertions(+), 6 deletions(-) diff --git a/bdk_electrum_streaming/src/chain_job.rs b/bdk_electrum_streaming/src/chain_job.rs index e11d156..f03c27c 100644 --- a/bdk_electrum_streaming/src/chain_job.rs +++ b/bdk_electrum_streaming/src/chain_job.rs @@ -16,10 +16,35 @@ use std::collections::{BTreeMap, BTreeSet}; /// [`process_blocks()`]: ChainJob::process_blocks #[derive(Debug, Clone)] pub struct ChainJob { + /// The block this job was created to reach. + /// + /// A `blockchain.block.headers` batch is answered from whichever chain the server holds when + /// it *replies*, which need not be the one it announced. Keeping the target lets + /// [`try_finish()`] tell the two apart instead of writing an unannounced chain into the + /// checkpoint chain. + /// + /// [`try_finish()`]: ChainJob::try_finish + target: BlockId, missing_headers: BTreeSet, cp_update: BTreeMap, } +/// The outcome of [`ChainJob::try_finish`]. +#[derive(Debug)] +pub enum ChainJobOutcome { + /// Every header arrived, and they agree with the block the job was created for. + Finished(CheckPoint), + /// Still waiting on headers. + Pending(ChainJob), + /// The headers arrived but disagree with the block the job was created for, so they describe + /// a chain other than the one announced. + /// + /// The job is abandoned rather than retried: the server only answers from a chain it has + /// moved to, and moving is what makes it announce a new tip, so the notification that + /// rebuilds the job is already on its way. + Superseded, +} + impl ChainJob { const CHAIN_SUFFIX_LENGTH: u32 = 21; @@ -32,6 +57,10 @@ impl ChainJob { header: Header, height: u32, ) -> Option { + let target = BlockId { + height, + hash: header.block_hash(), + }; let cp = local_tip .iter() .find(|cp| cp.height() <= height) @@ -46,6 +75,7 @@ impl ChainJob { if let Some(prev_height) = height.checked_sub(1) { if prev_height == prev_cp.height() && header.prev_blockhash == prev_cp.hash() { return Some(Self { + target, missing_headers: BTreeSet::new(), cp_update: core::iter::once((height, header.block_hash())).collect(), }); @@ -68,6 +98,7 @@ impl ChainJob { count, }); Some(Self { + target, missing_headers: (start_height..=remote_height).collect(), cp_update: BTreeMap::new(), }) @@ -82,6 +113,7 @@ impl ChainJob { count: (remote_height + 1 - remote_start_height) as usize, }); Some(Self { + target, missing_headers: (local_start_height..=local_height) .chain(remote_start_height..=remote_height) .collect(), @@ -105,13 +137,30 @@ impl ChainJob { self } - pub fn try_finish(self, local_tip: &mut CheckPoint) -> Result { + pub fn try_finish(self, local_tip: &mut CheckPoint) -> ChainJobOutcome { if !self.missing_headers.is_empty() { tracing::trace!( missing = self.missing_headers.len(), "Chain job not finished" ); - return Err(self); + return ChainJobOutcome::Pending(self); + } + + // The batch is only evidence about the chain it was answered from. If it does not put + // the announced block at the announced height, that is a chain we were never told about + // and must not adopt — the same reasoning the merkle and header handlers apply to a + // proof built against a block other than ours. + if self.cp_update.get(&self.target.height) != Some(&self.target.hash) { + tracing::info!( + height = self.target.height, + announced = self.target.hash.to_string(), + received = self + .cp_update + .get(&self.target.height) + .map(|h| h.to_string()), + "Headers describe a chain other than the one announced. Abandoning chain job.", + ); + return ChainJobOutcome::Superseded; } let mut cp = local_tip.clone(); @@ -124,6 +173,6 @@ impl ChainJob { tip_hash = cp.hash().to_string(), "Chain job finished" ); - Ok(cp) + ChainJobOutcome::Finished(cp) } } diff --git a/bdk_electrum_streaming/src/req.rs b/bdk_electrum_streaming/src/req.rs index 85745f7..54ecce6 100644 --- a/bdk_electrum_streaming/src/req.rs +++ b/bdk_electrum_streaming/src/req.rs @@ -138,6 +138,24 @@ impl ReqCoord { }) } + /// Forget every request `job_id` is still waiting on. + /// + /// A request wanted by another job stays in flight and merely loses `job_id` as an owner. + /// One that nothing else wants is dropped outright, so its response is ignored on arrival + /// and an identical request is no longer deduplicated against it. + pub fn forget_job(&mut self, job_id: JobId) { + let mut orphaned = Vec::new(); + self.req_to_job.retain(|req, job_ids| { + if !job_ids.remove(&job_id) || !job_ids.is_empty() { + return true; + } + orphaned.push(req.clone()); + false + }); + self.awaiting_responses + .retain(|_, (req, _)| !orphaned.contains(req)); + } + /// To be called when the local chain drops blocks. /// /// Requests already in flight were made against the old chain, so their responses can no diff --git a/bdk_electrum_streaming/src/state.rs b/bdk_electrum_streaming/src/state.rs index 30b052f..e6e57af 100644 --- a/bdk_electrum_streaming/src/state.rs +++ b/bdk_electrum_streaming/src/state.rs @@ -12,7 +12,7 @@ use serde_json::from_value; use crate::{ cache::{Cache, SpkHistories}, - chain_job::ChainJob, + chain_job::{ChainJob, ChainJobOutcome}, req::{JobRequest, PoppedRequest, ReqCoord, ReqQueue}, spk_job::SpkJob, DerivedSpkTracker, Update, @@ -167,6 +167,13 @@ impl State { let header = *header_notification.header(); self.cache.headers.insert(header.block_hash(), header); + // A new tip supersedes whatever the previous job was resolving against, + // and its request has to go with it. The replacement job asks for the + // same heights, so a surviving request would be deduplicated against — + // nothing would be sent, and the answer already in flight would fill the + // new job with the chain the server has just left. + self.coord.forget_job(JobId::Chain); + // Always replace prev job since a new notification means a new tip. self.chain_job = ChainJob::new( self.coord.queuer(req_queue, JobId::Chain), @@ -497,11 +504,14 @@ impl State { let job = self.chain_job.take()?; let prev_cp = self.cp.clone(); match job.try_finish(&mut self.cp) { - Ok(cp) => Some(self.on_chain_job_completed(req_queue, &prev_cp, cp)), - Err(job) => { + ChainJobOutcome::Finished(cp) => { + Some(self.on_chain_job_completed(req_queue, &prev_cp, cp)) + } + ChainJobOutcome::Pending(job) => { self.chain_job = Some(job); None } + ChainJobOutcome::Superseded => None, } } diff --git a/bdk_electrum_streaming/tests/state.rs b/bdk_electrum_streaming/tests/state.rs index c2c1050..934eab9 100644 --- a/bdk_electrum_streaming/tests/state.rs +++ b/bdk_electrum_streaming/tests/state.rs @@ -1477,3 +1477,142 @@ fn anchor_survives_a_pass_that_happens_after_it_resolved() -> anyhow::Result<()> ); Ok(()) } + +/// A tip notification landing while the previous one's headers are still in flight. +/// +/// The replacement job asks for the same heights, so its request is byte-identical to the one +/// already out. Deduplicated against that one, nothing new is sent, and the answer already on its +/// way describes the chain the server has just left. +#[test] +fn a_tip_that_moves_while_headers_are_in_flight_is_not_lost() -> anyhow::Result<()> { + let (descriptor, spk_hash, _spk) = tracked_descriptor()?; + let (genesis, header_1) = base_headers(); + let build = |nonce: u32| { + let mut chain = vec![genesis, header_1]; + for height in 2..=3 { + let prev = *chain.last().expect("non-empty"); + chain.push(block_with_root( + &prev, + TxMerkleNode::all_zeros(), + 1000 + height, + nonce, + )); + } + chain + }; + let (chain_a, chain_b) = (build(0), build(1)); + assert_ne!(chain_a[3].block_hash(), chain_b[3].block_hash()); + + let mut state = new_state(Cache::default(), descriptor, genesis); + let mut queue = ReqQueue::new(); + let mut server = Server { + headers: chain_a.clone(), + spk_hash, + txs: Vec::new(), + merkle_proof: (Vec::new(), 0), + }; + + let notify = |chain: &[block::Header]| { + raw_msg(json!({ + "jsonrpc": "2.0", + "method": "blockchain.headers.subscribe", + "params": [{ "hex": serialize_hex(chain.last().expect("non-empty")), "height": 3 }], + })) + }; + + // The A-chain tip. Hold its headers request so the job is still waiting. + state.advance(&mut queue, notify(&chain_a))?; + let held = queue + .drain(..) + .filter(|req| req.method.as_ref() == "blockchain.block.headers") + .collect::>(); + assert!(!held.is_empty(), "the job must have asked for headers"); + let held_resp = held + .iter() + .map(|req| response(req, &server)) + .collect::>(); + + // The server reorgs to B and notifies the new tip at the same height. + server.headers = chain_b.clone(); + state.advance(&mut queue, notify(&chain_b))?; + assert!( + queue + .iter() + .any(|req| req.method.as_ref() == "blockchain.block.headers"), + "the replacement job must send its own request rather than adopt the one in flight" + ); + + // The held answer describes the chain the server has left. + let mut updates = Vec::new(); + for resp in held_resp { + updates.extend(state.advance(&mut queue, resp)?); + } + updates.extend(drain_requests(&mut state, &mut queue, &server)); + + let tip = updates + .iter() + .rev() + .find_map(|u| u.chain_update.clone()) + .expect("must get a chain update"); + assert_eq!( + tip.hash(), + chain_b[3].block_hash(), + "the local chain must end on the tip the server actually has" + ); + Ok(()) +} + +/// A headers batch answered from a chain other than the one announced. +/// +/// `blockchain.block.headers` is answered from whichever chain the server holds when it *replies*, +/// so a reorg between receiving the request and answering it returns blocks for a tip we were +/// never told about. Adopting them would put the checkpoint chain on a chain no notification ever +/// announced, and no notification would arrive to correct it. +#[test] +fn headers_for_a_chain_we_were_not_told_about_are_not_adopted() -> anyhow::Result<()> { + let (descriptor, spk_hash, _spk) = tracked_descriptor()?; + let (genesis, h1) = base_headers(); + let h2 = block_with_root(&h1, TxMerkleNode::all_zeros(), 200, 0); + let a3 = block_with_root(&h2, TxMerkleNode::all_zeros(), 300, 0); + let b3 = block_with_root(&h2, TxMerkleNode::all_zeros(), 300, 1); + assert_ne!(a3.block_hash(), b3.block_hash()); + + let mut state = new_state(Cache::default(), descriptor, genesis); + let mut queue = ReqQueue::new(); + let mut server = Server { + headers: vec![genesis, h1, h2], + spk_hash, + txs: Vec::new(), + merkle_proof: (Vec::new(), 0), + }; + state.init(&mut queue); + drain_requests(&mut state, &mut queue, &server); + + // A3 is announced, so that is the block the job is created to reach. + state.advance( + &mut queue, + raw_msg(json!({ + "jsonrpc": "2.0", + "method": "blockchain.headers.subscribe", + "params": [{ "hex": serialize_hex(&a3), "height": 3 }], + })), + )?; + + // But by the time the server answers, it is on B3 — and it does not announce it, because + // this response is the reorg's only appearance. + server.headers = vec![genesis, h1, h2, b3]; + let updates = drain_requests(&mut state, &mut queue, &server); + + let tip = updates.iter().rev().find_map(|u| u.chain_update.clone()); + assert_ne!( + tip.as_ref().map(|cp| cp.hash()), + Some(b3.block_hash()), + "a chain no notification announced must not be adopted" + ); + assert_ne!( + tip.as_ref().map(|cp| cp.hash()), + Some(a3.block_hash()), + "and the announced block was never actually delivered" + ); + Ok(()) +}