]> git.bitcoin.ninja Git - rust-lightning/blob - lightning-transaction-sync/src/esplora.rs
Merge pull request #3054 from TheBlueMatt/2024-04-fuzz-bolt11
[rust-lightning] / lightning-transaction-sync / src / esplora.rs
1 use crate::error::{TxSyncError, InternalError};
2 use crate::common::{SyncState, FilterQueue, ConfirmedTx};
3
4 use lightning::util::logger::Logger;
5 use lightning::{log_error, log_debug, log_trace};
6 use lightning::chain::WatchedOutput;
7 use lightning::chain::{Confirm, Filter};
8
9 use bitcoin::{BlockHash, Script, Txid};
10
11 use esplora_client::Builder;
12 #[cfg(feature = "async-interface")]
13 use esplora_client::r#async::AsyncClient;
14 #[cfg(not(feature = "async-interface"))]
15 use esplora_client::blocking::BlockingClient;
16
17 use std::collections::HashSet;
18 use core::ops::Deref;
19
20 /// Synchronizes LDK with a given [`Esplora`] server.
21 ///
22 /// Needs to be registered with a [`ChainMonitor`] via the [`Filter`] interface to be informed of
23 /// transactions and outputs to monitor for on-chain confirmation, unconfirmation, and
24 /// reconfirmation.
25 ///
26 /// Note that registration via [`Filter`] needs to happen before any calls to
27 /// [`Watch::watch_channel`] to ensure we get notified of the items to monitor.
28 ///
29 /// This uses and exposes either a blocking or async client variant dependent on whether the
30 /// `esplora-blocking` or the `esplora-async` feature is enabled.
31 ///
32 /// [`Esplora`]: https://github.com/Blockstream/electrs
33 /// [`ChainMonitor`]: lightning::chain::chainmonitor::ChainMonitor
34 /// [`Watch::watch_channel`]: lightning::chain::Watch::watch_channel
35 /// [`Filter`]: lightning::chain::Filter
36 pub struct EsploraSyncClient<L: Deref>
37 where
38         L::Target: Logger,
39 {
40         sync_state: MutexType<SyncState>,
41         queue: std::sync::Mutex<FilterQueue>,
42         client: EsploraClientType,
43         logger: L,
44 }
45
46 impl<L: Deref> EsploraSyncClient<L>
47 where
48         L::Target: Logger,
49 {
50         /// Returns a new [`EsploraSyncClient`] object.
51         pub fn new(server_url: String, logger: L) -> Self {
52                 let builder = Builder::new(&server_url);
53                 #[cfg(not(feature = "async-interface"))]
54                 let client = builder.build_blocking().unwrap();
55                 #[cfg(feature = "async-interface")]
56                 let client = builder.build_async().unwrap();
57
58                 EsploraSyncClient::from_client(client, logger)
59         }
60
61         /// Returns a new [`EsploraSyncClient`] object using the given Esplora client.
62         pub fn from_client(client: EsploraClientType, logger: L) -> Self {
63                 let sync_state = MutexType::new(SyncState::new());
64                 let queue = std::sync::Mutex::new(FilterQueue::new());
65                 Self {
66                         sync_state,
67                         queue,
68                         client,
69                         logger,
70                 }
71         }
72
73         /// Synchronizes the given `confirmables` via their [`Confirm`] interface implementations. This
74         /// method should be called regularly to keep LDK up-to-date with current chain data.
75         ///
76         /// For example, instances of [`ChannelManager`] and [`ChainMonitor`] can be informed about the
77         /// newest on-chain activity related to the items previously registered via the [`Filter`]
78         /// interface.
79         ///
80         /// [`Confirm`]: lightning::chain::Confirm
81         /// [`ChainMonitor`]: lightning::chain::chainmonitor::ChainMonitor
82         /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
83         /// [`Filter`]: lightning::chain::Filter
84         #[maybe_async]
85         pub fn sync(&self, confirmables: Vec<&(dyn Confirm + Sync + Send)>) -> Result<(), TxSyncError> {
86                 // This lock makes sure we're syncing once at a time.
87                 #[cfg(not(feature = "async-interface"))]
88                 let mut sync_state = self.sync_state.lock().unwrap();
89                 #[cfg(feature = "async-interface")]
90                 let mut sync_state = self.sync_state.lock().await;
91
92                 log_trace!(self.logger, "Starting transaction sync.");
93                 #[cfg(feature = "time")]
94                 let start_time = std::time::Instant::now();
95                 let mut num_confirmed = 0;
96                 let mut num_unconfirmed = 0;
97
98                 let mut tip_hash = maybe_await!(self.client.get_tip_hash())?;
99
100                 loop {
101                         let pending_registrations = self.queue.lock().unwrap().process_queues(&mut sync_state);
102                         let tip_is_new = Some(tip_hash) != sync_state.last_sync_hash;
103
104                         // We loop until any registered transactions have been processed at least once, or the
105                         // tip hasn't been updated during the last iteration.
106                         if !sync_state.pending_sync && !pending_registrations && !tip_is_new {
107                                 // Nothing to do.
108                                 break;
109                         } else {
110                                 // Update the known tip to the newest one.
111                                 if tip_is_new {
112                                         // First check for any unconfirmed transactions and act on it immediately.
113                                         match maybe_await!(self.get_unconfirmed_transactions(&confirmables)) {
114                                                 Ok(unconfirmed_txs) => {
115                                                         // Double-check the tip hash. If it changed, a reorg happened since
116                                                         // we started syncing and we need to restart last-minute.
117                                                         match maybe_await!(self.client.get_tip_hash()) {
118                                                                 Ok(check_tip_hash) => {
119                                                                         if check_tip_hash != tip_hash {
120                                                                                 tip_hash = check_tip_hash;
121
122                                                                                 log_debug!(self.logger, "Encountered inconsistency during transaction sync, restarting.");
123                                                                                 sync_state.pending_sync = true;
124                                                                                 continue;
125                                                                         }
126                                                                         num_unconfirmed += unconfirmed_txs.len();
127                                                                         sync_state.sync_unconfirmed_transactions(
128                                                                                 &confirmables,
129                                                                                 unconfirmed_txs
130                                                                         );
131                                                                 }
132                                                                 Err(err) => {
133                                                                         // (Semi-)permanent failure, retry later.
134                                                                         log_error!(self.logger,
135                                                                                 "Failed during transaction sync, aborting. Synced so far: {} confirmed, {} unconfirmed.",
136                                                                                 num_confirmed,
137                                                                                 num_unconfirmed
138                                                                                 );
139                                                                         sync_state.pending_sync = true;
140                                                                         return Err(TxSyncError::from(err));
141                                                                 }
142                                                         }
143                                                 },
144                                                 Err(err) => {
145                                                         // (Semi-)permanent failure, retry later.
146                                                         log_error!(self.logger,
147                                                                 "Failed during transaction sync, aborting. Synced so far: {} confirmed, {} unconfirmed.",
148                                                                 num_confirmed,
149                                                                 num_unconfirmed
150                                                         );
151                                                         sync_state.pending_sync = true;
152                                                         return Err(TxSyncError::from(err));
153                                                 }
154                                         }
155
156                                         match maybe_await!(self.sync_best_block_updated(&confirmables, &mut sync_state, &tip_hash)) {
157                                                 Ok(()) => {}
158                                                 Err(InternalError::Inconsistency) => {
159                                                         // Immediately restart syncing when we encounter any inconsistencies.
160                                                         log_debug!(self.logger, "Encountered inconsistency during transaction sync, restarting.");
161                                                         sync_state.pending_sync = true;
162                                                         continue;
163                                                 }
164                                                 Err(err) => {
165                                                         // (Semi-)permanent failure, retry later.
166                                                         log_error!(self.logger,
167                                                                 "Failed during transaction sync, aborting. Synced so far: {} confirmed, {} unconfirmed.",
168                                                                 num_confirmed,
169                                                                 num_unconfirmed
170                                                         );
171                                                         sync_state.pending_sync = true;
172                                                         return Err(TxSyncError::from(err));
173                                                 }
174                                         }
175                                 }
176
177                                 match maybe_await!(self.get_confirmed_transactions(&sync_state)) {
178                                         Ok(confirmed_txs) => {
179                                                 // Double-check the tip hash. If it changed, a reorg happened since
180                                                 // we started syncing and we need to restart last-minute.
181                                                 match maybe_await!(self.client.get_tip_hash()) {
182                                                         Ok(check_tip_hash) => {
183                                                                 if check_tip_hash != tip_hash {
184                                                                         tip_hash = check_tip_hash;
185
186                                                                         log_debug!(self.logger,
187                                                                                 "Encountered inconsistency during transaction sync, restarting.");
188                                                                         sync_state.pending_sync = true;
189                                                                         continue;
190                                                                 }
191                                                                 num_confirmed += confirmed_txs.len();
192                                                                 sync_state.sync_confirmed_transactions(
193                                                                         &confirmables,
194                                                                         confirmed_txs
195                                                                 );
196                                                         }
197                                                         Err(err) => {
198                                                                 // (Semi-)permanent failure, retry later.
199                                                                 log_error!(self.logger,
200                                                                         "Failed during transaction sync, aborting. Synced so far: {} confirmed, {} unconfirmed.",
201                                                                         num_confirmed,
202                                                                         num_unconfirmed
203                                                                 );
204                                                                 sync_state.pending_sync = true;
205                                                                 return Err(TxSyncError::from(err));
206                                                         }
207                                                 }
208                                         }
209                                         Err(InternalError::Inconsistency) => {
210                                                 // Immediately restart syncing when we encounter any inconsistencies.
211                                                 log_debug!(self.logger, "Encountered inconsistency during transaction sync, restarting.");
212                                                 sync_state.pending_sync = true;
213                                                 continue;
214                                         }
215                                         Err(err) => {
216                                                 // (Semi-)permanent failure, retry later.
217                                                 log_error!(self.logger,
218                                                         "Failed during transaction sync, aborting. Synced so far: {} confirmed, {} unconfirmed.",
219                                                         num_confirmed,
220                                                         num_unconfirmed
221                                                 );
222                                                 sync_state.pending_sync = true;
223                                                 return Err(TxSyncError::from(err));
224                                         }
225                                 }
226                                 sync_state.last_sync_hash = Some(tip_hash);
227                                 sync_state.pending_sync = false;
228                         }
229                 }
230                 #[cfg(feature = "time")]
231                 log_debug!(self.logger, "Finished transaction sync at tip {} in {}ms: {} confirmed, {} unconfirmed.",
232                                 tip_hash, start_time.elapsed().as_millis(), num_confirmed, num_unconfirmed);
233                 #[cfg(not(feature = "time"))]
234                 log_debug!(self.logger, "Finished transaction sync at tip {}: {} confirmed, {} unconfirmed.",
235                                 tip_hash, num_confirmed, num_unconfirmed);
236                 Ok(())
237         }
238
239         #[maybe_async]
240         fn sync_best_block_updated(
241                 &self, confirmables: &Vec<&(dyn Confirm + Sync + Send)>, sync_state: &mut SyncState, tip_hash: &BlockHash,
242         ) -> Result<(), InternalError> {
243
244                 // Inform the interface of the new block.
245                 let tip_header = maybe_await!(self.client.get_header_by_hash(tip_hash))?;
246                 let tip_status = maybe_await!(self.client.get_block_status(&tip_hash))?;
247                 if tip_status.in_best_chain {
248                         if let Some(tip_height) = tip_status.height {
249                                 for c in confirmables {
250                                         c.best_block_updated(&tip_header, tip_height);
251                                 }
252
253                                 // Prune any sufficiently confirmed output spends
254                                 sync_state.prune_output_spends(tip_height);
255                         }
256                 } else {
257                         return Err(InternalError::Inconsistency);
258                 }
259                 Ok(())
260         }
261
262         #[maybe_async]
263         fn get_confirmed_transactions(
264                 &self, sync_state: &SyncState,
265         ) -> Result<Vec<ConfirmedTx>, InternalError> {
266
267                 // First, check the confirmation status of registered transactions as well as the
268                 // status of dependent transactions of registered outputs.
269
270                 let mut confirmed_txs: Vec<ConfirmedTx> = Vec::new();
271
272                 for txid in &sync_state.watched_transactions {
273                         if confirmed_txs.iter().any(|ctx| ctx.txid == *txid) {
274                                 continue;
275                         }
276                         if let Some(confirmed_tx) = maybe_await!(self.get_confirmed_tx(*txid, None, None))? {
277                                 confirmed_txs.push(confirmed_tx);
278                         }
279                 }
280
281                 for (_, output) in &sync_state.watched_outputs {
282                         if let Some(output_status) = maybe_await!(self.client
283                                 .get_output_status(&output.outpoint.txid, output.outpoint.index as u64))?
284                         {
285                                 if let Some(spending_txid) = output_status.txid {
286                                         if let Some(spending_tx_status) = output_status.status {
287                                                 if confirmed_txs.iter().any(|ctx| ctx.txid == spending_txid) {
288                                                         if spending_tx_status.confirmed {
289                                                                 // Skip inserting duplicate ConfirmedTx entry
290                                                                 continue;
291                                                         } else {
292                                                                 log_trace!(self.logger, "Inconsistency: Detected previously-confirmed Tx {} as unconfirmed", spending_txid);
293                                                                 return Err(InternalError::Inconsistency);
294                                                         }
295                                                 }
296
297                                                 if let Some(confirmed_tx) = maybe_await!(self
298                                                         .get_confirmed_tx(
299                                                                 spending_txid,
300                                                                 spending_tx_status.block_hash,
301                                                                 spending_tx_status.block_height,
302                                                         ))?
303                                                 {
304                                                         confirmed_txs.push(confirmed_tx);
305                                                 }
306                                         }
307                                 }
308                         }
309                 }
310
311                 // Sort all confirmed transactions first by block height, then by in-block
312                 // position, and finally feed them to the interface in order.
313                 confirmed_txs.sort_unstable_by(|tx1, tx2| {
314                         tx1.block_height.cmp(&tx2.block_height).then_with(|| tx1.pos.cmp(&tx2.pos))
315                 });
316
317                 Ok(confirmed_txs)
318         }
319
320         #[maybe_async]
321         fn get_confirmed_tx(
322                 &self, txid: Txid, expected_block_hash: Option<BlockHash>, known_block_height: Option<u32>,
323         ) -> Result<Option<ConfirmedTx>, InternalError> {
324                 if let Some(merkle_block) = maybe_await!(self.client.get_merkle_block(&txid))? {
325                         let block_header = merkle_block.header;
326                         let block_hash = block_header.block_hash();
327                         if let Some(expected_block_hash) = expected_block_hash {
328                                 if expected_block_hash != block_hash {
329                                         log_trace!(self.logger, "Inconsistency: Tx {} expected in block {}, but is confirmed in {}", txid, expected_block_hash, block_hash);
330                                         return Err(InternalError::Inconsistency);
331                                 }
332                         }
333
334                         let mut matches = Vec::new();
335                         let mut indexes = Vec::new();
336                         let _ = merkle_block.txn.extract_matches(&mut matches, &mut indexes);
337                         if indexes.len() != 1 || matches.len() != 1 || matches[0] != txid {
338                                 log_error!(self.logger, "Retrieved Merkle block for txid {} doesn't match expectations. This should not happen. Please verify server integrity.", txid);
339                                 return Err(InternalError::Failed);
340                         }
341
342                         // unwrap() safety: len() > 0 is checked above
343                         let pos = *indexes.first().unwrap() as usize;
344                         if let Some(tx) = maybe_await!(self.client.get_tx(&txid))? {
345                                 if tx.txid() != txid {
346                                         log_error!(self.logger, "Retrieved transaction for txid {} doesn't match expectations. This should not happen. Please verify server integrity.", txid);
347                                         return Err(InternalError::Failed);
348                                 }
349
350                                 if let Some(block_height) = known_block_height {
351                                         // We can take a shortcut here if a previous call already gave us the height.
352                                         return Ok(Some(ConfirmedTx { tx, txid, block_header, pos, block_height }));
353                                 }
354
355                                 let block_status = maybe_await!(self.client.get_block_status(&block_hash))?;
356                                 if let Some(block_height) = block_status.height {
357                                         return Ok(Some(ConfirmedTx { tx, txid, block_header, pos, block_height }));
358                                 } else {
359                                         // If any previously-confirmed block suddenly is no longer confirmed, we found
360                                         // an inconsistency and should start over.
361                                         log_trace!(self.logger, "Inconsistency: Tx {} was unconfirmed during syncing.", txid);
362                                         return Err(InternalError::Inconsistency);
363                                 }
364                         }
365                 }
366                 Ok(None)
367         }
368
369         #[maybe_async]
370         fn get_unconfirmed_transactions(
371                 &self, confirmables: &Vec<&(dyn Confirm + Sync + Send)>,
372         ) -> Result<Vec<Txid>, InternalError> {
373                 // Query the interface for relevant txids and check whether the relevant blocks are still
374                 // in the best chain, mark them unconfirmed otherwise
375                 let relevant_txids = confirmables
376                         .iter()
377                         .flat_map(|c| c.get_relevant_txids())
378                         .collect::<HashSet<(Txid, u32, Option<BlockHash>)>>();
379
380                 let mut unconfirmed_txs = Vec::new();
381
382                 for (txid, _conf_height, block_hash_opt) in relevant_txids {
383                         if let Some(block_hash) = block_hash_opt {
384                                 let block_status = maybe_await!(self.client.get_block_status(&block_hash))?;
385                                 if block_status.in_best_chain {
386                                         // Skip if the block in question is still confirmed.
387                                         continue;
388                                 }
389
390                                 unconfirmed_txs.push(txid);
391                         } else {
392                                 log_error!(self.logger, "Untracked confirmation of funding transaction. Please ensure none of your channels had been created with LDK prior to version 0.0.113!");
393                                 panic!("Untracked confirmation of funding transaction. Please ensure none of your channels had been created with LDK prior to version 0.0.113!");
394                         }
395                 }
396                 Ok(unconfirmed_txs)
397         }
398
399         /// Returns a reference to the underlying esplora client.
400         pub fn client(&self) -> &EsploraClientType {
401                 &self.client
402         }
403 }
404
405 #[cfg(feature = "async-interface")]
406 type MutexType<I> = futures::lock::Mutex<I>;
407 #[cfg(not(feature = "async-interface"))]
408 type MutexType<I> = std::sync::Mutex<I>;
409
410 // The underlying client type.
411 #[cfg(feature = "async-interface")]
412 type EsploraClientType = AsyncClient;
413 #[cfg(not(feature = "async-interface"))]
414 type EsploraClientType = BlockingClient;
415
416
417 impl<L: Deref> Filter for EsploraSyncClient<L>
418 where
419         L::Target: Logger,
420 {
421         fn register_tx(&self, txid: &Txid, _script_pubkey: &Script) {
422                 let mut locked_queue = self.queue.lock().unwrap();
423                 locked_queue.transactions.insert(*txid);
424         }
425
426         fn register_output(&self, output: WatchedOutput) {
427                 let mut locked_queue = self.queue.lock().unwrap();
428                 locked_queue.outputs.insert(output.outpoint.into_bitcoin_outpoint(), output);
429         }
430 }