Fix race condition in async UtxoFuture resolution

Previously, we refactored the `GossipVerifier` to not require holding a
circular reference. As part of this, we moved to a model where the
`UtxoFuture`s are now polled by the background processor which checks
for completion through `get_and_clear_pending_msg_events`.

However, as part of this refactor we introduced race-condition: as we
only held `Weak` references in `PendingChecksContext` and the
`UtxoFuture` was directly dropped by the `GossipVerifier` after calling
`resolve`, the actual data was dropped with the future and gone when the
background processor attempted to retrieve and apply it via
`check_resolved_futures`.

Here, we fix this issue by simply converting the held channels field in
`PendingChecksContext` to hold an `Arc`, ensuring the data is still
present even after the `UtxoFuture` is dropped.

Signed-off-by: Elias Rohrer <dev@tnull.de>

Author: tnull

lightning/src/routing/utxo.rs

@@ -166,7 +166,7 @@ impl UtxoFuture {
 }
 
 struct PendingChecksContext {
-	channels: HashMap<u64, Weak<Mutex<UtxoMessages>>>,
+	channels: HashMap<u64, Arc<Mutex<UtxoMessages>>>,
 	nodes: HashMap<NodeId, Vec<Weak<Mutex<UtxoMessages>>>>,
 }
 
@@ -194,9 +194,13 @@ impl PendingChecks {
 	) -> Result<(), LightningError> {
 		let mut pending_checks = self.internal.lock().unwrap();
 		if let hash_map::Entry::Occupied(e) = pending_checks.channels.entry(msg.short_channel_id) {
-			let is_from_a = (msg.channel_flags & 1) == 1;
-			match Weak::upgrade(e.get()) {
-				Some(msgs_ref) => {
+			if Arc::strong_count(e.get()) == 1 {
+				// The future was dropped and the entry is the last reference held.
+				e.remove();
+			} else {
+				let is_from_a = (msg.channel_flags & 1) == 1;
+				{
+					let msgs_ref = e.get();
 					let mut messages = msgs_ref.lock().unwrap();
 					let latest_update = if is_from_a {
 						&mut messages.latest_channel_update_a
@@ -216,15 +220,12 @@ impl PendingChecks {
 							ChannelUpdate::Unsigned(msg.clone())
 						});
 					}
-					return Err(LightningError {
-						err: "Awaiting channel_announcement validation to accept channel_update"
-							.to_owned(),
-						action: ErrorAction::IgnoreAndLog(Level::Gossip),
-					});
-				},
-				None => {
-					e.remove();
-				},
+				}
+				return Err(LightningError {
+					err: "Awaiting channel_announcement validation to accept channel_update"
+						.to_owned(),
+					action: ErrorAction::IgnoreAndLog(Level::Gossip),
+				});
 			}
 		}
 		Ok(())
@@ -284,22 +285,31 @@ impl PendingChecks {
 
 	fn check_replace_previous_entry(
 		msg: &msgs::UnsignedChannelAnnouncement, full_msg: Option<&msgs::ChannelAnnouncement>,
-		replacement: Option<Weak<Mutex<UtxoMessages>>>,
-		pending_channels: &mut HashMap<u64, Weak<Mutex<UtxoMessages>>>,
+		replacement: Option<Arc<Mutex<UtxoMessages>>>,
+		pending_channels: &mut HashMap<u64, Arc<Mutex<UtxoMessages>>>,
 	) -> Result<(), msgs::LightningError> {
 		match pending_channels.entry(msg.short_channel_id) {
 			hash_map::Entry::Occupied(mut e) => {
-				// There's already a pending lookup for the given SCID. Check if the messages
-				// are the same and, if so, return immediately (don't bother spawning another
-				// lookup if we haven't gotten that far yet).
-				match Weak::upgrade(&e.get()) {
-					Some(pending_msgs) => {
-						// This may be called with the mutex held on a different UtxoMessages
-						// struct, however in that case we have a global lockorder of new messages
-						// -> old messages, which makes this safe.
-						let pending_matches = match &pending_msgs
-							.unsafe_well_ordered_double_lock_self()
-							.channel_announce
+				if Arc::strong_count(e.get()) == 1 {
+					// The future was dropped and the entry is the last reference held.
+					//
+					// The earlier lookup already resolved. We can't be sure its the same
+					// so just remove/replace it and move on.
+					if let Some(item) = replacement {
+						*e.get_mut() = item;
+					} else {
+						e.remove();
+					}
+				} else {
+					// There's already a pending lookup for the given SCID. Check if the messages
+					// are the same and, if so, return immediately (don't bother spawning another
+					// lookup if we haven't gotten that far yet).
+					let pending_msgs = e.get();
+					// This may be called with the mutex held on a different UtxoMessages
+					// struct, however in that case we have a global lockorder of new messages
+					// -> old messages, which makes this safe.
+					let pending_matches =
+						match &pending_msgs.unsafe_well_ordered_double_lock_self().channel_announce
 						{
 							Some(ChannelAnnouncement::Full(pending_msg)) => {
 								Some(pending_msg) == full_msg
@@ -314,31 +324,21 @@ impl PendingChecks {
 								false
 							},
 						};
-						if pending_matches {
-							return Err(LightningError {
-								err: "Channel announcement is already being checked".to_owned(),
-								action: ErrorAction::IgnoreDuplicateGossip,
-							});
-						} else {
-							// The earlier lookup is a different message. If we have another
-							// request in-flight now replace the original.
-							// Note that in the replace case whether to replace is somewhat
-							// arbitrary - both results will be handled, we're just updating the
-							// value that will be compared to future lookups with the same SCID.
-							if let Some(item) = replacement {
-								*e.get_mut() = item;
-							}
-						}
-					},
-					None => {
-						// The earlier lookup already resolved. We can't be sure its the same
-						// so just remove/replace it and move on.
+					if pending_matches {
+						return Err(LightningError {
+							err: "Channel announcement is already being checked".to_owned(),
+							action: ErrorAction::IgnoreDuplicateGossip,
+						});
+					} else {
+						// The earlier lookup is a different message. If we have another
+						// request in-flight now replace the original.
+						// Note that in the replace case whether to replace is somewhat
+						// arbitrary - both results will be handled, we're just updating the
+						// value that will be compared to future lookups with the same SCID.
 						if let Some(item) = replacement {
 							*e.get_mut() = item;
-						} else {
-							e.remove();
 						}
-					},
+					}
 				}
 			},
 			hash_map::Entry::Vacant(v) => {
@@ -416,7 +416,7 @@ impl PendingChecks {
 							Self::check_replace_previous_entry(
 								msg,
 								full_msg,
-								Some(Arc::downgrade(&future.state)),
+								Some(Arc::clone(&future.state)),
 								&mut pending_checks.channels,
 							)?;
 							async_messages.channel_announce = Some(if let Some(msg) = full_msg {
@@ -464,7 +464,7 @@ impl PendingChecks {
 			// If we have many channel checks pending, ensure we don't have any dangling checks
 			// (i.e. checks where the user told us they'd call back but drop'd the `UtxoFuture`
 			// instead) before we commit to applying backpressure.
-			pending_checks.channels.retain(|_, chan| Weak::upgrade(&chan).is_some());
+			pending_checks.channels.retain(|_, chan| Arc::strong_count(chan) > 1);
 			pending_checks.nodes.retain(|_, channels| {
 				channels.retain(|chan| Weak::upgrade(&chan).is_some());
 				!channels.is_empty()
@@ -575,16 +575,16 @@ impl PendingChecks {
 		{
 			let mut lck = self.internal.lock().unwrap();
 			lck.channels.retain(|_, state| {
-				if let Some(state) = state.upgrade() {
-					if state.lock().unwrap().complete.is_some() {
-						completed_states.push(state);
-						false
-					} else {
+				if state.lock().unwrap().complete.is_some() {
+					completed_states.push(Arc::clone(&state));
+					false
+				} else {
+					if Arc::strong_count(&state) > 1 {
 						true
+					} else {
+						// The UtxoFuture has been dropped, drop the pending-lookup state.
+						false
 					}
-				} else {
-					// The UtxoFuture has been dropped, drop the pending-lookup state.
-					false
 				}
 			});
 			lck.nodes.retain(|_, lookups| {