Forward gossip messages which were verified asynchronously

Gossip messages which were verified against the chain
asynchronously should still be forwarded to peers, but must now go
out via a new `P2PGossipSync` parameter in the
`AccessResolver::resolve` method, allowing us to wire them up to
the `P2PGossipSync`'s `MessageSendEventsProvider` implementation.

Author: TheBlueMatt

fuzz/src/router.rs

@@ -99,12 +99,12 @@ impl<Out: test_logger::Output> ChainAccess for FuzzChainSource<'_, '_, Out> {
 			&[1, _] => ChainAccessResult::Sync(Err(ChainAccessError::UnknownTx)),
 			&[2, _] => {
 				let future = AccessFuture::new();
-				future.resolve(self.net_graph, Ok(txo_res));
+				future.resolve_without_forwarding(self.net_graph, Ok(txo_res));
 				ChainAccessResult::Async(future.clone())
 			},
 			&[3, _] => {
 				let future = AccessFuture::new();
-				future.resolve(self.net_graph, Err(ChainAccessError::UnknownTx));
+				future.resolve_without_forwarding(self.net_graph, Err(ChainAccessError::UnknownTx));
 				ChainAccessResult::Async(future.clone())
 			},
 			&[4, _] => {

lightning/src/routing/gossip.rs

@@ -268,6 +268,36 @@ where C::Target: ChainAccess, L::Target: Logger
 			false
 		}
 	}
+
+	/// Used to broadcast forward gossip messages which were validated async.
+	///
+	/// Note that this will ignore events other than `Broadcast*` or messages with too much excess
+	/// data.
+	pub(super) fn forward_gossip_msg(&self, mut ev: MessageSendEvent) {
+		match &mut ev {
+			MessageSendEvent::BroadcastChannelAnnouncement { msg, ref mut update_msg } => {
+				if msg.contents.excess_data.len() > MAX_EXCESS_BYTES_FOR_RELAY { return; }
+				if update_msg.as_ref()
+					.map(|msg| msg.contents.excess_data.len()).unwrap_or(0) > MAX_EXCESS_BYTES_FOR_RELAY
+				{
+					*update_msg = None;
+				}
+			},
+			MessageSendEvent::BroadcastChannelUpdate { msg } => {
+				if msg.contents.excess_data.len() > MAX_EXCESS_BYTES_FOR_RELAY { return; }
+			},
+			MessageSendEvent::BroadcastNodeAnnouncement { msg } => {
+				if msg.contents.excess_data.len() >  MAX_EXCESS_BYTES_FOR_RELAY ||
+				   msg.contents.excess_address_data.len() > MAX_EXCESS_BYTES_FOR_RELAY ||
+				   msg.contents.excess_data.len() + msg.contents.excess_address_data.len() > MAX_EXCESS_BYTES_FOR_RELAY
+				{
+					return;
+				}
+			},
+			_ => return,
+		}
+		self.pending_events.lock().unwrap().push(ev);
+	}
 }
 
 impl<L: Deref> NetworkGraph<L> where L::Target: Logger {

lightning/src/routing/gossip_checking.rs

@@ -20,7 +20,8 @@ use bitcoin::secp256k1::PublicKey;
 
 use crate::ln::chan_utils::make_funding_redeemscript;
 use crate::ln::msgs::{self, LightningError, ErrorAction};
-use crate::routing::gossip::{NetworkGraph, NodeId};
+use crate::routing::gossip::{NetworkGraph, NodeId, P2PGossipSync};
+use crate::util::events::MessageSendEvent;
 use crate::util::logger::{Level, Logger};
 use crate::util::ser::Writeable;
 
@@ -147,8 +148,32 @@ impl AccessFuture {
 	}
 
 	/// Resolves this future against the given `graph` and with the given `result`.
-	pub fn resolve<L: Deref>(&self, graph: &NetworkGraph<L>, result: Result<TxOut, ChainAccessError>)
+	///
+	/// This is identical to calling [`AccessFuture::resolve`] with a dummy `gossip`, disabling
+	/// forwarding the validated gossip message onwards to peers.
+	pub fn resolve_without_forwarding<L: Deref>(&self,
+		graph: &NetworkGraph<L>, result: Result<TxOut, ChainAccessError>)
 	where L::Target: Logger {
+		self.do_resolve(graph, result);
+	}
+
+	/// Resolves this future against the given `graph` and with the given `result`.
+	///
+	/// The given `gossip` is used to broadcast any validated messages onwards to all peers which
+	/// have available buffer space.
+	pub fn resolve<L: Deref, G: Deref<Target=NetworkGraph<L>>, C: Deref, GS: Deref<Target = P2PGossipSync<G, C, L>>>(&self,
+		graph: &NetworkGraph<L>, gossip: GS, result: Result<TxOut, ChainAccessError>
+	) where L::Target: Logger, C::Target: ChainAccess {
+		let mut res = self.do_resolve(graph, result);
+		for msg_opt in res.iter_mut() {
+			if let Some(msg) = msg_opt.take() {
+				gossip.forward_gossip_msg(msg);
+			}
+		}
+	}
+
+	fn do_resolve<L: Deref>(&self, graph: &NetworkGraph<L>, result: Result<TxOut, ChainAccessError>)
+	-> [Option<MessageSendEvent>; 5] where L::Target: Logger {
 		let (announcement, node_a, node_b, update_a, update_b) = {
 			let mut pending_checks = graph.pending_checks.internal.lock().unwrap();
 			let mut async_messages = self.state.lock().unwrap();
@@ -158,7 +183,7 @@ impl AccessFuture {
 				// `channel_announce` yet. That's okay, we can set the `complete` field which it will
 				// check once it gets control again.
 				async_messages.complete = Some(result);
-				return;
+				return [None, None, None, None, None];
 			}
 
 			let announcement_msg = match async_messages.channel_announce.as_ref().unwrap() {
@@ -175,14 +200,22 @@ impl AccessFuture {
 				async_messages.latest_channel_update_b.take())
 		};
 
+		let mut res = [None, None, None, None, None];
+		let mut res_idx = 0;
+
 		// Now that we've updated our internal state, pass the pending messages back through the
 		// network graph with a different `ChainAccess` which will resolve immediately.
 		// Note that we ignore errors as we don't disconnect peers anyway, so there's nothing to do
 		// with them.
 		let resolver = AccessResolver(result);
 		match announcement {
 			ChannelAnnouncement::Full(signed_msg) => {
-				let _ = graph.update_channel_from_announcement(&signed_msg, &Some(&resolver));
+				if graph.update_channel_from_announcement(&signed_msg, &Some(&resolver)).is_ok() {
+					res[res_idx] = Some(MessageSendEvent::BroadcastChannelAnnouncement {
+						msg: signed_msg, update_msg: None,
+					});
+					res_idx += 1;
+				}
 			},
 			ChannelAnnouncement::Unsigned(msg) => {
 				let _ = graph.update_channel_from_unsigned_announcement(&msg, &Some(&resolver));
@@ -192,7 +225,12 @@ impl AccessFuture {
 		for announce in core::iter::once(node_a).chain(core::iter::once(node_b)) {
 			match announce {
 				Some(NodeAnnouncement::Full(signed_msg)) => {
-					let _ = graph.update_node_from_announcement(&signed_msg);
+					if graph.update_node_from_announcement(&signed_msg).is_ok() {
+						res[res_idx] = Some(MessageSendEvent::BroadcastNodeAnnouncement {
+							msg: signed_msg,
+						});
+						res_idx += 1;
+					}
 				},
 				Some(NodeAnnouncement::Unsigned(msg)) => {
 					let _ = graph.update_node_from_unsigned_announcement(&msg);
@@ -204,14 +242,21 @@ impl AccessFuture {
 		for update in core::iter::once(update_a).chain(core::iter::once(update_b)) {
 			match update {
 				Some(ChannelUpdate::Full(signed_msg)) => {
-					let _ = graph.update_channel(&signed_msg);
+					if graph.update_channel(&signed_msg).is_ok() {
+						res[res_idx] = Some(MessageSendEvent::BroadcastChannelUpdate {
+							msg: signed_msg,
+						});
+						res_idx += 1;
+					}
 				},
 				Some(ChannelUpdate::Unsigned(msg)) => {
 					let _ = graph.update_channel_unsigned(&msg);
 				},
 				None => {},
 			}
 		}
+
+		res
 	}
 }