diff --git a/Cargo.toml b/Cargo.toml index 8b441fa58..520dcec4d 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -58,7 +58,7 @@ bdk_chain = { version = "0.23.3", default-features = false, features = ["std"] } bdk_esplora = { version = "0.22.2", default-features = false, features = ["async-https-rustls", "tokio"]} bdk_electrum = { version = "0.24.0", default-features = false, features = ["use-rustls-ring"]} bdk_wallet = { version = "3.1.0", default-features = false, features = ["std", "keys-bip39"]} -bip157 = { version = "0.6.0", default-features = false } +bip157 = { version = "0.6.1", default-features = false } bitreq = { version = "0.3", default-features = false, features = ["async-https", "json-using-serde"] } rustls = { version = "0.23", default-features = false } diff --git a/src/chain/cbf.rs b/src/chain/cbf.rs index 20b52fd84..7649370d7 100644 --- a/src/chain/cbf.rs +++ b/src/chain/cbf.rs @@ -66,10 +66,38 @@ enum CbfRuntimeStatus { #[derive(Clone, Copy)] enum CbfSyncState { - Active { applied_tip: Option }, + Active { + /// Highest tip whose preceding chain updates have been applied to all listeners. + applied_tip: Option, + /// Whether kyoto has reported catching up to the network tip (via `FiltersSynced`) and + /// the resulting blocks have been applied. `wait_until_synced` blocks until this is set. + /// + /// This must not be derived from a locally-sampled chain tip: kyoto does not persist, so a + /// freshly (re)started node's local header chain sits at genesis until it syncs from peers. + /// Comparing against that would make `wait_until_synced` return before any sync happens. + synced_to_tip: bool, + }, Failed(Error), } +/// Marks that we are applying a block past the last `FiltersSynced` tip, so a `sync_wallets` call +/// issued after new blocks are mined waits for the next `FiltersSynced` rather than returning on a +/// stale `synced_to_tip`. Only flips (and notifies waiters) when currently set. +/// +/// Called both when a new block's filter is received (before it is fetched and applied) and after +/// it is applied, so `synced_to_tip` reflects "behind by an unapplied block" as soon as we learn +/// that block exists, not only once we've finished catching up to it. +fn mark_syncing(sync_state_tx: &watch::Sender) { + // Copy the current state out and drop the `watch` read guard before calling `send_replace`: + // `borrow()` holds a read lock for the lifetime of its temporary, and `send_replace` takes + // the write lock, so holding the borrow across it deadlocks. `CbfSyncState` is `Copy`, so the + // deref copies and the guard is released at the end of this statement. + let current = *sync_state_tx.borrow(); + if let CbfSyncState::Active { applied_tip, synced_to_tip: true } = current { + sync_state_tx.send_replace(CbfSyncState::Active { applied_tip, synced_to_tip: false }); + } +} + /// Struct for holding cbf chain source pub struct CbfChainSource { /// Trusted peer addresses for kyoto's `Builder::add_peers`. @@ -91,24 +119,13 @@ pub struct CbfChainSource { logger: Arc, } +#[derive(Debug)] enum ChainOp { - ConnectFull { - block: IndexedBlock, - }, - ConnectFiltered { - header: Header, - height: u32, - }, - Disconnect { - fork_point: BlockLocator, - }, - /// Marks reaching the chain tip. - Synced { - tip_height: u32, - }, - Failed { - error: Error, - }, + ConnectFull { block: IndexedBlock }, + ConnectFiltered { header: Header, height: u32 }, + Disconnect { fork_point: BlockLocator }, + Synced { tip_height: u32 }, + Failed { error: Error }, } struct BlockApplicator { @@ -140,6 +157,7 @@ impl BlockApplicator { } self.chain_listener.block_connected(&ib.block, ib.height); self.next_height += 1; + mark_syncing(&self.sync_state_tx); if let Some(cache) = &self.block_fee_cache { let fee_rate = coinbase_fee_rate(&ib.block, ib.height); cache @@ -160,12 +178,14 @@ impl BlockApplicator { } self.chain_listener.filtered_block_connected(&header, &[], height); self.next_height += 1; + mark_syncing(&self.sync_state_tx); }, ChainOp::Disconnect { fork_point } => { self.chain_listener.blocks_disconnected(fork_point); self.next_height = fork_point.height + 1; self.sync_state_tx.send_replace(CbfSyncState::Active { applied_tip: Some(fork_point.height), + synced_to_tip: false, }); }, ChainOp::Synced { tip_height } => { @@ -180,8 +200,10 @@ impl BlockApplicator { self.next_height ); } + log_info!(self.logger, "we set new tip and published at {}", tip_height); }, ChainOp::Failed { error } => { + log_info!(self.logger, "we received error chain op {}", error); self.sync_state_tx.send_replace(CbfSyncState::Failed(error)); }, } @@ -192,14 +214,23 @@ impl BlockApplicator { let already_published = { let sync_state = *self.sync_state_tx.borrow(); match sync_state { - CbfSyncState::Active { applied_tip } => applied_tip, + CbfSyncState::Active { applied_tip, .. } => applied_tip, CbfSyncState::Failed(_) => None, } }; if already_published.map_or(false, |published_height| published_height >= tip_height) { + // Even if the applied tip is unchanged, we have now confirmed we are caught up to the + // network tip, so ensure the synced flag is set for any `wait_until_synced` waiter. + self.sync_state_tx.send_replace(CbfSyncState::Active { + applied_tip: already_published, + synced_to_tip: true, + }); return; } - self.sync_state_tx.send_replace(CbfSyncState::Active { applied_tip: Some(tip_height) }); + self.sync_state_tx.send_replace(CbfSyncState::Active { + applied_tip: Some(tip_height), + synced_to_tip: true, + }); let unix_time_secs_opt = SystemTime::now().duration_since(UNIX_EPOCH).ok().map(|d| d.as_secs()); if let Err(e) = update_and_persist_node_metrics( @@ -225,9 +256,12 @@ const FEE_WINDOW_BLOCKS: u32 = BLOCK_FEE_CACHE_CAPACITY as u32; /// Electrum fee sources. Coinbase-derived rates are frequently zero on regtest/signet. const CBF_MIN_FEERATE_SAT_PER_KWU: u64 = 250; -/// Per-block timeout when downloading a block to derive its coinbase fee rate. Kept short so a -/// slow peer only delays a single sample rather than the whole fee update. -const CBF_FEE_BLOCK_FETCH_TIMEOUT_SECS: u64 = 10; +/// Per-attempt timeout when downloading a block from a peer — used both for matched blocks we apply +/// to the listeners and for the coinbase-fee-rate samples. Kyoto queues the request and awaits a +/// peer response with no timeout of its own, so a slow or unresponsive peer would otherwise park the +/// fetch forever. Kept short so a single request is bounded and can be retried (or, for fees, only +/// delays one sample) rather than stalling. +const CBF_BLOCK_FETCH_TIMEOUT_SECS: u64 = 10; /// Recent per-block coinbase-derived fee rates, keyed by height so we can window on the tip, evict /// stale entries, and detect reorged-out blocks (a height whose cached hash no longer matches the @@ -276,7 +310,8 @@ impl CbfChainSource { }; let registered_scripts = Arc::new(Mutex::new(HashSet::new())); let cbf_runtime_status = Arc::new(Mutex::new(CbfRuntimeStatus::Stopped)); - let (sync_state_tx, _) = watch::channel(CbfSyncState::Active { applied_tip: None }); + let (sync_state_tx, _) = + watch::channel(CbfSyncState::Active { applied_tip: None, synced_to_tip: false }); Ok(Self { trusted_peers, fee_source, @@ -343,8 +378,10 @@ impl CbfChainSource { _ => None, }; let best_block_height = chain_listener.get_best_block().height; - self.sync_state_tx - .send_replace(CbfSyncState::Active { applied_tip: Some(best_block_height) }); + self.sync_state_tx.send_replace(CbfSyncState::Active { + applied_tip: Some(best_block_height), + synced_to_tip: false, + }); let block_applicator = BlockApplicator { next_height: best_block_height + 1, sync_state_tx: self.sync_state_tx.clone(), @@ -393,6 +430,7 @@ impl CbfChainSource { Arc::clone(&restart_cbf_runtime_status), ops_tx.clone(), Arc::clone(&restart_listener.onchain_wallet), + restart_sync_state_tx.clone(), )); match current_node.run().await { @@ -460,6 +498,7 @@ impl CbfChainSource { *status = CbfRuntimeStatus::Started { requester: new_requester }; restart_sync_state_tx.send_replace(CbfSyncState::Active { applied_tip: Some(restart_listener.get_best_block().height), + synced_to_tip: false, }); } @@ -495,21 +534,20 @@ impl CbfChainSource { } pub(crate) async fn wait_until_synced(&self) -> Result<(), Error> { - let requester = match &*self.cbf_runtime_status.lock().expect("lock") { - CbfRuntimeStatus::Started { requester } => requester.clone(), - CbfRuntimeStatus::Stopped => return Err(Error::NotRunning), - }; - let target_tip = requester.chain_tip().await.map_err(|e| { - log_error!(self.logger, "Failed to fetch CBF chain tip before syncing: {:?}", e); - Error::TxSyncFailed - })?; - let target_height = target_tip.height; + if matches!(&*self.cbf_runtime_status.lock().expect("lock"), CbfRuntimeStatus::Stopped) { + return Err(Error::NotRunning); + } let mut sync_state_rx = self.sync_state_tx.subscribe(); + // Wait for kyoto to report catching up to the network tip (a `FiltersSynced`-driven + // `synced_to_tip`) and for the resulting blocks to be applied. We must not target a + // locally-sampled `chain_tip()`: kyoto does not persist, so a freshly (re)started node's + // local header chain sits at genesis until it syncs from peers, which would let this return + // before any sync happens. loop { match *sync_state_rx.borrow() { - CbfSyncState::Active { applied_tip } => { - if applied_tip.map_or(false, |applied_height| applied_height >= target_height) { + CbfSyncState::Active { synced_to_tip, .. } => { + if synced_to_tip { return Ok(()); } }, @@ -542,11 +580,17 @@ impl CbfChainSource { logger: Arc, mut event_rx: mpsc::UnboundedReceiver, registered_scripts: Arc>>, cbf_runtime_status: Arc>, ops_tx: mpsc::UnboundedSender, - onchain_wallet: Arc, + onchain_wallet: Arc, sync_state_tx: watch::Sender, ) { while let Some(event) = event_rx.recv().await { match event { Event::IndexedFilter(indexed_filter) => { + // A new block's filter arrived, so we're behind by at least this block until it + // is fetched (if matched) and applied. Flip this before the fetch, not after, + // so a `sync_wallets` call issued in between doesn't return on a stale + // `synced_to_tip` that predates this block. + mark_syncing(&sync_state_tx); + let requester = match &*cbf_runtime_status.lock().expect("lock") { CbfRuntimeStatus::Started { requester } => requester.clone(), CbfRuntimeStatus::Stopped => { @@ -559,7 +603,7 @@ impl CbfChainSource { //each time we receive an IndexedFilter event, we ask bdk to give us all //revealed scripts. We create all_scripts starting from onchain wallet's //scripts and extend them with LDK's ones - let mut all_scripts = onchain_wallet.list_revealed_scripts(); + let mut all_scripts = onchain_wallet.list_watched_scripts(); all_scripts.extend(registered_scripts.lock().expect("lock").iter().cloned()); let block_hash = indexed_filter.block_hash(); @@ -583,42 +627,37 @@ impl CbfChainSource { }, }; - match handle.await { - Ok(Ok(block)) => break block, - Ok(Err(e)) if attempt < CBF_BLOCK_FETCH_RETRIES => { + // Bound the download so an unresponsive peer can't park the fetch forever, + // then flatten the three error layers (timeout / receiver dropped / fetch + // error) into a single reason so the retry-or-fail decision is written once. + let fetched = tokio::time::timeout( + Duration::from_secs(CBF_BLOCK_FETCH_TIMEOUT_SECS), + handle, + ) + .await + .map_err(|_| { + format!("timed out after {}s", CBF_BLOCK_FETCH_TIMEOUT_SECS) + }) + .and_then(|recv| recv.map_err(|_| "receiver was dropped".to_string())) + .and_then(|fetch| fetch.map_err(|e| format!("failed: {:?}", e))); + + match fetched { + Ok(block) => break block, + Err(reason) if attempt < CBF_BLOCK_FETCH_RETRIES => { log_debug!( logger, - "CBF block fetch for {} failed on attempt {}: {:?}; retrying", - block_hash, - attempt, - e - ); - }, - Ok(Err(e)) => { - log_error!( - logger, - "CBF block fetch for {} failed after {} attempts: {:?}", - block_hash, - CBF_BLOCK_FETCH_RETRIES, - e - ); - let _ = - ops_tx.send(ChainOp::Failed { error: Error::TxSyncFailed }); - return; - }, - Err(_) if attempt < CBF_BLOCK_FETCH_RETRIES => { - log_debug!( - logger, - "CBF block receiver for {} dropped on attempt {}; retrying", + "CBF block fetch for {} {} on attempt {}; retrying", block_hash, + reason, attempt ); }, - Err(_) => { + Err(reason) => { log_error!( logger, - "CBF block receiver for {} dropped after {} attempts", + "CBF block fetch for {} {} after {} attempts; giving up", block_hash, + reason, CBF_BLOCK_FETCH_RETRIES ); let _ = @@ -632,9 +671,7 @@ impl CbfChainSource { let height = indexed_filter.height(); //TODO we need to recheck that a particular height has not been //reorganized, and we retrieve indeed the same block header that we - //received `IndexedFilter` event of. right now this would block - //the further sync, as we cannot apply blocks in order. - //Future solution would use something like `get_header_by_hash`. + //received `IndexedFilter` event of. match requester.get_header(height).await { Ok(Some(indexed_header)) => { if indexed_header.block_hash() != block_hash { @@ -713,10 +750,6 @@ impl CbfChainSource { self.registered_scripts.lock().expect("lock").insert(output.script_pubkey); } - // pub(crate) fn register_script(&self, script: ScriptBuf) { - // self.registered_scripts.lock().expect("lock").insert(script); - // } - pub(crate) async fn continuously_update_fee_rate_estimates( &self, mut stop_sync_receiver: watch::Receiver<()>, ) { @@ -935,7 +968,7 @@ impl CbfChainSource { } match tokio::time::timeout( - Duration::from_secs(CBF_FEE_BLOCK_FETCH_TIMEOUT_SECS), + Duration::from_secs(CBF_BLOCK_FETCH_TIMEOUT_SECS), requester.average_fee_rate(canonical_hash), ) .await diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index e25c259d4..690ab7caa 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -216,14 +216,12 @@ impl Wallet { Ok(()) } - pub(crate) fn list_revealed_scripts(&self) -> Vec { - self.inner - .lock() - .expect("lock") - .spk_index() - .revealed_spks(..) - .map(|((_keychain, _index), spk)| spk) - .collect() + /// Returns every script pubkey the wallet is watching for on-chain activity: all revealed + /// SPKs plus the lookahead window BDK derives beyond the last revealed index on each keychain. + /// A block may pay an address we have not explicitly revealed yet (e.g. on recovery, where a fresh + /// wallet has revealed nothing) but which is still within the gap limit. + pub(crate) fn list_watched_scripts(&self) -> Vec { + self.inner.lock().expect("lock").spk_index().inner().all_spks().values().cloned().collect() } async fn update_payment_store(&self, mut events: Vec) -> Result<(), Error> {