Merge pull request #2656 from TheBlueMatt/2023-09-scoring-decay-timer

TheBlueMatt · web-flow · commit c92db6918314 · 2023-12-15T20:06:30.000Z
Stop decaying liquidity information during scoring
diff --git a/bench/benches/bench.rs b/bench/benches/bench.rs
@@ -21,5 +21,6 @@ criterion_group!(benches,
 	lightning_persister::fs_store::bench::bench_sends,
 	lightning_rapid_gossip_sync::bench::bench_reading_full_graph_from_file,
 	lightning::routing::gossip::benches::read_network_graph,
-	lightning::routing::gossip::benches::write_network_graph);
+	lightning::routing::gossip::benches::write_network_graph,
+	lightning::routing::scoring::benches::decay_100k_channel_bounds);
 criterion_main!(benches);
diff --git a/lightning-background-processor/src/lib.rs b/lightning-background-processor/src/lib.rs
@@ -113,7 +113,7 @@ const ONION_MESSAGE_HANDLER_TIMER: u64 = 1;
 const NETWORK_PRUNE_TIMER: u64 = 60 * 60;
 
 #[cfg(not(test))]
-const SCORER_PERSIST_TIMER: u64 = 60 * 60;
+const SCORER_PERSIST_TIMER: u64 = 60 * 5;
 #[cfg(test)]
 const SCORER_PERSIST_TIMER: u64 = 1;
 
@@ -244,30 +244,30 @@ fn handle_network_graph_update<L: Deref>(
 /// Updates scorer based on event and returns whether an update occurred so we can decide whether
 /// to persist.
 fn update_scorer<'a, S: 'static + Deref<Target = SC> + Send + Sync, SC: 'a + WriteableScore<'a>>(
-	scorer: &'a S, event: &Event
+	scorer: &'a S, event: &Event, duration_since_epoch: Duration,
 ) -> bool {
 	match event {
 		Event::PaymentPathFailed { ref path, short_channel_id: Some(scid), .. } => {
 			let mut score = scorer.write_lock();
-			score.payment_path_failed(path, *scid);
+			score.payment_path_failed(path, *scid, duration_since_epoch);
 		},
 		Event::PaymentPathFailed { ref path, payment_failed_permanently: true, .. } => {
 			// Reached if the destination explicitly failed it back. We treat this as a successful probe
 			// because the payment made it all the way to the destination with sufficient liquidity.
 			let mut score = scorer.write_lock();
-			score.probe_successful(path);
+			score.probe_successful(path, duration_since_epoch);
 		},
 		Event::PaymentPathSuccessful { path, .. } => {
 			let mut score = scorer.write_lock();
-			score.payment_path_successful(path);
+			score.payment_path_successful(path, duration_since_epoch);
 		},
 		Event::ProbeSuccessful { path, .. } => {
 			let mut score = scorer.write_lock();
-			score.probe_successful(path);
+			score.probe_successful(path, duration_since_epoch);
 		},
 		Event::ProbeFailed { path, short_channel_id: Some(scid), .. } => {
 			let mut score = scorer.write_lock();
-			score.probe_failed(path, *scid);
+			score.probe_failed(path, *scid, duration_since_epoch);
 		},
 		_ => return false,
 	}
@@ -280,7 +280,7 @@ macro_rules! define_run_body {
 		$channel_manager: ident, $process_channel_manager_events: expr,
 		$peer_manager: ident, $process_onion_message_handler_events: expr, $gossip_sync: ident,
 		$logger: ident, $scorer: ident, $loop_exit_check: expr, $await: expr, $get_timer: expr,
-		$timer_elapsed: expr, $check_slow_await: expr
+		$timer_elapsed: expr, $check_slow_await: expr, $time_fetch: expr,
 	) => { {
 		log_trace!($logger, "Calling ChannelManager's timer_tick_occurred on startup");
 		$channel_manager.timer_tick_occurred();
@@ -294,6 +294,7 @@ macro_rules! define_run_body {
 		let mut last_scorer_persist_call = $get_timer(SCORER_PERSIST_TIMER);
 		let mut last_rebroadcast_call = $get_timer(REBROADCAST_TIMER);
 		let mut have_pruned = false;
+		let mut have_decayed_scorer = false;
 
 		loop {
 			$process_channel_manager_events;
@@ -383,11 +384,10 @@ macro_rules! define_run_body {
 			if should_prune {
 				// The network graph must not be pruned while rapid sync completion is pending
 				if let Some(network_graph) = $gossip_sync.prunable_network_graph() {
-					#[cfg(feature = "std")] {
+					if let Some(duration_since_epoch) = $time_fetch() {
 						log_trace!($logger, "Pruning and persisting network graph.");
-						network_graph.remove_stale_channels_and_tracking();
-					}
-					#[cfg(not(feature = "std"))] {
+						network_graph.remove_stale_channels_and_tracking_with_time(duration_since_epoch.as_secs());
+					} else {
 						log_warn!($logger, "Not pruning network graph, consider enabling `std` or doing so manually with remove_stale_channels_and_tracking_with_time.");
 						log_trace!($logger, "Persisting network graph.");
 					}
@@ -402,9 +402,24 @@ macro_rules! define_run_body {
 				last_prune_call = $get_timer(prune_timer);
 			}
 
+			if !have_decayed_scorer {
+				if let Some(ref scorer) = $scorer {
+					if let Some(duration_since_epoch) = $time_fetch() {
+						log_trace!($logger, "Calling time_passed on scorer at startup");
+						scorer.write_lock().time_passed(duration_since_epoch);
+					}
+				}
+				have_decayed_scorer = true;
+			}
+
 			if $timer_elapsed(&mut last_scorer_persist_call, SCORER_PERSIST_TIMER) {
 				if let Some(ref scorer) = $scorer {
-					log_trace!($logger, "Persisting scorer");
+					if let Some(duration_since_epoch) = $time_fetch() {
+						log_trace!($logger, "Calling time_passed and persisting scorer");
+						scorer.write_lock().time_passed(duration_since_epoch);
+					} else {
+						log_trace!($logger, "Persisting scorer");
+					}
 					if let Err(e) = $persister.persist_scorer(&scorer) {
 						log_error!($logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
 					}
@@ -510,12 +525,16 @@ use core::task;
 /// are unsure, you should set the flag, as the performance impact of it is minimal unless there
 /// are hundreds or thousands of simultaneous process calls running.
 ///
+/// The `fetch_time` parameter should return the current wall clock time, if one is available. If
+/// no time is available, some features may be disabled, however the node will still operate fine.
+///
 /// For example, in order to process background events in a [Tokio](https://tokio.rs/) task, you
 /// could setup `process_events_async` like this:
 /// ```
 /// # use lightning::io;
 /// # use std::sync::{Arc, RwLock};
 /// # use std::sync::atomic::{AtomicBool, Ordering};
+/// # use std::time::SystemTime;
 /// # use lightning_background_processor::{process_events_async, GossipSync};
 /// # struct MyStore {}
 /// # impl lightning::util::persist::KVStore for MyStore {
@@ -584,6 +603,7 @@ use core::task;
 ///			Some(background_scorer),
 ///			sleeper,
 ///			mobile_interruptable_platform,
+///			|| Some(SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap())
 ///			)
 ///			.await
 ///			.expect("Failed to process events");
@@ -620,11 +640,12 @@ pub async fn process_events_async<
 	S: 'static + Deref<Target = SC> + Send + Sync,
 	SC: for<'b> WriteableScore<'b>,
 	SleepFuture: core::future::Future<Output = bool> + core::marker::Unpin,
-	Sleeper: Fn(Duration) -> SleepFuture
+	Sleeper: Fn(Duration) -> SleepFuture,
+	FetchTime: Fn() -> Option<Duration>,
 >(
 	persister: PS, event_handler: EventHandler, chain_monitor: M, channel_manager: CM,
 	gossip_sync: GossipSync<PGS, RGS, G, UL, L>, peer_manager: PM, logger: L, scorer: Option<S>,
-	sleeper: Sleeper, mobile_interruptable_platform: bool,
+	sleeper: Sleeper, mobile_interruptable_platform: bool, fetch_time: FetchTime,
 ) -> Result<(), lightning::io::Error>
 where
 	UL::Target: 'static + UtxoLookup,
@@ -648,15 +669,18 @@ where
 		let scorer = &scorer;
 		let logger = &logger;
 		let persister = &persister;
+		let fetch_time = &fetch_time;
 		async move {
 			if let Some(network_graph) = network_graph {
 				handle_network_graph_update(network_graph, &event)
 			}
 			if let Some(ref scorer) = scorer {
-				if update_scorer(scorer, &event) {
-					log_trace!(logger, "Persisting scorer after update");
-					if let Err(e) = persister.persist_scorer(&scorer) {
-						log_error!(logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
+				if let Some(duration_since_epoch) = fetch_time() {
+					if update_scorer(scorer, &event, duration_since_epoch) {
+						log_trace!(logger, "Persisting scorer after update");
+						if let Err(e) = persister.persist_scorer(&scorer) {
+							log_error!(logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
+						}
 					}
 				}
 			}
@@ -688,7 +712,7 @@ where
 				task::Poll::Ready(exit) => { should_break = exit; true },
 				task::Poll::Pending => false,
 			}
-		}, mobile_interruptable_platform
+		}, mobile_interruptable_platform, fetch_time,
 	)
 }
 
@@ -810,7 +834,10 @@ impl BackgroundProcessor {
 					handle_network_graph_update(network_graph, &event)
 				}
 				if let Some(ref scorer) = scorer {
-					if update_scorer(scorer, &event) {
+					use std::time::SystemTime;
+					let duration_since_epoch = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH)
+						.expect("Time should be sometime after 1970");
+					if update_scorer(scorer, &event, duration_since_epoch) {
 						log_trace!(logger, "Persisting scorer after update");
 						if let Err(e) = persister.persist_scorer(&scorer) {
 							log_error!(logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
@@ -829,7 +856,12 @@ impl BackgroundProcessor {
 					channel_manager.get_event_or_persistence_needed_future(),
 					chain_monitor.get_update_future()
 				).wait_timeout(Duration::from_millis(100)); },
-				|_| Instant::now(), |time: &Instant, dur| time.elapsed().as_secs() > dur, false
+				|_| Instant::now(), |time: &Instant, dur| time.elapsed().as_secs() > dur, false,
+				|| {
+					use std::time::SystemTime;
+					Some(SystemTime::now().duration_since(SystemTime::UNIX_EPOCH)
+						.expect("Time should be sometime after 1970"))
+				},
 			)
 		});
 		Self { stop_thread: stop_thread_clone, thread_handle: Some(handle) }
@@ -1117,7 +1149,7 @@ mod tests {
 	}
 
 	impl ScoreUpdate for TestScorer {
-		fn payment_path_failed(&mut self, actual_path: &Path, actual_short_channel_id: u64) {
+		fn payment_path_failed(&mut self, actual_path: &Path, actual_short_channel_id: u64, _: Duration) {
 			if let Some(expectations) = &mut self.event_expectations {
 				match expectations.pop_front().unwrap() {
 					TestResult::PaymentFailure { path, short_channel_id } => {
@@ -1137,7 +1169,7 @@ mod tests {
 			}
 		}
 
-		fn payment_path_successful(&mut self, actual_path: &Path) {
+		fn payment_path_successful(&mut self, actual_path: &Path, _: Duration) {
 			if let Some(expectations) = &mut self.event_expectations {
 				match expectations.pop_front().unwrap() {
 					TestResult::PaymentFailure { path, .. } => {
@@ -1156,7 +1188,7 @@ mod tests {
 			}
 		}
 
-		fn probe_failed(&mut self, actual_path: &Path, _: u64) {
+		fn probe_failed(&mut self, actual_path: &Path, _: u64, _: Duration) {
 			if let Some(expectations) = &mut self.event_expectations {
 				match expectations.pop_front().unwrap() {
 					TestResult::PaymentFailure { path, .. } => {
@@ -1174,7 +1206,7 @@ mod tests {
 				}
 			}
 		}
-		fn probe_successful(&mut self, actual_path: &Path) {
+		fn probe_successful(&mut self, actual_path: &Path, _: Duration) {
 			if let Some(expectations) = &mut self.event_expectations {
 				match expectations.pop_front().unwrap() {
 					TestResult::PaymentFailure { path, .. } => {
@@ -1192,6 +1224,7 @@ mod tests {
 				}
 			}
 		}
+		fn time_passed(&mut self, _: Duration) {}
 	}
 
 	#[cfg(c_bindings)]
@@ -1469,7 +1502,7 @@ mod tests {
 					tokio::time::sleep(dur).await;
 					false // Never exit
 				})
-			}, false,
+			}, false, || Some(Duration::ZERO),
 		);
 		match bp_future.await {
 			Ok(_) => panic!("Expected error persisting manager"),
@@ -1600,7 +1633,7 @@ mod tests {
 
 		loop {
 			let log_entries = nodes[0].logger.lines.lock().unwrap();
-			let expected_log = "Persisting scorer".to_string();
+			let expected_log = "Calling time_passed and persisting scorer".to_string();
 			if log_entries.get(&("lightning_background_processor", expected_log)).is_some() {
 				break
 			}
@@ -1699,7 +1732,7 @@ mod tests {
 						_ = exit_receiver.changed() => true,
 					}
 				})
-			}, false,
+			}, false, || Some(Duration::from_secs(1696300000)),
 		);
 
 		let t1 = tokio::spawn(bp_future);
@@ -1874,7 +1907,7 @@ mod tests {
 						_ = exit_receiver.changed() => true,
 					}
 				})
-			}, false,
+			}, false, || Some(Duration::ZERO),
 		);
 		let t1 = tokio::spawn(bp_future);
 		let t2 = tokio::spawn(async move {
diff --git a/lightning/Cargo.toml b/lightning/Cargo.toml
@@ -31,7 +31,7 @@ unsafe_revoked_tx_signing = []
 # Override signing to not include randomness when generating signatures for test vectors.
 _test_vectors = []
 
-no-std = ["hashbrown", "bitcoin/no-std", "core2/alloc"]
+no-std = ["hashbrown", "bitcoin/no-std", "core2/alloc", "libm"]
 std = ["bitcoin/std"]
 
 # Generates low-r bitcoin signatures, which saves 1 byte in 50% of the cases
@@ -48,6 +48,7 @@ regex = { version = "1.5.6", optional = true }
 backtrace = { version = "0.3", optional = true }
 
 core2 = { version = "0.3.0", optional = true, default-features = false }
+libm = { version = "0.2", optional = true, default-features = false }
 
 [dev-dependencies]
 regex = "1.5.6"
diff --git a/lightning/src/lib.rs b/lightning/src/lib.rs
@@ -69,6 +69,7 @@ extern crate hex;
 #[cfg(any(test, feature = "_test_utils"))] extern crate regex;
 
 #[cfg(not(feature = "std"))] extern crate core2;
+#[cfg(not(feature = "std"))] extern crate libm;
 
 #[cfg(ldk_bench)] extern crate criterion;
 
diff --git a/lightning/src/routing/router.rs b/lightning/src/routing/router.rs
@@ -8160,6 +8160,7 @@ mod tests {
 pub(crate) mod bench_utils {
 	use super::*;
 	use std::fs::File;
+	use std::time::Duration;
 
 	use bitcoin::hashes::Hash;
 	use bitcoin::secp256k1::{PublicKey, Secp256k1, SecretKey};
@@ -8308,10 +8309,10 @@ pub(crate) mod bench_utils {
 						if let Ok(route) = route_res {
 							for path in route.paths {
 								if seed & 0x80 == 0 {
-									scorer.payment_path_successful(&path);
+									scorer.payment_path_successful(&path, Duration::ZERO);
 								} else {
 									let short_channel_id = path.hops[path.hops.len() / 2].short_channel_id;
-									scorer.payment_path_failed(&path, short_channel_id);
+									scorer.payment_path_failed(&path, short_channel_id, Duration::ZERO);
 								}
 								seed = seed.overflowing_mul(6364136223846793005).0.overflowing_add(1).0;
 							}
diff --git a/lightning/src/routing/scoring.rs b/lightning/src/routing/scoring.rs
diff --git a/lightning/src/util/test_utils.rs b/lightning/src/util/test_utils.rs
