Create probing service

Create new background task for probing, add its initialization to the
builder.

Add new trait ProobingStrategy with two default implementations.

Total amount of sats in current non-finished probes is tracked, it is
changed on ProbeSucceded and ProbeFailed events.

Author: randomlogin

src/builder.rs

@@ -9,8 +9,9 @@ use std::collections::HashMap;
 use std::convert::TryInto;
 use std::default::Default;
 use std::path::PathBuf;
+use std::sync::atomic::AtomicU64;
 use std::sync::{Arc, Mutex, Once, RwLock};
-use std::time::SystemTime;
+use std::time::{Duration, SystemTime};
 use std::{fmt, fs};
 
 use bdk_wallet::template::Bip84;
@@ -47,6 +48,7 @@ use crate::config::{
 	default_user_config, may_announce_channel, AnnounceError, AsyncPaymentsRole,
 	BitcoindRestClientConfig, Config, ElectrumSyncConfig, EsploraSyncConfig,
 	DEFAULT_ESPLORA_SERVER_URL, DEFAULT_LOG_FILENAME, DEFAULT_LOG_LEVEL,
+	DEFAULT_MAX_PROBE_LOCKED_MSAT, DEFAULT_PROBING_INTERVAL_SECS, MIN_PROBE_AMOUNT_MSAT,
 };
 use crate::connection::ConnectionManager;
 use crate::entropy::NodeEntropy;
@@ -72,6 +74,8 @@ use crate::logger::{log_error, LdkLogger, LogLevel, LogWriter, Logger};
 use crate::message_handler::NodeCustomMessageHandler;
 use crate::payment::asynchronous::om_mailbox::OnionMessageMailbox;
 use crate::peer_store::PeerStore;
+use crate::probing;
+use crate::probing::ProbingStrategy;
 use crate::runtime::{Runtime, RuntimeSpawner};
 use crate::tx_broadcaster::TransactionBroadcaster;
 use crate::types::{
@@ -150,6 +154,37 @@ impl std::fmt::Debug for LogWriterConfig {
 	}
 }
 
+enum ProbingStrategyKind {
+	HighDegree { top_n: usize },
+	Random { max_hops: usize },
+	Custom(Arc<dyn probing::ProbingStrategy>),
+}
+
+struct ProbingStrategyConfig {
+	kind: ProbingStrategyKind,
+	interval: Duration,
+	max_locked_msat: u64,
+}
+
+impl fmt::Debug for ProbingStrategyConfig {
+	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+		let kind_str = match &self.kind {
+			ProbingStrategyKind::HighDegree { top_n } => {
+				format!("HighDegree {{ top_n: {} }}", top_n)
+			},
+			ProbingStrategyKind::Random { max_hops } => {
+				format!("Random {{ max_hops: {} }}", max_hops)
+			},
+			ProbingStrategyKind::Custom(_) => "Custom(<probing strategy>)".to_string(),
+		};
+		f.debug_struct("ProbingStrategyConfig")
+			.field("kind", &kind_str)
+			.field("interval", &self.interval)
+			.field("max_locked_msat", &self.max_locked_msat)
+			.finish()
+	}
+}
+
 /// An error encountered during building a [`Node`].
 ///
 /// [`Node`]: crate::Node
@@ -245,6 +280,7 @@ pub struct NodeBuilder {
 	runtime_handle: Option<tokio::runtime::Handle>,
 	pathfinding_scores_sync_config: Option<PathfindingScoresSyncConfig>,
 	recovery_mode: bool,
+	probing_strategy: Option<ProbingStrategyConfig>,
 }
 
 impl NodeBuilder {
@@ -273,6 +309,7 @@ impl NodeBuilder {
 			async_payments_role: None,
 			pathfinding_scores_sync_config,
 			recovery_mode,
+			probing_strategy: None,
 		}
 	}
 
@@ -557,6 +594,64 @@ impl NodeBuilder {
 		self
 	}
 
+	/// Configures background probing toward the highest-degree nodes in the network graph.
+	///
+	/// `top_n` controls how many of the most-connected nodes are cycled through.
+	pub fn set_high_degree_probing_strategy(&mut self, top_n: usize) -> &mut Self {
+		let kind = ProbingStrategyKind::HighDegree { top_n };
+		self.probing_strategy = Some(self.make_probing_config(kind));
+		self
+	}
+
+	/// Configures background probing via random graph walks of up to `max_hops` hops.
+	pub fn set_random_probing_strategy(&mut self, max_hops: usize) -> &mut Self {
+		let kind = ProbingStrategyKind::Random { max_hops };
+		self.probing_strategy = Some(self.make_probing_config(kind));
+		self
+	}
+
+	/// Configures a custom probing strategy for background channel probing.
+	///
+	/// When set, the node will periodically call [`ProbingStrategy::next_probe`] and dispatch the
+	/// returned probe via the channel manager.
+	pub fn set_probing_strategy(
+		&mut self, strategy: Arc<dyn probing::ProbingStrategy>,
+	) -> &mut Self {
+		let kind = ProbingStrategyKind::Custom(strategy);
+		self.probing_strategy = Some(self.make_probing_config(kind));
+		self
+	}
+
+	/// Overrides the interval between probe attempts. Only has effect if a probing strategy is set.
+	pub fn set_probing_interval(&mut self, interval: Duration) -> &mut Self {
+		if let Some(cfg) = &mut self.probing_strategy {
+			cfg.interval = interval;
+		}
+		self
+	}
+
+	/// Overrides the maximum millisatoshis that may be locked in in-flight probes at any time.
+	/// Only has effect if a probing strategy is set.
+	pub fn set_max_probe_locked_msat(&mut self, max_msat: u64) -> &mut Self {
+		if let Some(cfg) = &mut self.probing_strategy {
+			cfg.max_locked_msat = max_msat;
+		}
+		self
+	}
+
+	fn make_probing_config(&self, kind: ProbingStrategyKind) -> ProbingStrategyConfig {
+		let existing = self.probing_strategy.as_ref();
+		ProbingStrategyConfig {
+			kind,
+			interval: existing
+				.map(|c| c.interval)
+				.unwrap_or(Duration::from_secs(DEFAULT_PROBING_INTERVAL_SECS)),
+			max_locked_msat: existing
+				.map(|c| c.max_locked_msat)
+				.unwrap_or(DEFAULT_MAX_PROBE_LOCKED_MSAT),
+		}
+	}
+
 	/// Builds a [`Node`] instance with a [`SqliteStore`] backend and according to the options
 	/// previously configured.
 	pub fn build(&self, node_entropy: NodeEntropy) -> Result<Node, BuildError> {
@@ -697,6 +792,7 @@ impl NodeBuilder {
 			runtime,
 			logger,
 			Arc::new(DynStoreWrapper(kv_store)),
+			self.probing_strategy.as_ref(),
 		)
 	}
 }
@@ -942,6 +1038,11 @@ impl ArcedNodeBuilder {
 		self.inner.write().unwrap().set_wallet_recovery_mode();
 	}
 
+	/// Configures a probing strategy for background channel probing.
+	pub fn set_probing_strategy(&self, strategy: Arc<dyn probing::ProbingStrategy>) {
+		self.inner.write().unwrap().set_probing_strategy(strategy);
+	}
+
 	/// Builds a [`Node`] instance with a [`SqliteStore`] backend and according to the options
 	/// previously configured.
 	pub fn build(&self, node_entropy: Arc<NodeEntropy>) -> Result<Arc<Node>, BuildError> {
@@ -1058,6 +1159,7 @@ fn build_with_store_internal(
 	pathfinding_scores_sync_config: Option<&PathfindingScoresSyncConfig>,
 	async_payments_role: Option<AsyncPaymentsRole>, recovery_mode: bool, seed_bytes: [u8; 64],
 	runtime: Arc<Runtime>, logger: Arc<Logger>, kv_store: Arc<DynStore>,
+	probing_config: Option<&ProbingStrategyConfig>,
 ) -> Result<Node, BuildError> {
 	optionally_install_rustls_cryptoprovider();
 
@@ -1783,6 +1885,36 @@ fn build_with_store_internal(
 		_leak_checker.0.push(Arc::downgrade(&wallet) as Weak<dyn Any + Send + Sync>);
 	}
 
+	let prober = probing_config.map(|config| {
+		let strategy: Arc<dyn probing::ProbingStrategy> = match &config.kind {
+			ProbingStrategyKind::HighDegree { top_n } => {
+				Arc::new(probing::HighDegreeStrategy::new(
+					network_graph.clone(),
+					*top_n,
+					MIN_PROBE_AMOUNT_MSAT,
+					config.max_locked_msat,
+				))
+			},
+			ProbingStrategyKind::Random { max_hops } => Arc::new(probing::RandomStrategy::new(
+				network_graph.clone(),
+				channel_manager.clone(),
+				*max_hops,
+				MIN_PROBE_AMOUNT_MSAT,
+				config.max_locked_msat,
+			)),
+			ProbingStrategyKind::Custom(s) => s.clone(),
+		};
+		Arc::new(probing::Prober {
+			channel_manager: channel_manager.clone(),
+			logger: logger.clone(),
+			strategy,
+			interval: config.interval,
+			liquidity_limit_multiplier: None,
+			max_locked_msat: config.max_locked_msat,
+			locked_msat: Arc::new(AtomicU64::new(0)),
+		})
+	});
+
 	Ok(Node {
 		runtime,
 		stop_sender,
@@ -1815,6 +1947,7 @@ fn build_with_store_internal(
 		om_mailbox,
 		async_payments_role,
 		hrn_resolver,
+		prober,
 		#[cfg(cycle_tests)]
 		_leak_checker,
 	})

src/config.rs

@@ -27,6 +27,9 @@ const DEFAULT_BDK_WALLET_SYNC_INTERVAL_SECS: u64 = 80;
 const DEFAULT_LDK_WALLET_SYNC_INTERVAL_SECS: u64 = 30;
 const DEFAULT_FEE_RATE_CACHE_UPDATE_INTERVAL_SECS: u64 = 60 * 10;
 const DEFAULT_PROBING_LIQUIDITY_LIMIT_MULTIPLIER: u64 = 3;
+pub(crate) const DEFAULT_PROBING_INTERVAL_SECS: u64 = 10;
+pub(crate) const DEFAULT_MAX_PROBE_LOCKED_MSAT: u64 = 100_000_000; // 100k sats
+pub(crate) const MIN_PROBE_AMOUNT_MSAT: u64 = 1_000_000; // 1k sats
 const DEFAULT_ANCHOR_PER_CHANNEL_RESERVE_SATS: u64 = 25_000;
 
 // The default timeout after which we abort a wallet syncing operation.

src/event.rs

@@ -9,6 +9,7 @@ use core::future::Future;
 use core::task::{Poll, Waker};
 use std::collections::VecDeque;
 use std::ops::Deref;
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, Mutex};
 
 use bitcoin::blockdata::locktime::absolute::LockTime;
@@ -494,6 +495,7 @@ where
 	static_invoice_store: Option<StaticInvoiceStore>,
 	onion_messenger: Arc<OnionMessenger>,
 	om_mailbox: Option<Arc<OnionMessageMailbox>>,
+	probe_locked_msat: Option<Arc<AtomicU64>>,
 }
 
 impl<L: Deref + Clone + Sync + Send + 'static> EventHandler<L>
@@ -509,7 +511,7 @@ where
 		payment_store: Arc<PaymentStore>, peer_store: Arc<PeerStore<L>>,
 		static_invoice_store: Option<StaticInvoiceStore>, onion_messenger: Arc<OnionMessenger>,
 		om_mailbox: Option<Arc<OnionMessageMailbox>>, runtime: Arc<Runtime>, logger: L,
-		config: Arc<Config>,
+		config: Arc<Config>, probe_locked_msat: Option<Arc<AtomicU64>>,
 	) -> Self {
 		Self {
 			event_queue,
@@ -528,6 +530,7 @@ where
 			static_invoice_store,
 			onion_messenger,
 			om_mailbox,
+			probe_locked_msat,
 		}
 	}
 
@@ -1111,8 +1114,22 @@ where
 
 			LdkEvent::PaymentPathSuccessful { .. } => {},
 			LdkEvent::PaymentPathFailed { .. } => {},
-			LdkEvent::ProbeSuccessful { .. } => {},
-			LdkEvent::ProbeFailed { .. } => {},
+			LdkEvent::ProbeSuccessful { path, .. } => {
+				if let Some(counter) = &self.probe_locked_msat {
+					let amount = path.hops.last().map_or(0, |h| h.fee_msat);
+					let _ = counter.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| {
+						Some(v.saturating_sub(amount))
+					});
+				}
+			},
+			LdkEvent::ProbeFailed { path, .. } => {
+				if let Some(counter) = &self.probe_locked_msat {
+					let amount = path.hops.last().map_or(0, |h| h.fee_msat);
+					let _ = counter.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| {
+						Some(v.saturating_sub(amount))
+					});
+				}
+			},
 			LdkEvent::HTLCHandlingFailed { failure_type, .. } => {
 				if let Some(liquidity_source) = self.liquidity_source.as_ref() {
 					liquidity_source.handle_htlc_handling_failed(failure_type).await;
@@ -1356,7 +1373,6 @@ where
 						);
 					}
 				}
-
 				if let Some(liquidity_source) = self.liquidity_source.as_ref() {
 					let skimmed_fee_msat = skimmed_fee_msat.unwrap_or(0);
 					liquidity_source

src/lib.rs

@@ -100,6 +100,7 @@ pub mod logger;
 mod message_handler;
 pub mod payment;
 mod peer_store;
+mod probing;
 mod runtime;
 mod scoring;
 mod tx_broadcaster;
@@ -157,6 +158,7 @@ use payment::{
 	UnifiedPayment,
 };
 use peer_store::{PeerInfo, PeerStore};
+pub use probing::{HighDegreeStrategy, Probe, ProbingStrategy, RandomStrategy};
 use rand::Rng;
 use runtime::Runtime;
 use types::{
@@ -227,6 +229,7 @@ pub struct Node {
 	om_mailbox: Option<Arc<OnionMessageMailbox>>,
 	async_payments_role: Option<AsyncPaymentsRole>,
 	hrn_resolver: Arc<HRNResolver>,
+	prober: Option<Arc<probing::Prober>>,
 	#[cfg(cycle_tests)]
 	_leak_checker: LeakChecker,
 }
@@ -563,6 +566,7 @@ impl Node {
 			None
 		};
 
+		let probe_locked_msat = self.prober.as_ref().map(|p| Arc::clone(&p.locked_msat));
 		let event_handler = Arc::new(EventHandler::new(
 			Arc::clone(&self.event_queue),
 			Arc::clone(&self.wallet),
@@ -580,8 +584,16 @@ impl Node {
 			Arc::clone(&self.runtime),
 			Arc::clone(&self.logger),
 			Arc::clone(&self.config),
+			probe_locked_msat,
 		));
 
+		if let Some(prober) = self.prober.clone() {
+			let stop_rx = self.stop_sender.subscribe();
+			self.runtime.spawn_cancellable_background_task(async move {
+				prober.run(stop_rx).await;
+			});
+		}
+
 		// Setup background processing
 		let background_persister = Arc::clone(&self.kv_store);
 		let background_event_handler = Arc::clone(&event_handler);
@@ -1032,6 +1044,17 @@ impl Node {
 		))
 	}
 
+	/// Returns the total millisatoshis currently locked in in-flight probes, or `None` if no
+	/// probing strategy is configured.
+	pub fn probe_locked_msat(&self) -> Option<u64> {
+		self.prober.as_ref().map(|p| p.locked_msat.load(std::sync::atomic::Ordering::Relaxed))
+	}
+
+    /// Gives access to the scorer; needed to valuate the probing tests.
+    #[cfg(test)]
+    pub fn scorer(&self) -> &Arc<Mutex<Scorer>> { &self.scorer }
+
+
 	/// Retrieve a list of known channels.
 	pub fn list_channels(&self) -> Vec<ChannelDetails> {
 		self.channel_manager.list_channels().into_iter().map(|c| c.into()).collect()