Score payment paths from events in BackgroundProcessor

Author: valentinewallace

lightning-background-processor/src/lib.rs

@@ -16,6 +16,9 @@
 #[cfg(any(test, feature = "std"))]
 extern crate core;
 
+#[cfg(not(feature = "std"))]
+extern crate alloc;
+
 #[macro_use] extern crate lightning;
 extern crate lightning_rapid_gossip_sync;
 
@@ -28,7 +31,7 @@ use lightning::ln::msgs::{ChannelMessageHandler, OnionMessageHandler, RoutingMes
 use lightning::ln::peer_handler::{CustomMessageHandler, PeerManager, SocketDescriptor};
 use lightning::routing::gossip::{NetworkGraph, P2PGossipSync};
 use lightning::routing::router::Router;
-use lightning::routing::scoring::WriteableScore;
+use lightning::routing::scoring::{Score, WriteableScore};
 use lightning::util::events::{Event, EventHandler, EventsProvider};
 use lightning::util::logger::Logger;
 use lightning::util::persist::Persister;
@@ -49,6 +52,8 @@ use std::time::Instant;
 
 #[cfg(feature = "futures")]
 use futures_util::{select_biased, future::FutureExt, task};
+#[cfg(not(feature = "std"))]
+use alloc::vec::Vec;
 
 /// `BackgroundProcessor` takes care of tasks that (1) need to happen periodically to keep
 /// Rust-Lightning running properly, and (2) either can or should be run in the background. Its
@@ -216,6 +221,31 @@ fn handle_network_graph_update<L: Deref>(
 	}
 }
 
+fn update_scorer<'a, S: 'static + Deref<Target = SC> + Send + Sync, SC: 'a + WriteableScore<'a>>(
+	scorer: &'a S, event: &Event
+) {
+	let mut score = scorer.lock();
+	match event {
+		Event::PaymentPathFailed { ref path, short_channel_id: Some(scid), .. } => {
+			let path = path.iter().collect::<Vec<_>>();
+			score.payment_path_failed(&path, *scid);
+		},
+		Event::PaymentPathSuccessful { path, .. } => {
+			let path = path.iter().collect::<Vec<_>>();
+			score.payment_path_successful(&path);
+		},
+		Event::ProbeSuccessful { path, .. } => {
+			let path = path.iter().collect::<Vec<_>>();
+			score.probe_successful(&path);
+		},
+		Event::ProbeFailed { path, short_channel_id: Some(scid), .. } => {
+			let path = path.iter().collect::<Vec<_>>();
+			score.probe_failed(&path, *scid);
+		},
+		_ => {},
+	}
+}
+
 macro_rules! define_run_body {
 	($persister: ident, $chain_monitor: ident, $process_chain_monitor_events: expr,
 	 $channel_manager: ident, $process_channel_manager_events: expr,
@@ -387,7 +417,7 @@ pub async fn process_events_async<
 	UMH: 'static + Deref + Send + Sync,
 	PM: 'static + Deref<Target = PeerManager<Descriptor, CMH, RMH, OMH, L, UMH, NS>> + Send + Sync,
 	S: 'static + Deref<Target = SC> + Send + Sync,
-	SC: WriteableScore<'a>,
+	SC: for<'b> WriteableScore<'b>,
 	SleepFuture: core::future::Future<Output = bool> + core::marker::Unpin,
 	Sleeper: Fn(Duration) -> SleepFuture
 >(
@@ -417,10 +447,14 @@ where
 	let async_event_handler = |event| {
 		let network_graph = gossip_sync.network_graph();
 		let event_handler = &event_handler;
+		let scorer = &scorer;
 		async move {
 			if let Some(network_graph) = network_graph {
 				handle_network_graph_update(network_graph, &event)
 			}
+			if let Some(ref scorer) = scorer {
+				update_scorer(scorer, &event);
+			}
 			event_handler(event).await;
 		}
 	};
@@ -516,7 +550,7 @@ impl BackgroundProcessor {
 		UMH: 'static + Deref + Send + Sync,
 		PM: 'static + Deref<Target = PeerManager<Descriptor, CMH, RMH, OMH, L, UMH, NS>> + Send + Sync,
 		S: 'static + Deref<Target = SC> + Send + Sync,
-		SC: WriteableScore<'a>,
+		SC: for <'b> WriteableScore<'b>,
 	>(
 		persister: PS, event_handler: EH, chain_monitor: M, channel_manager: CM,
 		gossip_sync: GossipSync<PGS, RGS, G, CA, L>, peer_manager: PM, logger: L, scorer: Option<S>,
@@ -547,6 +581,9 @@ impl BackgroundProcessor {
 				if let Some(network_graph) = network_graph {
 					handle_network_graph_update(network_graph, &event)
 				}
+				if let Some(ref scorer) = scorer {
+					update_scorer(scorer, &event);
+				}
 				event_handler.handle_event(event);
 			};
 			define_run_body!(persister, chain_monitor, chain_monitor.process_pending_events(&event_handler),
@@ -618,15 +655,16 @@ mod tests {
 	use lightning::chain::keysinterface::{InMemorySigner, EntropySource, KeysManager};
 	use lightning::chain::transaction::OutPoint;
 	use lightning::get_event_msg;
-	use lightning::ln::channelmanager::{BREAKDOWN_TIMEOUT, ChainParameters, ChannelManager, SimpleArcChannelManager};
+	use lightning::ln::channelmanager::{BREAKDOWN_TIMEOUT, ChainParameters, ChannelManager, PaymentId, Retry as ChannelManagerRetry, SimpleArcChannelManager};
+	use lightning::ln::functional_test_utils::*;
 	use lightning::ln::features::ChannelFeatures;
-	use lightning::ln::msgs::{ChannelMessageHandler, Init};
+	use lightning::ln::msgs::{ChannelMessageHandler, Init, RoutingMessageHandler};
 	use lightning::ln::peer_handler::{PeerManager, MessageHandler, SocketDescriptor, IgnoringMessageHandler};
-	use lightning::routing::gossip::{NetworkGraph, P2PGossipSync};
-	use lightning::routing::router::DefaultRouter;
+	use lightning::routing::gossip::{NetworkGraph, NodeId, P2PGossipSync};
+	use lightning::routing::router::{DefaultRouter, PaymentParameters, RouteParameters};
 	use lightning::routing::scoring::{ProbabilisticScoringParameters, ProbabilisticScorer};
 	use lightning::util::config::UserConfig;
-	use lightning::util::events::{Event, MessageSendEventsProvider, MessageSendEvent};
+	use lightning::util::events::{Event, MessageSendEventsProvider, MessageSendEvent, PaymentPurpose};
 	use lightning::util::ser::Writeable;
 	use lightning::util::test_utils;
 	use lightning::util::persist::KVStorePersister;
@@ -782,8 +820,8 @@ mod tests {
 			let chain_monitor = Arc::new(chainmonitor::ChainMonitor::new(Some(chain_source.clone()), tx_broadcaster.clone(), logger.clone(), fee_estimator.clone(), persister.clone()));
 			let best_block = BestBlock::from_genesis(network);
 			let params = ChainParameters { network, best_block };
-			let manager = Arc::new(ChannelManager::new(fee_estimator.clone(), chain_monitor.clone(), tx_broadcaster.clone(), router.clone(), logger.clone(), keys_manager.clone(), keys_manager.clone(), keys_manager.clone(), UserConfig::default(), params));
-			let p2p_gossip_sync = Arc::new(P2PGossipSync::new(network_graph.clone(), Some(chain_source.clone()), logger.clone()));
+			let manager = Arc::new(ChannelManager::new(fee_estimator.clone(), chain_monitor.clone(), tx_broadcaster.clone(), router.clone(), logger.clone(), keys_manager.clone(), keys_manager.clone(), keys_manager.clone(), test_default_channel_config(), params));
+			let p2p_gossip_sync = Arc::new(P2PGossipSync::new(network_graph.clone(), None, logger.clone()));
 			let rapid_gossip_sync = Arc::new(RapidGossipSync::new(network_graph.clone()));
 			let msg_handler = MessageHandler { chan_handler: Arc::new(test_utils::TestChannelMessageHandler::new()), route_handler: Arc::new(test_utils::TestRoutingMessageHandler::new()), onion_message_handler: IgnoringMessageHandler{}};
 			let peer_manager = Arc::new(PeerManager::new(msg_handler, 0, &seed, logger.clone(), IgnoringMessageHandler{}, keys_manager.clone()));
@@ -803,7 +841,7 @@ mod tests {
 
 	macro_rules! open_channel {
 		($node_a: expr, $node_b: expr, $channel_value: expr) => {{
-			begin_open_channel!($node_a, $node_b, $channel_value);
+			begin_open_channel!($node_a, $node_b, $channel_value, None);
 			let events = $node_a.node.get_and_clear_pending_events();
 			assert_eq!(events.len(), 1);
 			let (temporary_channel_id, tx) = handle_funding_generation_ready!(events[0], $channel_value);
@@ -813,8 +851,8 @@ mod tests {
 	}
 
 	macro_rules! begin_open_channel {
-		($node_a: expr, $node_b: expr, $channel_value: expr) => {{
-			$node_a.node.create_channel($node_b.node.get_our_node_id(), $channel_value, 100, 42, None).unwrap();
+		($node_a: expr, $node_b: expr, $channel_value: expr, $override_cfg: expr) => {{
+			$node_a.node.create_channel($node_b.node.get_our_node_id(), $channel_value, 100, 42, $override_cfg).unwrap();
 			$node_b.node.handle_open_channel(&$node_a.node.get_our_node_id(), &get_event_msg!($node_a, MessageSendEvent::SendOpenChannel, $node_b.node.get_our_node_id()));
 			$node_a.node.handle_accept_channel(&$node_b.node.get_our_node_id(), &get_event_msg!($node_b, MessageSendEvent::SendAcceptChannel, $node_a.node.get_our_node_id()));
 		}}
@@ -1034,7 +1072,7 @@ mod tests {
 		let bg_processor = BackgroundProcessor::start(persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), Some(nodes[0].scorer.clone()));
 
 		// Open a channel and check that the FundingGenerationReady event was handled.
-		begin_open_channel!(nodes[0], nodes[1], channel_value);
+		begin_open_channel!(nodes[0], nodes[1], channel_value, Some(UserConfig::default()));
 		let (temporary_channel_id, funding_tx) = receiver
 			.recv_timeout(Duration::from_secs(EVENT_DEADLINE))
 			.expect("FundingGenerationReady not handled within deadline");
@@ -1171,4 +1209,164 @@ mod tests {
 		let bg_processor = BackgroundProcessor::start(persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), Some(nodes[0].scorer.clone()));
 		assert!(bg_processor.stop().is_ok());
 	}
+
+	#[test]
+	fn test_payment_path_scoring() {
+		// Ensure that we update the scorer when relevant events are processed. In this case, we ensure
+		// that we update the scorer upon a payment path succeeding (note that the channel must be
+		// public or else we won't score it).
+		let mut nodes = create_nodes(2, "test_payment_path_scoring".to_string());
+		let channel_value = 100000;
+		let data_dir_0 = nodes[0].persister.get_data_dir();
+		let persister_0 = Arc::new(Persister::new(data_dir_0.clone()));
+
+		// Set up a background event handler for FundingGenerationReady events.
+		let (sender, receiver) = std::sync::mpsc::sync_channel(1);
+		let event_handler_0 = move |event: Event| match event {
+			Event::FundingGenerationReady { .. } => sender.send(handle_funding_generation_ready!(event, channel_value)).unwrap(),
+			Event::ChannelReady { .. } => {},
+			_ => panic!("Unexpected event: {:?}", event),
+		};
+
+		let bg_processor_0 = BackgroundProcessor::start(persister_0, event_handler_0, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), Some(nodes[0].scorer.clone()));
+
+		// Open a channel and check that the FundingGenerationReady event was handled.
+		begin_open_channel!(nodes[0], nodes[1], channel_value, None);
+		let (temporary_channel_id, funding_tx) = receiver
+			.recv_timeout(Duration::from_secs(EVENT_DEADLINE))
+			.expect("FundingGenerationReady not handled within deadline");
+		end_open_channel!(nodes[0], nodes[1], temporary_channel_id, funding_tx);
+
+		// Confirm the funding transaction.
+		confirm_transaction(&mut nodes[0], &funding_tx);
+		let as_funding = get_event_msg!(nodes[0], MessageSendEvent::SendChannelReady, nodes[1].node.get_our_node_id());
+		confirm_transaction(&mut nodes[1], &funding_tx);
+		let bs_funding = get_event_msg!(nodes[1], MessageSendEvent::SendChannelReady, nodes[0].node.get_our_node_id());
+		nodes[0].node.handle_channel_ready(&nodes[1].node.get_our_node_id(), &bs_funding);
+		let as_ann_sigs = get_event_msg!(nodes[0], MessageSendEvent::SendAnnouncementSignatures, nodes[1].node.get_our_node_id());
+		nodes[1].node.handle_channel_ready(&nodes[0].node.get_our_node_id(), &as_funding);
+		let bs_ann_sigs = get_event_msg!(nodes[1], MessageSendEvent::SendAnnouncementSignatures, nodes[0].node.get_our_node_id());
+		nodes[1].node.handle_announcement_signatures(&nodes[0].node.get_our_node_id(), &as_ann_sigs);
+		let events = nodes[1].node.get_and_clear_pending_msg_events();
+		assert_eq!(events.len(), 1);
+		let (ann, bs_update) = match events[0] {
+			MessageSendEvent::BroadcastChannelAnnouncement { ref msg, ref update_msg } => {
+				(msg, update_msg)
+			},
+			_ => panic!("Unexpected event"),
+		};
+		let scid = bs_update.contents.short_channel_id;
+		nodes[0].node.handle_announcement_signatures(&nodes[1].node.get_our_node_id(), &bs_ann_sigs);
+		let events = nodes[0].node.get_and_clear_pending_msg_events();
+		assert_eq!(events.len(), 1);
+		let as_update = match events[0] {
+			MessageSendEvent::BroadcastChannelAnnouncement { ref msg, ref update_msg } => {
+				assert!(*ann == *msg);
+				assert_eq!(update_msg.contents.short_channel_id, ann.contents.short_channel_id);
+				assert_eq!(update_msg.contents.short_channel_id, bs_update.contents.short_channel_id);
+				update_msg
+			},
+			_ => panic!("Unexpected event"),
+		};
+
+		for node in nodes.iter() {
+			assert!(node.p2p_gossip_sync.handle_channel_announcement(ann).unwrap());
+			node.p2p_gossip_sync.handle_channel_update(as_update).unwrap();
+			node.p2p_gossip_sync.handle_channel_update(bs_update).unwrap();
+
+			node.node.handle_channel_update(&nodes[0].node.get_our_node_id(), &as_update);
+			node.node.handle_channel_update(&nodes[1].node.get_our_node_id(), &bs_update);
+		}
+
+		assert!(bg_processor_0.stop().is_ok());
+
+		let (sender_0, receiver_0) = std::sync::mpsc::sync_channel(1);
+		let event_handler_0 = move |event: Event| match event {
+			Event::ChannelReady { .. } => {},
+			Event::PendingHTLCsForwardable { .. } => {},
+			Event::PaymentSent { .. } => {},
+			Event::PaymentPathSuccessful { .. } => sender_0.send(event).unwrap(),
+			_ => panic!("Unexpected event: {:?}", event),
+		};
+		let persister_0 = Arc::new(Persister::new(data_dir_0));
+		let _bg_processor_0 = BackgroundProcessor::start(persister_0, event_handler_0, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), Some(nodes[0].scorer.clone()));
+
+		let amt_msat = 40000;
+		let payment_params = PaymentParameters::from_node_id(nodes[1].node.get_our_node_id())
+			.with_features(nodes[1].node.invoice_features());
+		let route_params = RouteParameters {
+			payment_params,
+			final_value_msat: amt_msat,
+			final_cltv_expiry_delta: 70,
+		};
+		let (payment_hash, payment_secret) = nodes[1].node.create_inbound_payment(Some(amt_msat), 60*60, None).unwrap();
+
+		let (sender_1, receiver_1) = std::sync::mpsc::sync_channel(1);
+		let event_handler_1 = move |event: Event| match event {
+			Event::ChannelReady { .. } => {},
+			Event::PendingHTLCsForwardable { .. } => {},
+			Event::PaymentClaimable { .. } => sender_1.send(event).unwrap(),
+			Event::PaymentClaimed { .. } => {},
+			_ => panic!("Unexpected event: {:?}", event),
+		};
+		let data_dir_1 = nodes[1].persister.get_data_dir();
+		let persister_1 = Arc::new(Persister::new(data_dir_1.clone()));
+		let _bg_processor_1 = BackgroundProcessor::start(persister_1, event_handler_1, nodes[1].chain_monitor.clone(), nodes[1].node.clone(), nodes[1].no_gossip_sync(), nodes[1].peer_manager.clone(), nodes[1].logger.clone(), Some(nodes[1].scorer.clone()));
+
+			nodes[0].node.send_payment_with_retry(payment_hash, &Some(payment_secret), PaymentId(payment_hash.0), route_params, ChannelManagerRetry::Attempts(0)).unwrap();
+
+		let mut msg_events = nodes[0].node.get_and_clear_pending_msg_events();
+
+		assert_eq!(msg_events.len(), 1);
+		let payment_event = SendEvent::from_event(msg_events.remove(0));
+
+		nodes[1].node.handle_update_add_htlc(&nodes[0].node.get_our_node_id(), &payment_event.msgs[0]);
+		nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), &payment_event.commitment_msg);
+		let (bs_first_raa, bs_first_cs) = get_revoke_commit_msgs!(nodes[1], nodes[0].node.get_our_node_id());
+
+		nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &bs_first_raa);
+		nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &bs_first_cs);
+		let as_first_raa = get_event_msg!(nodes[0], MessageSendEvent::SendRevokeAndACK, nodes[1].node.get_our_node_id());
+
+		nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &as_first_raa);
+		nodes[1].node.process_pending_htlc_forwards();
+
+		let event = receiver_1
+			.recv_timeout(Duration::from_secs(EVENT_DEADLINE))
+			.expect("Events not handled within deadline");
+		match event {
+			Event::PaymentClaimable { purpose: PaymentPurpose::InvoicePayment { payment_preimage: Some(payment_preimage), .. }, .. } => nodes[1].node.claim_funds(payment_preimage),
+			Event::ChannelReady { .. } => {},
+			_ => panic!("Unexpected event: {:?}", event),
+		}
+
+		nodes[1].node.process_pending_htlc_forwards();
+		let update_1 = get_htlc_update_msgs!(nodes[1], nodes[0].node.get_our_node_id());
+		assert!(update_1.update_fulfill_htlcs.len() == 1);
+		let fulfill_msg = update_1.update_fulfill_htlcs[0].clone();
+
+		nodes[0].node.handle_update_fulfill_htlc(&nodes[1].node.get_our_node_id(), &fulfill_msg);
+		nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &update_1.commitment_signed);
+		let (as_first_raa, as_first_cs) = get_revoke_commit_msgs!(nodes[0], nodes[1].node.get_our_node_id());
+
+		nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &as_first_raa);
+		nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), &as_first_cs);
+		let bs_first_raa = get_event_msg!(nodes[1], MessageSendEvent::SendRevokeAndACK, nodes[0].node.get_our_node_id());
+
+		// Ensure that nodes[0] has no capacity data for its channel before the PaymentPathSuccess event
+		// is processed.
+		assert_eq!(nodes[0].scorer.lock().unwrap().estimated_channel_liquidity_range(scid, &NodeId::from_pubkey(&nodes[0].node.get_our_node_id())), None);
+
+		nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &bs_first_raa);
+
+		let event = receiver_0
+			.recv_timeout(Duration::from_secs(EVENT_DEADLINE))
+			.expect("Events not handled within deadline");
+		match event {
+			Event::PaymentPathSuccessful { .. } => {
+				assert_eq!(nodes[0].scorer.lock().unwrap().estimated_channel_liquidity_range(scid, &NodeId::from_pubkey(&nodes[0].node.get_our_node_id())), Some((40000, 10000000)));
+			},
+			_ => panic!("Unexpected event: {:?}", event),
+		}
+	}
 }