X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=lightning-transaction-sync%2Fsrc%2Fesplora.rs;h=82c49b9f9d2e8dfda23ea941ae41f5d5e3ebf943;hb=777ce7b059c0e81cbb36813d9ad028f767985fdd;hp=953f8b0718c3526cff701142c17a0a5d7e32e339;hpb=f07f4b90f8de76d594328e11e36d094cdb936097;p=rust-lightning diff --git a/lightning-transaction-sync/src/esplora.rs b/lightning-transaction-sync/src/esplora.rs index 953f8b07..82c49b9f 100644 --- a/lightning-transaction-sync/src/esplora.rs +++ b/lightning-transaction-sync/src/esplora.rs @@ -14,7 +14,6 @@ use esplora_client::r#async::AsyncClient; #[cfg(not(feature = "async-interface"))] use esplora_client::blocking::BlockingClient; -use std::time::Instant; use std::collections::HashSet; use core::ops::Deref; @@ -52,7 +51,7 @@ where pub fn new(server_url: String, logger: L) -> Self { let builder = Builder::new(&server_url); #[cfg(not(feature = "async-interface"))] - let client = builder.build_blocking().unwrap(); + let client = builder.build_blocking(); #[cfg(feature = "async-interface")] let client = builder.build_async().unwrap(); @@ -60,6 +59,8 @@ where } /// Returns a new [`EsploraSyncClient`] object using the given Esplora client. + /// + /// This is not exported to bindings users as the underlying client from BDK is not exported. pub fn from_client(client: EsploraClientType, logger: L) -> Self { let sync_state = MutexType::new(SyncState::new()); let queue = std::sync::Mutex::new(FilterQueue::new()); @@ -83,7 +84,9 @@ where /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager /// [`Filter`]: lightning::chain::Filter #[maybe_async] - pub fn sync(&self, confirmables: Vec<&(dyn Confirm + Sync + Send)>) -> Result<(), TxSyncError> { + pub fn sync(&self, confirmables: Vec) -> Result<(), TxSyncError> + where C::Target: Confirm + { // This lock makes sure we're syncing once at a time. #[cfg(not(feature = "async-interface"))] let mut sync_state = self.sync_state.lock().unwrap(); @@ -91,7 +94,8 @@ where let mut sync_state = self.sync_state.lock().await; log_trace!(self.logger, "Starting transaction sync."); - let start_time = Instant::now(); + #[cfg(feature = "time")] + let start_time = std::time::Instant::now(); let mut num_confirmed = 0; let mut num_unconfirmed = 0; @@ -153,7 +157,7 @@ where } } - match maybe_await!(self.sync_best_block_updated(&confirmables, &tip_hash)) { + match maybe_await!(self.sync_best_block_updated(&confirmables, &mut sync_state, &tip_hash)) { Ok(()) => {} Err(InternalError::Inconsistency) => { // Immediately restart syncing when we encounter any inconsistencies. @@ -227,16 +231,21 @@ where sync_state.pending_sync = false; } } + #[cfg(feature = "time")] log_debug!(self.logger, "Finished transaction sync at tip {} in {}ms: {} confirmed, {} unconfirmed.", tip_hash, start_time.elapsed().as_millis(), num_confirmed, num_unconfirmed); + #[cfg(not(feature = "time"))] + log_debug!(self.logger, "Finished transaction sync at tip {}: {} confirmed, {} unconfirmed.", + tip_hash, num_confirmed, num_unconfirmed); Ok(()) } #[maybe_async] - fn sync_best_block_updated( - &self, confirmables: &Vec<&(dyn Confirm + Sync + Send)>, tip_hash: &BlockHash, - ) -> Result<(), InternalError> { - + fn sync_best_block_updated( + &self, confirmables: &Vec, sync_state: &mut SyncState, tip_hash: &BlockHash, + ) -> Result<(), InternalError> + where C::Target: Confirm + { // Inform the interface of the new block. let tip_header = maybe_await!(self.client.get_header_by_hash(tip_hash))?; let tip_status = maybe_await!(self.client.get_block_status(&tip_hash))?; @@ -245,6 +254,9 @@ where for c in confirmables { c.best_block_updated(&tip_header, tip_height); } + + // Prune any sufficiently confirmed output spends + sync_state.prune_output_spends(tip_height); } } else { return Err(InternalError::Inconsistency); @@ -260,10 +272,13 @@ where // First, check the confirmation status of registered transactions as well as the // status of dependent transactions of registered outputs. - let mut confirmed_txs = Vec::new(); + let mut confirmed_txs: Vec = Vec::new(); for txid in &sync_state.watched_transactions { - if let Some(confirmed_tx) = maybe_await!(self.get_confirmed_tx(&txid, None, None))? { + if confirmed_txs.iter().any(|ctx| ctx.txid == *txid) { + continue; + } + if let Some(confirmed_tx) = maybe_await!(self.get_confirmed_tx(*txid, None, None))? { confirmed_txs.push(confirmed_tx); } } @@ -274,9 +289,19 @@ where { if let Some(spending_txid) = output_status.txid { if let Some(spending_tx_status) = output_status.status { + if confirmed_txs.iter().any(|ctx| ctx.txid == spending_txid) { + if spending_tx_status.confirmed { + // Skip inserting duplicate ConfirmedTx entry + continue; + } else { + log_trace!(self.logger, "Inconsistency: Detected previously-confirmed Tx {} as unconfirmed", spending_txid); + return Err(InternalError::Inconsistency); + } + } + if let Some(confirmed_tx) = maybe_await!(self .get_confirmed_tx( - &spending_txid, + spending_txid, spending_tx_status.block_hash, spending_tx_status.block_height, ))? @@ -299,7 +324,7 @@ where #[maybe_async] fn get_confirmed_tx( - &self, txid: &Txid, expected_block_hash: Option, known_block_height: Option, + &self, txid: Txid, expected_block_hash: Option, known_block_height: Option, ) -> Result, InternalError> { if let Some(merkle_block) = maybe_await!(self.client.get_merkle_block(&txid))? { let block_header = merkle_block.header; @@ -314,7 +339,7 @@ where let mut matches = Vec::new(); let mut indexes = Vec::new(); let _ = merkle_block.txn.extract_matches(&mut matches, &mut indexes); - if indexes.len() != 1 || matches.len() != 1 || matches[0] != *txid { + if indexes.len() != 1 || matches.len() != 1 || matches[0] != txid { log_error!(self.logger, "Retrieved Merkle block for txid {} doesn't match expectations. This should not happen. Please verify server integrity.", txid); return Err(InternalError::Failed); } @@ -322,14 +347,19 @@ where // unwrap() safety: len() > 0 is checked above let pos = *indexes.first().unwrap() as usize; if let Some(tx) = maybe_await!(self.client.get_tx(&txid))? { + if tx.txid() != txid { + log_error!(self.logger, "Retrieved transaction for txid {} doesn't match expectations. This should not happen. Please verify server integrity.", txid); + return Err(InternalError::Failed); + } + if let Some(block_height) = known_block_height { // We can take a shortcut here if a previous call already gave us the height. - return Ok(Some(ConfirmedTx { tx, block_header, pos, block_height })); + return Ok(Some(ConfirmedTx { tx, txid, block_header, pos, block_height })); } let block_status = maybe_await!(self.client.get_block_status(&block_hash))?; if let Some(block_height) = block_status.height { - return Ok(Some(ConfirmedTx { tx, block_header, pos, block_height })); + return Ok(Some(ConfirmedTx { tx, txid, block_header, pos, block_height })); } else { // If any previously-confirmed block suddenly is no longer confirmed, we found // an inconsistency and should start over. @@ -342,9 +372,11 @@ where } #[maybe_async] - fn get_unconfirmed_transactions( - &self, confirmables: &Vec<&(dyn Confirm + Sync + Send)>, - ) -> Result, InternalError> { + fn get_unconfirmed_transactions( + &self, confirmables: &Vec, + ) -> Result, InternalError> + where C::Target: Confirm + { // Query the interface for relevant txids and check whether the relevant blocks are still // in the best chain, mark them unconfirmed otherwise let relevant_txids = confirmables @@ -372,6 +404,8 @@ where } /// Returns a reference to the underlying esplora client. + /// + /// This is not exported to bindings users as the underlying client from BDK is not exported. pub fn client(&self) -> &EsploraClientType { &self.client }