Add probing service

Introduce a background probing service that periodically dispatches
probes to improve the scorer's liquidity estimates. Includes two
built-in strategies.

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,8 @@ use crate::config::{
 	default_user_config, may_announce_channel, AnnounceError, AsyncPaymentsRole,
 	BitcoindRestClientConfig, Config, ElectrumSyncConfig, EsploraSyncConfig, TorConfig,
 	DEFAULT_ESPLORA_SERVER_URL, DEFAULT_LOG_FILENAME, DEFAULT_LOG_LEVEL,
+	DEFAULT_MAX_PROBE_AMOUNT_MSAT, DEFAULT_MAX_PROBE_LOCKED_MSAT, DEFAULT_PROBING_INTERVAL_SECS,
+	MIN_PROBE_AMOUNT_MSAT,
 };
 use crate::connection::ConnectionManager;
 use crate::entropy::NodeEntropy;
@@ -73,6 +76,7 @@ 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::runtime::{Runtime, RuntimeSpawner};
 use crate::tx_broadcaster::TransactionBroadcaster;
 use crate::types::{
@@ -151,6 +155,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
@@ -281,6 +316,8 @@ pub struct NodeBuilder {
 	runtime_handle: Option<tokio::runtime::Handle>,
 	pathfinding_scores_sync_config: Option<PathfindingScoresSyncConfig>,
 	recovery_mode: bool,
+	probing_strategy: Option<ProbingStrategyConfig>,
+	probing_diversity_penalty_msat: Option<u64>,
 }
 
 impl NodeBuilder {
@@ -299,16 +336,21 @@ impl NodeBuilder {
 		let runtime_handle = None;
 		let pathfinding_scores_sync_config = None;
 		let recovery_mode = false;
+		let async_payments_role = None;
+		let probing_strategy = None;
+		let probing_diversity_penalty_msat = None;
 		Self {
 			config,
 			chain_data_source_config,
 			gossip_source_config,
 			liquidity_source_config,
 			log_writer_config,
 			runtime_handle,
-			async_payments_role: None,
+			async_payments_role,
 			pathfinding_scores_sync_config,
 			recovery_mode,
+			probing_strategy,
+			probing_diversity_penalty_msat,
 		}
 	}
 
@@ -614,6 +656,80 @@ 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_custom_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
+	}
+
+	/// Sets the probing diversity penalty applied by the probabilistic scorer.
+	///
+	/// When set, the scorer will penalize channels that have been recently probed,
+	/// encouraging path diversity during background probing. The penalty decays
+	/// quadratically over 24 hours.
+	///
+	/// This is only useful for probing strategies that route through the scorer
+	/// (e.g., [`HighDegreeStrategy`]). Strategies that build paths manually
+	/// (e.g., [`RandomStrategy`]) bypass the scorer entirely.
+	///
+	/// If unset, LDK's default of `0` (no penalty) is used.
+	pub fn set_probing_diversity_penalty_msat(&mut self, penalty_msat: u64) -> &mut Self {
+		self.probing_diversity_penalty_msat = Some(penalty_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> {
@@ -791,6 +907,8 @@ impl NodeBuilder {
 			runtime,
 			logger,
 			Arc::new(DynStoreWrapper(kv_store)),
+			self.probing_strategy.as_ref(),
+			self.probing_diversity_penalty_msat,
 		)
 	}
 }
@@ -1081,6 +1199,11 @@ impl ArcedNodeBuilder {
 		self.inner.write().unwrap().set_wallet_recovery_mode();
 	}
 
+	/// Configures a probing strategy for background channel probing.
+	pub fn set_custom_probing_strategy(&self, strategy: Arc<dyn probing::ProbingStrategy>) {
+		self.inner.write().unwrap().set_custom_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> {
@@ -1226,6 +1349,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>, probing_diversity_penalty_msat: Option<u64>,
 ) -> Result<Node, BuildError> {
 	optionally_install_rustls_cryptoprovider();
 
@@ -1626,7 +1750,10 @@ fn build_with_store_internal(
 		},
 	}
 
-	let scoring_fee_params = ProbabilisticScoringFeeParameters::default();
+	let mut scoring_fee_params = ProbabilisticScoringFeeParameters::default();
+	if let Some(penalty) = probing_diversity_penalty_msat {
+		scoring_fee_params.probing_diversity_penalty_msat = penalty;
+	}
 	let router = Arc::new(DefaultRouter::new(
 		Arc::clone(&network_graph),
 		Arc::clone(&logger),
@@ -1965,6 +2092,36 @@ fn build_with_store_internal(
 		_leak_checker.0.push(Arc::downgrade(&wallet) as Weak<dyn Any + Send + Sync>);
 	}
 
+	let prober = probing_config.map(|probing_cfg| {
+		let strategy: Arc<dyn probing::ProbingStrategy> = match &probing_cfg.kind {
+			ProbingStrategyKind::HighDegree { top_n } => {
+				Arc::new(probing::HighDegreeStrategy::new(
+					network_graph.clone(),
+					*top_n,
+					MIN_PROBE_AMOUNT_MSAT,
+					DEFAULT_MAX_PROBE_AMOUNT_MSAT,
+				))
+			},
+			ProbingStrategyKind::Random { max_hops } => Arc::new(probing::RandomStrategy::new(
+				network_graph.clone(),
+				channel_manager.clone(),
+				*max_hops,
+				MIN_PROBE_AMOUNT_MSAT,
+				DEFAULT_MAX_PROBE_AMOUNT_MSAT,
+			)),
+			ProbingStrategyKind::Custom(s) => s.clone(),
+		};
+		Arc::new(probing::Prober {
+			channel_manager: channel_manager.clone(),
+			logger: logger.clone(),
+			strategy,
+			interval: probing_cfg.interval,
+			liquidity_limit_multiplier: Some(config.probing_liquidity_limit_multiplier),
+			max_locked_msat: probing_cfg.max_locked_msat,
+			locked_msat: Arc::new(AtomicU64::new(0)),
+		})
+	});
+
 	Ok(Node {
 		runtime,
 		stop_sender,
@@ -1998,6 +2155,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,10 @@ 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
+pub(crate) const DEFAULT_MAX_PROBE_AMOUNT_MSAT: u64 = 10_000_000; // 10k 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;
@@ -515,6 +516,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>
@@ -531,6 +533,7 @@ where
 		keys_manager: Arc<KeysManager>, static_invoice_store: Option<StaticInvoiceStore>,
 		onion_messenger: Arc<OnionMessenger>, om_mailbox: Option<Arc<OnionMessageMailbox>>,
 		runtime: Arc<Runtime>, logger: L, config: Arc<Config>,
+		probe_locked_msat: Option<Arc<AtomicU64>>,
 	) -> Self {
 		Self {
 			event_queue,
@@ -550,6 +553,7 @@ where
 			static_invoice_store,
 			onion_messenger,
 			om_mailbox,
+			probe_locked_msat,
 		}
 	}
 
@@ -1135,8 +1139,22 @@ where
 
 			LdkEvent::PaymentPathSuccessful { .. } => {},
 			LdkEvent::PaymentPathFailed { .. } => {},
-			LdkEvent::ProbeSuccessful { .. } => {},
-			LdkEvent::ProbeFailed { .. } => {},
+			LdkEvent::ProbeSuccessful { path, .. } => {
+				if let Some(counter) = &self.probe_locked_msat {
+					let amount: u64 = path.hops.iter().map(|h| h.fee_msat).sum();
+					let _ = counter.fetch_update(Ordering::AcqRel, Ordering::Acquire, |v| {
+						Some(v.saturating_sub(amount))
+					});
+				}
+			},
+			LdkEvent::ProbeFailed { path, .. } => {
+				if let Some(counter) = &self.probe_locked_msat {
+					let amount: u64 = path.hops.iter().map(|h| h.fee_msat).sum();
+					let _ = counter.fetch_update(Ordering::AcqRel, Ordering::Acquire, |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;