]> git.bitcoin.ninja Git - rust-lightning/blob - lightning/src/routing/utxo.rs
Persist scorer upon update based on event handling
[rust-lightning] / lightning / src / routing / utxo.rs
1 // This file is Copyright its original authors, visible in version control
2 // history.
3 //
4 // This file is licensed under the Apache License, Version 2.0 <LICENSE-APACHE
5 // or http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
6 // <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your option.
7 // You may not use this file except in accordance with one or both of these
8 // licenses.
9
10 //! This module contains traits for LDK to access UTXOs to check gossip data is correct.
11 //!
12 //! When lightning nodes gossip channel information, they resist DoS attacks by checking that each
13 //! channel matches a UTXO on-chain, requiring at least some marginal on-chain transacting in
14 //! order to announce a channel. This module handles that checking.
15
16 use bitcoin::{BlockHash, TxOut};
17 use bitcoin::hashes::hex::ToHex;
18
19 use crate::events::MessageSendEvent;
20 use crate::ln::chan_utils::make_funding_redeemscript_from_slices;
21 use crate::ln::msgs::{self, LightningError, ErrorAction};
22 use crate::routing::gossip::{NetworkGraph, NodeId, P2PGossipSync};
23 use crate::util::logger::{Level, Logger};
24 use crate::util::ser::Writeable;
25
26 use crate::prelude::*;
27
28 use alloc::sync::{Arc, Weak};
29 use crate::sync::{Mutex, LockTestExt};
30 use core::ops::Deref;
31
32 /// An error when accessing the chain via [`UtxoLookup`].
33 #[derive(Clone, Debug)]
34 pub enum UtxoLookupError {
35         /// The requested chain is unknown.
36         UnknownChain,
37
38         /// The requested transaction doesn't exist or hasn't confirmed.
39         UnknownTx,
40 }
41
42 /// The result of a [`UtxoLookup::get_utxo`] call. A call may resolve either synchronously,
43 /// returning the `Sync` variant, or asynchronously, returning an [`UtxoFuture`] in the `Async`
44 /// variant.
45 #[derive(Clone)]
46 pub enum UtxoResult {
47         /// A result which was resolved synchronously. It either includes a [`TxOut`] for the output
48         /// requested or a [`UtxoLookupError`].
49         Sync(Result<TxOut, UtxoLookupError>),
50         /// A result which will be resolved asynchronously. It includes a [`UtxoFuture`], a `clone` of
51         /// which you must keep locally and call [`UtxoFuture::resolve`] on once the lookup completes.
52         ///
53         /// Note that in order to avoid runaway memory usage, the number of parallel checks is limited,
54         /// but only fairly loosely. Because a pending checks block all message processing, leaving
55         /// checks pending for an extended time may cause DoS of other functions. It is recommended you
56         /// keep a tight timeout on lookups, on the order of a few seconds.
57         Async(UtxoFuture),
58 }
59
60 /// The `UtxoLookup` trait defines behavior for accessing on-chain UTXOs.
61 pub trait UtxoLookup {
62         /// Returns the transaction output of a funding transaction encoded by [`short_channel_id`].
63         /// Returns an error if `genesis_hash` is for a different chain or if such a transaction output
64         /// is unknown.
65         ///
66         /// [`short_channel_id`]: https://github.com/lightning/bolts/blob/master/07-routing-gossip.md#definition-of-short_channel_id
67         fn get_utxo(&self, genesis_hash: &BlockHash, short_channel_id: u64) -> UtxoResult;
68 }
69
70 enum ChannelAnnouncement {
71         Full(msgs::ChannelAnnouncement),
72         Unsigned(msgs::UnsignedChannelAnnouncement),
73 }
74 impl ChannelAnnouncement {
75         fn node_id_1(&self) -> &NodeId {
76                 match self {
77                         ChannelAnnouncement::Full(msg) => &msg.contents.node_id_1,
78                         ChannelAnnouncement::Unsigned(msg) => &msg.node_id_1,
79                 }
80         }
81 }
82
83 enum NodeAnnouncement {
84         Full(msgs::NodeAnnouncement),
85         Unsigned(msgs::UnsignedNodeAnnouncement),
86 }
87 impl NodeAnnouncement {
88         fn timestamp(&self) -> u32 {
89                 match self {
90                         NodeAnnouncement::Full(msg) => msg.contents.timestamp,
91                         NodeAnnouncement::Unsigned(msg) => msg.timestamp,
92                 }
93         }
94 }
95
96 enum ChannelUpdate {
97         Full(msgs::ChannelUpdate),
98         Unsigned(msgs::UnsignedChannelUpdate),
99 }
100 impl ChannelUpdate {
101         fn timestamp(&self) -> u32 {
102                 match self {
103                         ChannelUpdate::Full(msg) => msg.contents.timestamp,
104                         ChannelUpdate::Unsigned(msg) => msg.timestamp,
105                 }
106         }
107 }
108
109 struct UtxoMessages {
110         complete: Option<Result<TxOut, UtxoLookupError>>,
111         channel_announce: Option<ChannelAnnouncement>,
112         latest_node_announce_a: Option<NodeAnnouncement>,
113         latest_node_announce_b: Option<NodeAnnouncement>,
114         latest_channel_update_a: Option<ChannelUpdate>,
115         latest_channel_update_b: Option<ChannelUpdate>,
116 }
117
118 /// Represents a future resolution of a [`UtxoLookup::get_utxo`] query resolving async.
119 ///
120 /// See [`UtxoResult::Async`] and [`UtxoFuture::resolve`] for more info.
121 #[derive(Clone)]
122 pub struct UtxoFuture {
123         state: Arc<Mutex<UtxoMessages>>,
124 }
125
126 /// A trivial implementation of [`UtxoLookup`] which is used to call back into the network graph
127 /// once we have a concrete resolution of a request.
128 struct UtxoResolver(Result<TxOut, UtxoLookupError>);
129 impl UtxoLookup for UtxoResolver {
130         fn get_utxo(&self, _genesis_hash: &BlockHash, _short_channel_id: u64) -> UtxoResult {
131                 UtxoResult::Sync(self.0.clone())
132         }
133 }
134
135 impl UtxoFuture {
136         /// Builds a new future for later resolution.
137         pub fn new() -> Self {
138                 Self { state: Arc::new(Mutex::new(UtxoMessages {
139                         complete: None,
140                         channel_announce: None,
141                         latest_node_announce_a: None,
142                         latest_node_announce_b: None,
143                         latest_channel_update_a: None,
144                         latest_channel_update_b: None,
145                 }))}
146         }
147
148         /// Resolves this future against the given `graph` and with the given `result`.
149         ///
150         /// This is identical to calling [`UtxoFuture::resolve`] with a dummy `gossip`, disabling
151         /// forwarding the validated gossip message onwards to peers.
152         ///
153         /// Because this may cause the [`NetworkGraph`]'s [`processing_queue_high`] to flip, in order
154         /// to allow us to interact with peers again, you should call [`PeerManager::process_events`]
155         /// after this.
156         ///
157         /// [`processing_queue_high`]: crate::ln::msgs::RoutingMessageHandler::processing_queue_high
158         /// [`PeerManager::process_events`]: crate::ln::peer_handler::PeerManager::process_events
159         pub fn resolve_without_forwarding<L: Deref>(&self,
160                 graph: &NetworkGraph<L>, result: Result<TxOut, UtxoLookupError>)
161         where L::Target: Logger {
162                 self.do_resolve(graph, result);
163         }
164
165         /// Resolves this future against the given `graph` and with the given `result`.
166         ///
167         /// The given `gossip` is used to broadcast any validated messages onwards to all peers which
168         /// have available buffer space.
169         ///
170         /// Because this may cause the [`NetworkGraph`]'s [`processing_queue_high`] to flip, in order
171         /// to allow us to interact with peers again, you should call [`PeerManager::process_events`]
172         /// after this.
173         ///
174         /// [`processing_queue_high`]: crate::ln::msgs::RoutingMessageHandler::processing_queue_high
175         /// [`PeerManager::process_events`]: crate::ln::peer_handler::PeerManager::process_events
176         pub fn resolve<L: Deref, G: Deref<Target=NetworkGraph<L>>, U: Deref, GS: Deref<Target = P2PGossipSync<G, U, L>>>(&self,
177                 graph: &NetworkGraph<L>, gossip: GS, result: Result<TxOut, UtxoLookupError>
178         ) where L::Target: Logger, U::Target: UtxoLookup {
179                 let mut res = self.do_resolve(graph, result);
180                 for msg_opt in res.iter_mut() {
181                         if let Some(msg) = msg_opt.take() {
182                                 gossip.forward_gossip_msg(msg);
183                         }
184                 }
185         }
186
187         fn do_resolve<L: Deref>(&self, graph: &NetworkGraph<L>, result: Result<TxOut, UtxoLookupError>)
188         -> [Option<MessageSendEvent>; 5] where L::Target: Logger {
189                 let (announcement, node_a, node_b, update_a, update_b) = {
190                         let mut pending_checks = graph.pending_checks.internal.lock().unwrap();
191                         let mut async_messages = self.state.lock().unwrap();
192
193                         if async_messages.channel_announce.is_none() {
194                                 // We raced returning to `check_channel_announcement` which hasn't updated
195                                 // `channel_announce` yet. That's okay, we can set the `complete` field which it will
196                                 // check once it gets control again.
197                                 async_messages.complete = Some(result);
198                                 return [None, None, None, None, None];
199                         }
200
201                         let announcement_msg = match async_messages.channel_announce.as_ref().unwrap() {
202                                 ChannelAnnouncement::Full(signed_msg) => &signed_msg.contents,
203                                 ChannelAnnouncement::Unsigned(msg) => &msg,
204                         };
205
206                         pending_checks.lookup_completed(announcement_msg, &Arc::downgrade(&self.state));
207
208                         (async_messages.channel_announce.take().unwrap(),
209                                 async_messages.latest_node_announce_a.take(),
210                                 async_messages.latest_node_announce_b.take(),
211                                 async_messages.latest_channel_update_a.take(),
212                                 async_messages.latest_channel_update_b.take())
213                 };
214
215                 let mut res = [None, None, None, None, None];
216                 let mut res_idx = 0;
217
218                 // Now that we've updated our internal state, pass the pending messages back through the
219                 // network graph with a different `UtxoLookup` which will resolve immediately.
220                 // Note that we ignore errors as we don't disconnect peers anyway, so there's nothing to do
221                 // with them.
222                 let resolver = UtxoResolver(result);
223                 match announcement {
224                         ChannelAnnouncement::Full(signed_msg) => {
225                                 if graph.update_channel_from_announcement(&signed_msg, &Some(&resolver)).is_ok() {
226                                         res[res_idx] = Some(MessageSendEvent::BroadcastChannelAnnouncement {
227                                                 msg: signed_msg, update_msg: None,
228                                         });
229                                         res_idx += 1;
230                                 }
231                         },
232                         ChannelAnnouncement::Unsigned(msg) => {
233                                 let _ = graph.update_channel_from_unsigned_announcement(&msg, &Some(&resolver));
234                         },
235                 }
236
237                 for announce in core::iter::once(node_a).chain(core::iter::once(node_b)) {
238                         match announce {
239                                 Some(NodeAnnouncement::Full(signed_msg)) => {
240                                         if graph.update_node_from_announcement(&signed_msg).is_ok() {
241                                                 res[res_idx] = Some(MessageSendEvent::BroadcastNodeAnnouncement {
242                                                         msg: signed_msg,
243                                                 });
244                                                 res_idx += 1;
245                                         }
246                                 },
247                                 Some(NodeAnnouncement::Unsigned(msg)) => {
248                                         let _ = graph.update_node_from_unsigned_announcement(&msg);
249                                 },
250                                 None => {},
251                         }
252                 }
253
254                 for update in core::iter::once(update_a).chain(core::iter::once(update_b)) {
255                         match update {
256                                 Some(ChannelUpdate::Full(signed_msg)) => {
257                                         if graph.update_channel(&signed_msg).is_ok() {
258                                                 res[res_idx] = Some(MessageSendEvent::BroadcastChannelUpdate {
259                                                         msg: signed_msg,
260                                                 });
261                                                 res_idx += 1;
262                                         }
263                                 },
264                                 Some(ChannelUpdate::Unsigned(msg)) => {
265                                         let _ = graph.update_channel_unsigned(&msg);
266                                 },
267                                 None => {},
268                         }
269                 }
270
271                 res
272         }
273 }
274
275 struct PendingChecksContext {
276         channels: HashMap<u64, Weak<Mutex<UtxoMessages>>>,
277         nodes: HashMap<NodeId, Vec<Weak<Mutex<UtxoMessages>>>>,
278 }
279
280 impl PendingChecksContext {
281         fn lookup_completed(&mut self,
282                 msg: &msgs::UnsignedChannelAnnouncement, completed_state: &Weak<Mutex<UtxoMessages>>
283         ) {
284                 if let hash_map::Entry::Occupied(e) = self.channels.entry(msg.short_channel_id) {
285                         if Weak::ptr_eq(e.get(), &completed_state) {
286                                 e.remove();
287                         }
288                 }
289
290                 if let hash_map::Entry::Occupied(mut e) = self.nodes.entry(msg.node_id_1) {
291                         e.get_mut().retain(|elem| !Weak::ptr_eq(&elem, &completed_state));
292                         if e.get().is_empty() { e.remove(); }
293                 }
294                 if let hash_map::Entry::Occupied(mut e) = self.nodes.entry(msg.node_id_2) {
295                         e.get_mut().retain(|elem| !Weak::ptr_eq(&elem, &completed_state));
296                         if e.get().is_empty() { e.remove(); }
297                 }
298         }
299 }
300
301 /// A set of messages which are pending UTXO lookups for processing.
302 pub(super) struct PendingChecks {
303         internal: Mutex<PendingChecksContext>,
304 }
305
306 impl PendingChecks {
307         pub(super) fn new() -> Self {
308                 PendingChecks { internal: Mutex::new(PendingChecksContext {
309                         channels: HashMap::new(), nodes: HashMap::new(),
310                 }) }
311         }
312
313         /// Checks if there is a pending `channel_update` UTXO validation for the given channel,
314         /// and, if so, stores the channel message for handling later and returns an `Err`.
315         pub(super) fn check_hold_pending_channel_update(
316                 &self, msg: &msgs::UnsignedChannelUpdate, full_msg: Option<&msgs::ChannelUpdate>
317         ) -> Result<(), LightningError> {
318                 let mut pending_checks = self.internal.lock().unwrap();
319                 if let hash_map::Entry::Occupied(e) = pending_checks.channels.entry(msg.short_channel_id) {
320                         let is_from_a = (msg.flags & 1) == 1;
321                         match Weak::upgrade(e.get()) {
322                                 Some(msgs_ref) => {
323                                         let mut messages = msgs_ref.lock().unwrap();
324                                         let latest_update = if is_from_a {
325                                                         &mut messages.latest_channel_update_a
326                                                 } else {
327                                                         &mut messages.latest_channel_update_b
328                                                 };
329                                         if latest_update.is_none() || latest_update.as_ref().unwrap().timestamp() < msg.timestamp {
330                                                 // If the messages we got has a higher timestamp, just blindly assume the
331                                                 // signatures on the new message are correct and drop the old message. This
332                                                 // may cause us to end up dropping valid `channel_update`s if a peer is
333                                                 // malicious, but we should get the correct ones when the node updates them.
334                                                 *latest_update = Some(
335                                                         if let Some(msg) = full_msg { ChannelUpdate::Full(msg.clone()) }
336                                                         else { ChannelUpdate::Unsigned(msg.clone()) });
337                                         }
338                                         return Err(LightningError {
339                                                 err: "Awaiting channel_announcement validation to accept channel_update".to_owned(),
340                                                 action: ErrorAction::IgnoreAndLog(Level::Gossip),
341                                         });
342                                 },
343                                 None => { e.remove(); },
344                         }
345                 }
346                 Ok(())
347         }
348
349         /// Checks if there is a pending `node_announcement` UTXO validation for a channel with the
350         /// given node and, if so, stores the channel message for handling later and returns an `Err`.
351         pub(super) fn check_hold_pending_node_announcement(
352                 &self, msg: &msgs::UnsignedNodeAnnouncement, full_msg: Option<&msgs::NodeAnnouncement>
353         ) -> Result<(), LightningError> {
354                 let mut pending_checks = self.internal.lock().unwrap();
355                 if let hash_map::Entry::Occupied(mut e) = pending_checks.nodes.entry(msg.node_id) {
356                         let mut found_at_least_one_chan = false;
357                         e.get_mut().retain(|node_msgs| {
358                                 match Weak::upgrade(&node_msgs) {
359                                         Some(chan_mtx) => {
360                                                 let mut chan_msgs = chan_mtx.lock().unwrap();
361                                                 if let Some(chan_announce) = &chan_msgs.channel_announce {
362                                                         let latest_announce =
363                                                                 if *chan_announce.node_id_1() == msg.node_id {
364                                                                         &mut chan_msgs.latest_node_announce_a
365                                                                 } else {
366                                                                         &mut chan_msgs.latest_node_announce_b
367                                                                 };
368                                                         if latest_announce.is_none() ||
369                                                                 latest_announce.as_ref().unwrap().timestamp() < msg.timestamp
370                                                         {
371                                                                 *latest_announce = Some(
372                                                                         if let Some(msg) = full_msg { NodeAnnouncement::Full(msg.clone()) }
373                                                                         else { NodeAnnouncement::Unsigned(msg.clone()) });
374                                                         }
375                                                         found_at_least_one_chan = true;
376                                                         true
377                                                 } else {
378                                                         debug_assert!(false, "channel_announce is set before struct is added to node map");
379                                                         false
380                                                 }
381                                         },
382                                         None => false,
383                                 }
384                         });
385                         if e.get().is_empty() { e.remove(); }
386                         if found_at_least_one_chan {
387                                 return Err(LightningError {
388                                         err: "Awaiting channel_announcement validation to accept node_announcement".to_owned(),
389                                         action: ErrorAction::IgnoreAndLog(Level::Gossip),
390                                 });
391                         }
392                 }
393                 Ok(())
394         }
395
396         fn check_replace_previous_entry(msg: &msgs::UnsignedChannelAnnouncement,
397                 full_msg: Option<&msgs::ChannelAnnouncement>, replacement: Option<Weak<Mutex<UtxoMessages>>>,
398                 pending_channels: &mut HashMap<u64, Weak<Mutex<UtxoMessages>>>
399         ) -> Result<(), msgs::LightningError> {
400                 match pending_channels.entry(msg.short_channel_id) {
401                         hash_map::Entry::Occupied(mut e) => {
402                                 // There's already a pending lookup for the given SCID. Check if the messages
403                                 // are the same and, if so, return immediately (don't bother spawning another
404                                 // lookup if we haven't gotten that far yet).
405                                 match Weak::upgrade(&e.get()) {
406                                         Some(pending_msgs) => {
407                                                 // This may be called with the mutex held on a different UtxoMessages
408                                                 // struct, however in that case we have a global lockorder of new messages
409                                                 // -> old messages, which makes this safe.
410                                                 let pending_matches = match &pending_msgs.unsafe_well_ordered_double_lock_self().channel_announce {
411                                                         Some(ChannelAnnouncement::Full(pending_msg)) => Some(pending_msg) == full_msg,
412                                                         Some(ChannelAnnouncement::Unsigned(pending_msg)) => pending_msg == msg,
413                                                         None => {
414                                                                 // This shouldn't actually be reachable. We set the
415                                                                 // `channel_announce` field under the same lock as setting the
416                                                                 // channel map entry. Still, we can just treat it as
417                                                                 // non-matching and let the new request fly.
418                                                                 debug_assert!(false);
419                                                                 false
420                                                         },
421                                                 };
422                                                 if pending_matches {
423                                                         return Err(LightningError {
424                                                                 err: "Channel announcement is already being checked".to_owned(),
425                                                                 action: ErrorAction::IgnoreDuplicateGossip,
426                                                         });
427                                                 } else {
428                                                         // The earlier lookup is a different message. If we have another
429                                                         // request in-flight now replace the original.
430                                                         // Note that in the replace case whether to replace is somewhat
431                                                         // arbitrary - both results will be handled, we're just updating the
432                                                         // value that will be compared to future lookups with the same SCID.
433                                                         if let Some(item) = replacement {
434                                                                 *e.get_mut() = item;
435                                                         }
436                                                 }
437                                         },
438                                         None => {
439                                                 // The earlier lookup already resolved. We can't be sure its the same
440                                                 // so just remove/replace it and move on.
441                                                 if let Some(item) = replacement {
442                                                         *e.get_mut() = item;
443                                                 } else { e.remove(); }
444                                         },
445                                 }
446                         },
447                         hash_map::Entry::Vacant(v) => {
448                                 if let Some(item) = replacement { v.insert(item); }
449                         },
450                 }
451                 Ok(())
452         }
453
454         pub(super) fn check_channel_announcement<U: Deref>(&self,
455                 utxo_lookup: &Option<U>, msg: &msgs::UnsignedChannelAnnouncement,
456                 full_msg: Option<&msgs::ChannelAnnouncement>
457         ) -> Result<Option<u64>, msgs::LightningError> where U::Target: UtxoLookup {
458                 let handle_result = |res| {
459                         match res {
460                                 Ok(TxOut { value, script_pubkey }) => {
461                                         let expected_script =
462                                                 make_funding_redeemscript_from_slices(msg.bitcoin_key_1.as_slice(), msg.bitcoin_key_2.as_slice()).to_v0_p2wsh();
463                                         if script_pubkey != expected_script {
464                                                 return Err(LightningError{
465                                                         err: format!("Channel announcement key ({}) didn't match on-chain script ({})",
466                                                                 expected_script.to_hex(), script_pubkey.to_hex()),
467                                                         action: ErrorAction::IgnoreError
468                                                 });
469                                         }
470                                         Ok(Some(value))
471                                 },
472                                 Err(UtxoLookupError::UnknownChain) => {
473                                         Err(LightningError {
474                                                 err: format!("Channel announced on an unknown chain ({})",
475                                                         msg.chain_hash.encode().to_hex()),
476                                                 action: ErrorAction::IgnoreError
477                                         })
478                                 },
479                                 Err(UtxoLookupError::UnknownTx) => {
480                                         Err(LightningError {
481                                                 err: "Channel announced without corresponding UTXO entry".to_owned(),
482                                                 action: ErrorAction::IgnoreError
483                                         })
484                                 },
485                         }
486                 };
487
488                 Self::check_replace_previous_entry(msg, full_msg, None,
489                         &mut self.internal.lock().unwrap().channels)?;
490
491                 match utxo_lookup {
492                         &None => {
493                                 // Tentatively accept, potentially exposing us to DoS attacks
494                                 Ok(None)
495                         },
496                         &Some(ref utxo_lookup) => {
497                                 match utxo_lookup.get_utxo(&msg.chain_hash, msg.short_channel_id) {
498                                         UtxoResult::Sync(res) => handle_result(res),
499                                         UtxoResult::Async(future) => {
500                                                 let mut pending_checks = self.internal.lock().unwrap();
501                                                 let mut async_messages = future.state.lock().unwrap();
502                                                 if let Some(res) = async_messages.complete.take() {
503                                                         // In the unlikely event the future resolved before we managed to get it,
504                                                         // handle the result in-line.
505                                                         handle_result(res)
506                                                 } else {
507                                                         Self::check_replace_previous_entry(msg, full_msg,
508                                                                 Some(Arc::downgrade(&future.state)), &mut pending_checks.channels)?;
509                                                         async_messages.channel_announce = Some(
510                                                                 if let Some(msg) = full_msg { ChannelAnnouncement::Full(msg.clone()) }
511                                                                 else { ChannelAnnouncement::Unsigned(msg.clone()) });
512                                                         pending_checks.nodes.entry(msg.node_id_1)
513                                                                 .or_insert(Vec::new()).push(Arc::downgrade(&future.state));
514                                                         pending_checks.nodes.entry(msg.node_id_2)
515                                                                 .or_insert(Vec::new()).push(Arc::downgrade(&future.state));
516                                                         Err(LightningError {
517                                                                 err: "Channel being checked async".to_owned(),
518                                                                 action: ErrorAction::IgnoreAndLog(Level::Gossip),
519                                                         })
520                                                 }
521                                         },
522                                 }
523                         }
524                 }
525         }
526
527         /// The maximum number of pending gossip checks before [`Self::too_many_checks_pending`]
528         /// returns `true`. Note that this isn't a strict upper-bound on the number of checks pending -
529         /// each peer may, at a minimum, read one more socket buffer worth of `channel_announcement`s
530         /// which we'll have to process. With a socket buffer of 4KB and a minimum
531         /// `channel_announcement` size of, roughly, 429 bytes, this may leave us with `10*our peer
532         /// count` messages to process beyond this limit. Because we'll probably have a few peers,
533         /// there's no reason for this constant to be materially less than 30 or so, and 32 in-flight
534         /// checks should be more than enough for decent parallelism.
535         const MAX_PENDING_LOOKUPS: usize = 32;
536
537         /// Returns true if there are a large number of async checks pending and future
538         /// `channel_announcement` messages should be delayed. Note that this is only a hint and
539         /// messages already in-flight may still have to be handled for various reasons.
540         pub(super) fn too_many_checks_pending(&self) -> bool {
541                 let mut pending_checks = self.internal.lock().unwrap();
542                 if pending_checks.channels.len() > Self::MAX_PENDING_LOOKUPS {
543                         // If we have many channel checks pending, ensure we don't have any dangling checks
544                         // (i.e. checks where the user told us they'd call back but drop'd the `UtxoFuture`
545                         // instead) before we commit to applying backpressure.
546                         pending_checks.channels.retain(|_, chan| {
547                                 Weak::upgrade(&chan).is_some()
548                         });
549                         pending_checks.nodes.retain(|_, channels| {
550                                 channels.retain(|chan| Weak::upgrade(&chan).is_some());
551                                 !channels.is_empty()
552                         });
553                         pending_checks.channels.len() > Self::MAX_PENDING_LOOKUPS
554                 } else {
555                         false
556                 }
557         }
558 }
559
560 #[cfg(test)]
561 mod tests {
562         use super::*;
563         use crate::routing::gossip::tests::*;
564         use crate::util::test_utils::{TestChainSource, TestLogger};
565         use crate::ln::msgs;
566
567         use bitcoin::secp256k1::{Secp256k1, SecretKey};
568
569         use core::sync::atomic::Ordering;
570
571         fn get_network() -> (TestChainSource, NetworkGraph<Box<TestLogger>>) {
572                 let logger = Box::new(TestLogger::new());
573                 let chain_source = TestChainSource::new(bitcoin::Network::Testnet);
574                 let network_graph = NetworkGraph::new(bitcoin::Network::Testnet, logger);
575
576                 (chain_source, network_graph)
577         }
578
579         fn get_test_objects() -> (msgs::ChannelAnnouncement, TestChainSource,
580                 NetworkGraph<Box<TestLogger>>, bitcoin::Script, msgs::NodeAnnouncement,
581                 msgs::NodeAnnouncement, msgs::ChannelUpdate, msgs::ChannelUpdate, msgs::ChannelUpdate)
582         {
583                 let secp_ctx = Secp256k1::new();
584
585                 let (chain_source, network_graph) = get_network();
586
587                 let good_script = get_channel_script(&secp_ctx);
588                 let node_1_privkey = &SecretKey::from_slice(&[42; 32]).unwrap();
589                 let node_2_privkey = &SecretKey::from_slice(&[41; 32]).unwrap();
590                 let valid_announcement = get_signed_channel_announcement(|_| {}, node_1_privkey, node_2_privkey, &secp_ctx);
591
592                 let node_a_announce = get_signed_node_announcement(|_| {}, node_1_privkey, &secp_ctx);
593                 let node_b_announce = get_signed_node_announcement(|_| {}, node_2_privkey, &secp_ctx);
594
595                 // Note that we have to set the "direction" flag correctly on both messages
596                 let chan_update_a = get_signed_channel_update(|msg| msg.flags = 0, node_1_privkey, &secp_ctx);
597                 let chan_update_b = get_signed_channel_update(|msg| msg.flags = 1, node_2_privkey, &secp_ctx);
598                 let chan_update_c = get_signed_channel_update(|msg| {
599                         msg.flags = 1; msg.timestamp += 1; }, node_2_privkey, &secp_ctx);
600
601                 (valid_announcement, chain_source, network_graph, good_script, node_a_announce,
602                         node_b_announce, chan_update_a, chan_update_b, chan_update_c)
603         }
604
605         #[test]
606         fn test_fast_async_lookup() {
607                 // Check that async lookups which resolve quicker than the future is returned to the
608                 // `get_utxo` call can read it still resolve properly.
609                 let (valid_announcement, chain_source, network_graph, good_script, ..) = get_test_objects();
610
611                 let future = UtxoFuture::new();
612                 future.resolve_without_forwarding(&network_graph,
613                         Ok(TxOut { value: 1_000_000, script_pubkey: good_script }));
614                 *chain_source.utxo_ret.lock().unwrap() = UtxoResult::Async(future.clone());
615
616                 network_graph.update_channel_from_announcement(&valid_announcement, &Some(&chain_source)).unwrap();
617                 assert!(network_graph.read_only().channels().get(&valid_announcement.contents.short_channel_id).is_some());
618         }
619
620         #[test]
621         fn test_async_lookup() {
622                 // Test a simple async lookup
623                 let (valid_announcement, chain_source, network_graph, good_script,
624                         node_a_announce, node_b_announce, ..) = get_test_objects();
625
626                 let future = UtxoFuture::new();
627                 *chain_source.utxo_ret.lock().unwrap() = UtxoResult::Async(future.clone());
628
629                 assert_eq!(
630                         network_graph.update_channel_from_announcement(&valid_announcement, &Some(&chain_source)).unwrap_err().err,
631                         "Channel being checked async");
632                 assert!(network_graph.read_only().channels().get(&valid_announcement.contents.short_channel_id).is_none());
633
634                 future.resolve_without_forwarding(&network_graph,
635                         Ok(TxOut { value: 0, script_pubkey: good_script }));
636                 network_graph.read_only().channels().get(&valid_announcement.contents.short_channel_id).unwrap();
637                 network_graph.read_only().channels().get(&valid_announcement.contents.short_channel_id).unwrap();
638
639                 assert!(network_graph.read_only().nodes().get(&valid_announcement.contents.node_id_1)
640                         .unwrap().announcement_info.is_none());
641
642                 network_graph.update_node_from_announcement(&node_a_announce).unwrap();
643                 network_graph.update_node_from_announcement(&node_b_announce).unwrap();
644
645                 assert!(network_graph.read_only().nodes().get(&valid_announcement.contents.node_id_1)
646                         .unwrap().announcement_info.is_some());
647         }
648
649         #[test]
650         fn test_invalid_async_lookup() {
651                 // Test an async lookup which returns an incorrect script
652                 let (valid_announcement, chain_source, network_graph, ..) = get_test_objects();
653
654                 let future = UtxoFuture::new();
655                 *chain_source.utxo_ret.lock().unwrap() = UtxoResult::Async(future.clone());
656
657                 assert_eq!(
658                         network_graph.update_channel_from_announcement(&valid_announcement, &Some(&chain_source)).unwrap_err().err,
659                         "Channel being checked async");
660                 assert!(network_graph.read_only().channels().get(&valid_announcement.contents.short_channel_id).is_none());
661
662                 future.resolve_without_forwarding(&network_graph,
663                         Ok(TxOut { value: 1_000_000, script_pubkey: bitcoin::Script::new() }));
664                 assert!(network_graph.read_only().channels().get(&valid_announcement.contents.short_channel_id).is_none());
665         }
666
667         #[test]
668         fn test_failing_async_lookup() {
669                 // Test an async lookup which returns an error
670                 let (valid_announcement, chain_source, network_graph, ..) = get_test_objects();
671
672                 let future = UtxoFuture::new();
673                 *chain_source.utxo_ret.lock().unwrap() = UtxoResult::Async(future.clone());
674
675                 assert_eq!(
676                         network_graph.update_channel_from_announcement(&valid_announcement, &Some(&chain_source)).unwrap_err().err,
677                         "Channel being checked async");
678                 assert!(network_graph.read_only().channels().get(&valid_announcement.contents.short_channel_id).is_none());
679
680                 future.resolve_without_forwarding(&network_graph, Err(UtxoLookupError::UnknownTx));
681                 assert!(network_graph.read_only().channels().get(&valid_announcement.contents.short_channel_id).is_none());
682         }
683
684         #[test]
685         fn test_updates_async_lookup() {
686                 // Test async lookups will process pending channel_update/node_announcements once they
687                 // complete.
688                 let (valid_announcement, chain_source, network_graph, good_script, node_a_announce,
689                         node_b_announce, chan_update_a, chan_update_b, ..) = get_test_objects();
690
691                 let future = UtxoFuture::new();
692                 *chain_source.utxo_ret.lock().unwrap() = UtxoResult::Async(future.clone());
693
694                 assert_eq!(
695                         network_graph.update_channel_from_announcement(&valid_announcement, &Some(&chain_source)).unwrap_err().err,
696                         "Channel being checked async");
697                 assert!(network_graph.read_only().channels().get(&valid_announcement.contents.short_channel_id).is_none());
698
699                 assert_eq!(
700                         network_graph.update_node_from_announcement(&node_a_announce).unwrap_err().err,
701                         "Awaiting channel_announcement validation to accept node_announcement");
702                 assert_eq!(
703                         network_graph.update_node_from_announcement(&node_b_announce).unwrap_err().err,
704                         "Awaiting channel_announcement validation to accept node_announcement");
705
706                 assert_eq!(network_graph.update_channel(&chan_update_a).unwrap_err().err,
707                         "Awaiting channel_announcement validation to accept channel_update");
708                 assert_eq!(network_graph.update_channel(&chan_update_b).unwrap_err().err,
709                         "Awaiting channel_announcement validation to accept channel_update");
710
711                 future.resolve_without_forwarding(&network_graph,
712                         Ok(TxOut { value: 1_000_000, script_pubkey: good_script }));
713
714                 assert!(network_graph.read_only().channels()
715                         .get(&valid_announcement.contents.short_channel_id).unwrap().one_to_two.is_some());
716                 assert!(network_graph.read_only().channels()
717                         .get(&valid_announcement.contents.short_channel_id).unwrap().two_to_one.is_some());
718
719                 assert!(network_graph.read_only().nodes().get(&valid_announcement.contents.node_id_1)
720                         .unwrap().announcement_info.is_some());
721                 assert!(network_graph.read_only().nodes().get(&valid_announcement.contents.node_id_2)
722                         .unwrap().announcement_info.is_some());
723         }
724
725         #[test]
726         fn test_latest_update_async_lookup() {
727                 // Test async lookups will process the latest channel_update if two are received while
728                 // awaiting an async UTXO lookup.
729                 let (valid_announcement, chain_source, network_graph, good_script, _,
730                         _, chan_update_a, chan_update_b, chan_update_c, ..) = get_test_objects();
731
732                 let future = UtxoFuture::new();
733                 *chain_source.utxo_ret.lock().unwrap() = UtxoResult::Async(future.clone());
734
735                 assert_eq!(
736                         network_graph.update_channel_from_announcement(&valid_announcement, &Some(&chain_source)).unwrap_err().err,
737                         "Channel being checked async");
738                 assert!(network_graph.read_only().channels().get(&valid_announcement.contents.short_channel_id).is_none());
739
740                 assert_eq!(network_graph.update_channel(&chan_update_a).unwrap_err().err,
741                         "Awaiting channel_announcement validation to accept channel_update");
742                 assert_eq!(network_graph.update_channel(&chan_update_b).unwrap_err().err,
743                         "Awaiting channel_announcement validation to accept channel_update");
744                 assert_eq!(network_graph.update_channel(&chan_update_c).unwrap_err().err,
745                         "Awaiting channel_announcement validation to accept channel_update");
746
747                 future.resolve_without_forwarding(&network_graph,
748                         Ok(TxOut { value: 1_000_000, script_pubkey: good_script }));
749
750                 assert_eq!(chan_update_a.contents.timestamp, chan_update_b.contents.timestamp);
751                 let graph_lock = network_graph.read_only();
752                 assert!(graph_lock.channels()
753                                 .get(&valid_announcement.contents.short_channel_id).as_ref().unwrap()
754                                 .one_to_two.as_ref().unwrap().last_update !=
755                         graph_lock.channels()
756                                 .get(&valid_announcement.contents.short_channel_id).as_ref().unwrap()
757                                 .two_to_one.as_ref().unwrap().last_update);
758         }
759
760         #[test]
761         fn test_no_double_lookups() {
762                 // Test that a pending async lookup will prevent a second async lookup from flying, but
763                 // only if the channel_announcement message is identical.
764                 let (valid_announcement, chain_source, network_graph, good_script, ..) = get_test_objects();
765
766                 let future = UtxoFuture::new();
767                 *chain_source.utxo_ret.lock().unwrap() = UtxoResult::Async(future.clone());
768
769                 assert_eq!(
770                         network_graph.update_channel_from_announcement(&valid_announcement, &Some(&chain_source)).unwrap_err().err,
771                         "Channel being checked async");
772                 assert_eq!(chain_source.get_utxo_call_count.load(Ordering::Relaxed), 1);
773
774                 // If we make a second request with the same message, the call count doesn't increase...
775                 let future_b = UtxoFuture::new();
776                 *chain_source.utxo_ret.lock().unwrap() = UtxoResult::Async(future_b.clone());
777                 assert_eq!(
778                         network_graph.update_channel_from_announcement(&valid_announcement, &Some(&chain_source)).unwrap_err().err,
779                         "Channel announcement is already being checked");
780                 assert_eq!(chain_source.get_utxo_call_count.load(Ordering::Relaxed), 1);
781
782                 // But if we make a third request with a tweaked message, we should get a second call
783                 // against our new future...
784                 let secp_ctx = Secp256k1::new();
785                 let replacement_pk_1 = &SecretKey::from_slice(&[99; 32]).unwrap();
786                 let replacement_pk_2 = &SecretKey::from_slice(&[98; 32]).unwrap();
787                 let invalid_announcement = get_signed_channel_announcement(|_| {}, replacement_pk_1, replacement_pk_2, &secp_ctx);
788                 assert_eq!(
789                         network_graph.update_channel_from_announcement(&invalid_announcement, &Some(&chain_source)).unwrap_err().err,
790                         "Channel being checked async");
791                 assert_eq!(chain_source.get_utxo_call_count.load(Ordering::Relaxed), 2);
792
793                 // Still, if we resolve the original future, the original channel will be accepted.
794                 future.resolve_without_forwarding(&network_graph,
795                         Ok(TxOut { value: 1_000_000, script_pubkey: good_script }));
796                 assert!(!network_graph.read_only().channels()
797                         .get(&valid_announcement.contents.short_channel_id).unwrap()
798                         .announcement_message.as_ref().unwrap()
799                         .contents.features.supports_unknown_test_feature());
800         }
801
802         #[test]
803         fn test_checks_backpressure() {
804                 // Test that too_many_checks_pending returns true when there are many checks pending, and
805                 // returns false once they complete.
806                 let secp_ctx = Secp256k1::new();
807                 let (chain_source, network_graph) = get_network();
808
809                 // We cheat and use a single future for all the lookups to complete them all at once.
810                 let future = UtxoFuture::new();
811                 *chain_source.utxo_ret.lock().unwrap() = UtxoResult::Async(future.clone());
812
813                 let node_1_privkey = &SecretKey::from_slice(&[42; 32]).unwrap();
814                 let node_2_privkey = &SecretKey::from_slice(&[41; 32]).unwrap();
815
816                 for i in 0..PendingChecks::MAX_PENDING_LOOKUPS {
817                         let valid_announcement = get_signed_channel_announcement(
818                                 |msg| msg.short_channel_id += 1 + i as u64, node_1_privkey, node_2_privkey, &secp_ctx);
819                         network_graph.update_channel_from_announcement(&valid_announcement, &Some(&chain_source)).unwrap_err();
820                         assert!(!network_graph.pending_checks.too_many_checks_pending());
821                 }
822
823                 let valid_announcement = get_signed_channel_announcement(
824                         |_| {}, node_1_privkey, node_2_privkey, &secp_ctx);
825                 network_graph.update_channel_from_announcement(&valid_announcement, &Some(&chain_source)).unwrap_err();
826                 assert!(network_graph.pending_checks.too_many_checks_pending());
827
828                 // Once the future completes the "too many checks" flag should reset.
829                 future.resolve_without_forwarding(&network_graph, Err(UtxoLookupError::UnknownTx));
830                 assert!(!network_graph.pending_checks.too_many_checks_pending());
831         }
832
833         #[test]
834         fn test_checks_backpressure_drop() {
835                 // Test that too_many_checks_pending returns true when there are many checks pending, and
836                 // returns false if we drop some of the futures without completion.
837                 let secp_ctx = Secp256k1::new();
838                 let (chain_source, network_graph) = get_network();
839
840                 // We cheat and use a single future for all the lookups to complete them all at once.
841                 *chain_source.utxo_ret.lock().unwrap() = UtxoResult::Async(UtxoFuture::new());
842
843                 let node_1_privkey = &SecretKey::from_slice(&[42; 32]).unwrap();
844                 let node_2_privkey = &SecretKey::from_slice(&[41; 32]).unwrap();
845
846                 for i in 0..PendingChecks::MAX_PENDING_LOOKUPS {
847                         let valid_announcement = get_signed_channel_announcement(
848                                 |msg| msg.short_channel_id += 1 + i as u64, node_1_privkey, node_2_privkey, &secp_ctx);
849                         network_graph.update_channel_from_announcement(&valid_announcement, &Some(&chain_source)).unwrap_err();
850                         assert!(!network_graph.pending_checks.too_many_checks_pending());
851                 }
852
853                 let valid_announcement = get_signed_channel_announcement(
854                         |_| {}, node_1_privkey, node_2_privkey, &secp_ctx);
855                 network_graph.update_channel_from_announcement(&valid_announcement, &Some(&chain_source)).unwrap_err();
856                 assert!(network_graph.pending_checks.too_many_checks_pending());
857
858                 // Once the future is drop'd (by resetting the `utxo_ret` value) the "too many checks" flag
859                 // should reset to false.
860                 *chain_source.utxo_ret.lock().unwrap() = UtxoResult::Sync(Err(UtxoLookupError::UnknownTx));
861                 assert!(!network_graph.pending_checks.too_many_checks_pending());
862         }
863 }