-use crate::error::{TxSyncError, InternalError};
-use crate::common::{SyncState, FilterQueue, ConfirmedTx};
+use crate::common::{ConfirmedTx, FilterQueue, SyncState};
+use crate::error::{InternalError, TxSyncError};
-use lightning::util::logger::Logger;
-use lightning::{log_error, log_debug, log_trace};
use lightning::chain::WatchedOutput;
use lightning::chain::{Confirm, Filter};
+use lightning::util::logger::Logger;
+use lightning::{log_debug, log_error, log_trace};
use bitcoin::{BlockHash, Script, Txid};
-use esplora_client::Builder;
-#[cfg(feature = "async-interface")]
-use esplora_client::r#async::AsyncClient;
#[cfg(not(feature = "async-interface"))]
use esplora_client::blocking::BlockingClient;
+#[cfg(feature = "async-interface")]
+use esplora_client::r#async::AsyncClient;
+use esplora_client::Builder;
-use std::collections::HashSet;
use core::ops::Deref;
+use std::collections::HashSet;
/// Synchronizes LDK with a given [`Esplora`] server.
///
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();
}
/// 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());
- Self {
- sync_state,
- queue,
- client,
- logger,
- }
+ Self { sync_state, queue, client, logger }
}
/// Synchronizes the given `confirmables` via their [`Confirm`] interface implementations. This
/// [`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<C: Deref>(&self, confirmables: Vec<C>) -> 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();
num_unconfirmed += unconfirmed_txs.len();
sync_state.sync_unconfirmed_transactions(
&confirmables,
- unconfirmed_txs
+ unconfirmed_txs,
);
- }
+ },
Err(err) => {
// (Semi-)permanent failure, retry later.
log_error!(self.logger,
);
sync_state.pending_sync = true;
return Err(TxSyncError::from(err));
- }
+ },
}
},
Err(err) => {
);
sync_state.pending_sync = true;
return Err(TxSyncError::from(err));
- }
+ },
}
- match maybe_await!(self.sync_best_block_updated(&confirmables, &tip_hash)) {
- Ok(()) => {}
+ 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.
- log_debug!(self.logger, "Encountered inconsistency during transaction sync, restarting.");
+ log_debug!(
+ self.logger,
+ "Encountered inconsistency during transaction sync, restarting."
+ );
sync_state.pending_sync = true;
continue;
- }
+ },
Err(err) => {
// (Semi-)permanent failure, retry later.
log_error!(self.logger,
);
sync_state.pending_sync = true;
return Err(TxSyncError::from(err));
- }
+ },
}
}
continue;
}
num_confirmed += confirmed_txs.len();
- sync_state.sync_confirmed_transactions(
- &confirmables,
- confirmed_txs
- );
- }
+ sync_state
+ .sync_confirmed_transactions(&confirmables, confirmed_txs);
+ },
Err(err) => {
// (Semi-)permanent failure, retry later.
log_error!(self.logger,
);
sync_state.pending_sync = true;
return Err(TxSyncError::from(err));
- }
+ },
}
- }
+ },
Err(InternalError::Inconsistency) => {
// Immediately restart syncing when we encounter any inconsistencies.
- log_debug!(self.logger, "Encountered inconsistency during transaction sync, restarting.");
+ log_debug!(
+ self.logger,
+ "Encountered inconsistency during transaction sync, restarting."
+ );
sync_state.pending_sync = true;
continue;
- }
+ },
Err(err) => {
// (Semi-)permanent failure, retry later.
log_error!(self.logger,
);
sync_state.pending_sync = true;
return Err(TxSyncError::from(err));
- }
+ },
}
sync_state.last_sync_hash = Some(tip_hash);
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);
+ 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);
+ 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<C: Deref>(
+ &self, confirmables: &Vec<C>, 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))?;
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);
fn get_confirmed_transactions(
&self, sync_state: &SyncState,
) -> Result<Vec<ConfirmedTx>, InternalError> {
-
// 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<ConfirmedTx> = 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);
}
}
for (_, output) in &sync_state.watched_outputs {
- if let Some(output_status) = maybe_await!(self.client
+ if let Some(output_status) = maybe_await!(self
+ .client
.get_output_status(&output.outpoint.txid, output.outpoint.index as u64))?
{
if let Some(spending_txid) = output_status.txid {
if let Some(spending_tx_status) = output_status.status {
- if let Some(confirmed_tx) = maybe_await!(self
- .get_confirmed_tx(
- &spending_txid,
- spending_tx_status.block_hash,
- spending_tx_status.block_height,
- ))?
- {
+ 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_tx_status.block_hash,
+ spending_tx_status.block_height,
+ ))? {
confirmed_txs.push(confirmed_tx);
}
}
#[maybe_async]
fn get_confirmed_tx(
- &self, txid: &Txid, expected_block_hash: Option<BlockHash>, known_block_height: Option<u32>,
+ &self, txid: Txid, expected_block_hash: Option<BlockHash>, known_block_height: Option<u32>,
) -> Result<Option<ConfirmedTx>, InternalError> {
if let Some(merkle_block) = maybe_await!(self.client.get_merkle_block(&txid))? {
let block_header = merkle_block.header;
let block_hash = block_header.block_hash();
if let Some(expected_block_hash) = expected_block_hash {
if expected_block_hash != block_hash {
- log_trace!(self.logger, "Inconsistency: Tx {} expected in block {}, but is confirmed in {}", txid, expected_block_hash, block_hash);
+ log_trace!(
+ self.logger,
+ "Inconsistency: Tx {} expected in block {}, but is confirmed in {}",
+ txid,
+ expected_block_hash,
+ block_hash
+ );
return Err(InternalError::Inconsistency);
}
}
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);
}
// 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.
- log_trace!(self.logger, "Inconsistency: Tx {} was unconfirmed during syncing.", txid);
+ log_trace!(
+ self.logger,
+ "Inconsistency: Tx {} was unconfirmed during syncing.",
+ txid
+ );
return Err(InternalError::Inconsistency);
}
}
}
#[maybe_async]
- fn get_unconfirmed_transactions(
- &self, confirmables: &Vec<&(dyn Confirm + Sync + Send)>,
- ) -> Result<Vec<Txid>, InternalError> {
+ fn get_unconfirmed_transactions<C: Deref>(
+ &self, confirmables: &Vec<C>,
+ ) -> Result<Vec<Txid>, 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
}
/// 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
}
#[cfg(not(feature = "async-interface"))]
type EsploraClientType = BlockingClient;
-
impl<L: Deref> Filter for EsploraSyncClient<L>
where
L::Target: Logger,
projects / rust-lightning / blobdiff