EventHandler decorator for applying NetworkUpdate

PaymentFailed events contain an optional NetworkUpdate describing
changes to the NetworkGraph as conveyed by a node along a failed payment
path according to BOLT 4. An EventHandler should apply the update to the
graph so that future routing decisions can account for it.

Provide an EventHandler decorator that updates the NetworkGraph and
then delegates Event handling to the decorated EventHandler. Previously,
NetGraphMsgHandler::handle_htlc_fail_channel_update implemented this
behavior.

Author: jkczyz

fuzz/src/full_stack.rs

@@ -38,10 +38,10 @@ use lightning::ln::peer_handler::{MessageHandler,PeerManager,SocketDescriptor,Ig
 use lightning::ln::msgs::DecodeError;
 use lightning::ln::script::ShutdownScript;
 use lightning::routing::router::get_route;
-use lightning::routing::network_graph::NetGraphMsgHandler;
+use lightning::routing::network_graph::{NetGraphMsgHandler, NetworkGraph};
 use lightning::util::config::UserConfig;
 use lightning::util::errors::APIError;
-use lightning::util::events::Event;
+use lightning::util::events::{Event, EventHandler};
 use lightning::util::enforcing_trait_impls::EnforcingSigner;
 use lightning::util::logger::Logger;
 use lightning::util::ser::Readable;
@@ -160,12 +160,12 @@ type ChannelMan = ChannelManager<
 	EnforcingSigner,
 	Arc<chainmonitor::ChainMonitor<EnforcingSigner, Arc<dyn chain::Filter>, Arc<TestBroadcaster>, Arc<FuzzEstimator>, Arc<dyn Logger>, Arc<TestPersister>>>,
 	Arc<TestBroadcaster>, Arc<KeyProvider>, Arc<FuzzEstimator>, Arc<dyn Logger>>;
-type PeerMan<'a> = PeerManager<Peer<'a>, Arc<ChannelMan>, Arc<NetGraphMsgHandler<Arc<dyn chain::Access>, Arc<dyn Logger>>>, Arc<dyn Logger>, IgnoringMessageHandler>;
+type PeerMan<'a, E> = PeerManager<Peer<'a>, Arc<ChannelMan>, Arc<NetGraphMsgHandler<Arc<dyn chain::Access>, Arc<dyn Logger>, E>>, Arc<dyn Logger>, IgnoringMessageHandler>;
 
-struct MoneyLossDetector<'a> {
+struct MoneyLossDetector<'a, E: EventHandler> {
 	manager: Arc<ChannelMan>,
 	monitor: Arc<chainmonitor::ChainMonitor<EnforcingSigner, Arc<dyn chain::Filter>, Arc<TestBroadcaster>, Arc<FuzzEstimator>, Arc<dyn Logger>, Arc<TestPersister>>>,
-	handler: PeerMan<'a>,
+	handler: PeerMan<'a, E>,
 
 	peers: &'a RefCell<[bool; 256]>,
 	funding_txn: Vec<Transaction>,
@@ -175,11 +175,11 @@ struct MoneyLossDetector<'a> {
 	max_height: usize,
 	blocks_connected: u32,
 }
-impl<'a> MoneyLossDetector<'a> {
+impl<'a, E: EventHandler> MoneyLossDetector<'a, E> {
 	pub fn new(peers: &'a RefCell<[bool; 256]>,
 	           manager: Arc<ChannelMan>,
 	           monitor: Arc<chainmonitor::ChainMonitor<EnforcingSigner, Arc<dyn chain::Filter>, Arc<TestBroadcaster>, Arc<FuzzEstimator>, Arc<dyn Logger>, Arc<TestPersister>>>,
-	           handler: PeerMan<'a>) -> Self {
+	           handler: PeerMan<'a, E>) -> Self {
 		MoneyLossDetector {
 			manager,
 			monitor,
@@ -238,7 +238,7 @@ impl<'a> MoneyLossDetector<'a> {
 	}
 }
 
-impl<'a> Drop for MoneyLossDetector<'a> {
+impl<'a, E: EventHandler> Drop for MoneyLossDetector<'a, E> {
 	fn drop(&mut self) {
 		if !::std::thread::panicking() {
 			// Disconnect all peers
@@ -371,7 +371,8 @@ pub fn do_test(data: &[u8], logger: &Arc<dyn Logger>) {
 	};
 	let channelmanager = Arc::new(ChannelManager::new(fee_est.clone(), monitor.clone(), broadcast.clone(), Arc::clone(&logger), keys_manager.clone(), config, params));
 	let our_id = PublicKey::from_secret_key(&Secp256k1::signing_only(), &keys_manager.get_node_secret());
-	let net_graph_msg_handler = Arc::new(NetGraphMsgHandler::new(genesis_block(network).block_hash(), None, Arc::clone(&logger)));
+	let network_graph = NetworkGraph::new(genesis_block(network).block_hash());
+	let net_graph_msg_handler = Arc::new(NetGraphMsgHandler::new(network_graph, None, Arc::clone(&logger), |_: &_| {}));
 
 	let peers = RefCell::new([false; 256]);
 	let mut loss_detector = MoneyLossDetector::new(&peers, channelmanager.clone(), monitor.clone(), PeerManager::new(MessageHandler {

lightning-net-tokio/src/lib.rs

@@ -37,10 +37,10 @@
 //! type DataPersister = dyn lightning::chain::channelmonitor::Persist<lightning::chain::keysinterface::InMemorySigner> + Send + Sync;
 //! type ChainMonitor = lightning::chain::chainmonitor::ChainMonitor<lightning::chain::keysinterface::InMemorySigner, Arc<ChainFilter>, Arc<TxBroadcaster>, Arc<FeeEstimator>, Arc<Logger>, Arc<DataPersister>>;
 //! type ChannelManager = Arc<lightning::ln::channelmanager::SimpleArcChannelManager<ChainMonitor, TxBroadcaster, FeeEstimator, Logger>>;
-//! type PeerManager = Arc<lightning::ln::peer_handler::SimpleArcPeerManager<lightning_net_tokio::SocketDescriptor, ChainMonitor, TxBroadcaster, FeeEstimator, ChainAccess, Logger>>;
+//! type PeerManager<E> = Arc<lightning::ln::peer_handler::SimpleArcPeerManager<lightning_net_tokio::SocketDescriptor, ChainMonitor, TxBroadcaster, FeeEstimator, ChainAccess, Logger, E>>;
 //!
 //! // Connect to node with pubkey their_node_id at addr:
-//! async fn connect_to_node(peer_manager: PeerManager, chain_monitor: Arc<ChainMonitor>, channel_manager: ChannelManager, their_node_id: PublicKey, addr: SocketAddr) {
+//! async fn connect_to_node<E: 'static + EventHandler + Send + Sync>(peer_manager: PeerManager<E>, chain_monitor: Arc<ChainMonitor>, channel_manager: ChannelManager, their_node_id: PublicKey, addr: SocketAddr) {
 //! 	lightning_net_tokio::connect_outbound(peer_manager, their_node_id, addr).await;
 //! 	loop {
 //! 		let event_handler = |event: &Event| {
@@ -53,7 +53,7 @@
 //! }
 //!
 //! // Begin reading from a newly accepted socket and talk to the peer:
-//! async fn accept_socket(peer_manager: PeerManager, chain_monitor: Arc<ChainMonitor>, channel_manager: ChannelManager, socket: TcpStream) {
+//! async fn accept_socket<E: 'static + EventHandler + Send + Sync>(peer_manager: PeerManager<E>, chain_monitor: Arc<ChainMonitor>, channel_manager: ChannelManager, socket: TcpStream) {
 //! 	lightning_net_tokio::setup_inbound(peer_manager, socket);
 //! 	loop {
 //! 		let event_handler = |event: &Event| {

lightning/src/ln/functional_test_utils.rs

@@ -208,7 +208,7 @@ pub struct Node<'a, 'b: 'a, 'c: 'b> {
 	pub chain_monitor: &'b test_utils::TestChainMonitor<'c>,
 	pub keys_manager: &'b test_utils::TestKeysInterface,
 	pub node: &'a ChannelManager<EnforcingSigner, &'b TestChainMonitor<'c>, &'c test_utils::TestBroadcaster, &'b test_utils::TestKeysInterface, &'c test_utils::TestFeeEstimator, &'c test_utils::TestLogger>,
-	pub net_graph_msg_handler: NetGraphMsgHandler<&'c test_utils::TestChainSource, &'c test_utils::TestLogger>,
+	pub net_graph_msg_handler: NetGraphMsgHandler<&'c test_utils::TestChainSource, &'c test_utils::TestLogger, test_utils::TestEventHandler>,
 	pub node_seed: [u8; 32],
 	pub network_payment_count: Rc<RefCell<u8>>,
 	pub network_chan_count: Rc<RefCell<u32>>,
@@ -243,8 +243,8 @@ impl<'a, 'b, 'c> Drop for Node<'a, 'b, 'c> {
 				network_graph_ser.write(&mut w).unwrap();
 				let network_graph_deser = <NetworkGraph>::read(&mut io::Cursor::new(&w.0)).unwrap();
 				assert!(network_graph_deser == self.net_graph_msg_handler.network_graph);
-				let net_graph_msg_handler = NetGraphMsgHandler::from_net_graph(
-					Some(self.chain_source), self.logger, network_graph_deser
+				let net_graph_msg_handler = NetGraphMsgHandler::new(
+					network_graph_deser, Some(self.chain_source), self.logger, test_utils::TestEventHandler {}
 				);
 				let mut chan_progress = 0;
 				loop {
@@ -1443,7 +1443,8 @@ pub fn create_network<'a, 'b: 'a, 'c: 'b>(node_count: usize, cfgs: &'b Vec<NodeC
 	let connect_style = Rc::new(RefCell::new(ConnectStyle::FullBlockViaListen));
 
 	for i in 0..node_count {
-		let net_graph_msg_handler = NetGraphMsgHandler::new(cfgs[i].chain_source.genesis_hash, None, cfgs[i].logger);
+		let network_graph = NetworkGraph::new(cfgs[i].chain_source.genesis_hash);
+		let net_graph_msg_handler = NetGraphMsgHandler::new(network_graph, None, cfgs[i].logger, test_utils::TestEventHandler {});
 		nodes.push(Node{ chain_source: cfgs[i].chain_source,
 		                 tx_broadcaster: cfgs[i].tx_broadcaster, chain_monitor: &cfgs[i].chain_monitor,
 		                 keys_manager: &cfgs[i].keys_manager, node: &chan_mgrs[i], net_graph_msg_handler,

lightning/src/ln/peer_handler.rs

@@ -354,15 +354,15 @@ fn _check_usize_is_32_or_64() {
 /// lifetimes). Other times you can afford a reference, which is more efficient, in which case
 /// SimpleRefPeerManager is the more appropriate type. Defining these type aliases prevents
 /// issues such as overly long function definitions.
-pub type SimpleArcPeerManager<SD, M, T, F, C, L> = PeerManager<SD, Arc<SimpleArcChannelManager<M, T, F, L>>, Arc<NetGraphMsgHandler<Arc<C>, Arc<L>>>, Arc<L>, Arc<IgnoringMessageHandler>>;
+pub type SimpleArcPeerManager<SD, M, T, F, C, L, E> = PeerManager<SD, Arc<SimpleArcChannelManager<M, T, F, L>>, Arc<NetGraphMsgHandler<Arc<C>, Arc<L>, E>>, Arc<L>, Arc<IgnoringMessageHandler>>;
 
 /// SimpleRefPeerManager is a type alias for a PeerManager reference, and is the reference
 /// counterpart to the SimpleArcPeerManager type alias. Use this type by default when you don't
 /// need a PeerManager with a static lifetime. You'll need a static lifetime in cases such as
 /// usage of lightning-net-tokio (since tokio::spawn requires parameters with static lifetimes).
 /// But if this is not necessary, using a reference is more efficient. Defining these type aliases
 /// helps with issues such as long function definitions.
-pub type SimpleRefPeerManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, SD, M, T, F, C, L> = PeerManager<SD, SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, M, T, F, L>, &'e NetGraphMsgHandler<&'g C, &'f L>, &'f L, IgnoringMessageHandler>;
+pub type SimpleRefPeerManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, SD, M, T, F, C, L, E> = PeerManager<SD, SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, M, T, F, L>, &'e NetGraphMsgHandler<&'g C, &'f L, E>, &'f L, IgnoringMessageHandler>;
 
 /// A PeerManager manages a set of peers, described by their [`SocketDescriptor`] and marshalls
 /// socket events into messages which it passes on to its [`MessageHandler`].