Add probing service

Cargo.toml

@@ -136,6 +136,25 @@ check-cfg = [
 name = "payments"
 harness = false
 
+# DROPME BUMP: redirect rust-lightning to a fork carrying a cherry-pick of
+# upstream LDK#4603 ("Expose probe status in recent payments"), which adds
+# `is_probe` to `RecentPaymentDetails::Pending`/`Abandoned`. Required by the
+# probing service. Drop this block once LDK#4603 is merged into LDK and
+# ldk-node `main` bumps past that commit.
+[patch."https://github.com/lightningdevkit/rust-lightning"]
+lightning = { git = "https://github.com/randomlogin/rust-lightning", rev = "992754573475740b4e406a45c5780b3d86847a93" }
+lightning-types = { git = "https://github.com/randomlogin/rust-lightning", rev = "992754573475740b4e406a45c5780b3d86847a93" }
+lightning-invoice = { git = "https://github.com/randomlogin/rust-lightning", rev = "992754573475740b4e406a45c5780b3d86847a93" }
+lightning-net-tokio = { git = "https://github.com/randomlogin/rust-lightning", rev = "992754573475740b4e406a45c5780b3d86847a93" }
+lightning-persister = { git = "https://github.com/randomlogin/rust-lightning", rev = "992754573475740b4e406a45c5780b3d86847a93" }
+lightning-background-processor = { git = "https://github.com/randomlogin/rust-lightning", rev = "992754573475740b4e406a45c5780b3d86847a93" }
+lightning-rapid-gossip-sync = { git = "https://github.com/randomlogin/rust-lightning", rev = "992754573475740b4e406a45c5780b3d86847a93" }
+lightning-block-sync = { git = "https://github.com/randomlogin/rust-lightning", rev = "992754573475740b4e406a45c5780b3d86847a93" }
+lightning-transaction-sync = { git = "https://github.com/randomlogin/rust-lightning", rev = "992754573475740b4e406a45c5780b3d86847a93" }
+lightning-liquidity = { git = "https://github.com/randomlogin/rust-lightning", rev = "992754573475740b4e406a45c5780b3d86847a93" }
+lightning-macros = { git = "https://github.com/randomlogin/rust-lightning", rev = "992754573475740b4e406a45c5780b3d86847a93" }
+lightning-dns-resolver = { git = "https://github.com/randomlogin/rust-lightning", rev = "992754573475740b4e406a45c5780b3d86847a93" }
+
 #[patch.crates-io]
 #lightning = { path = "../rust-lightning/lightning" }
 #lightning-types = { path = "../rust-lightning/lightning-types" }

bindings/ldk_node.udl

@@ -13,6 +13,10 @@ typedef dictionary TorConfig;
 
 typedef interface NodeEntropy;
 
+typedef interface ProbingConfig;
+
+typedef interface ProbingConfigBuilder;
+
 typedef enum WordCount;
 
 [Remote]
@@ -61,6 +65,7 @@ interface Builder {
 	[Throws=BuildError]
 	void set_async_payments_role(AsyncPaymentsRole? role);
 	void set_wallet_recovery_mode();
+	void set_probing_config(ProbingConfig config);
 	[Throws=BuildError]
 	Node build(NodeEntropy node_entropy);
 	[Throws=BuildError]

src/builder.rs

@@ -51,6 +51,7 @@ use crate::config::{
 	default_user_config, may_announce_channel, AnnounceError, AsyncPaymentsRole,
 	BitcoindRestClientConfig, Config, ElectrumSyncConfig, EsploraSyncConfig, HRNResolverConfig,
 	TorConfig, DEFAULT_ESPLORA_SERVER_URL, DEFAULT_LOG_FILENAME, DEFAULT_LOG_LEVEL,
+	DEFAULT_MAX_PROBE_AMOUNT_MSAT, DEFAULT_MIN_PROBE_AMOUNT_MSAT,
 };
 use crate::connection::ConnectionManager;
 use crate::entropy::NodeEntropy;
@@ -76,6 +77,9 @@ 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::{
+	HighDegreeStrategy, Prober, ProbingConfig, ProbingStrategy, ProbingStrategyKind, RandomStrategy,
+};
 use crate::runtime::{Runtime, RuntimeSpawner};
 use crate::tx_broadcaster::TransactionBroadcaster;
 use crate::types::{
@@ -292,6 +296,7 @@ pub struct NodeBuilder {
 	runtime_handle: Option<tokio::runtime::Handle>,
 	pathfinding_scores_sync_config: Option<PathfindingScoresSyncConfig>,
 	recovery_mode: bool,
+	probing_config: Option<ProbingConfig>,
 }
 
 impl NodeBuilder {
@@ -310,16 +315,19 @@ impl NodeBuilder {
 		let runtime_handle = None;
 		let pathfinding_scores_sync_config = None;
 		let recovery_mode = false;
+		let async_payments_role = None;
+		let probing_config = 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_config,
 		}
 	}
 
@@ -625,6 +633,31 @@ impl NodeBuilder {
 		self
 	}
 
+	/// Configures background probing.
+	///
+	/// Use [`ProbingConfigBuilder`] to build the configuration:
+	/// ```no_run
+	/// # #[cfg(not(feature = "uniffi"))]
+	/// # {
+	/// use std::time::Duration;
+	/// use ldk_node::Builder;
+	/// use ldk_node::probing::ProbingConfigBuilder;
+	///
+	/// let mut builder = Builder::new();
+	/// builder.set_probing_config(
+	///     ProbingConfigBuilder::high_degree(100)
+	///         .interval(Duration::from_secs(30))
+	///         .build()
+	/// );
+	/// # }
+	/// ```
+	///
+	/// [`ProbingConfigBuilder`]: crate::probing::ProbingConfigBuilder
+	pub fn set_probing_config(&mut self, config: ProbingConfig) -> &mut Self {
+		self.probing_config = Some(config);
+		self
+	}
+
 	/// 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> {
@@ -809,6 +842,7 @@ impl NodeBuilder {
 			self.gossip_source_config.as_ref(),
 			self.liquidity_source_config.as_ref(),
 			self.pathfinding_scores_sync_config.as_ref(),
+			self.probing_config.as_ref(),
 			self.async_payments_role,
 			self.recovery_mode,
 			seed_bytes,
@@ -1109,6 +1143,15 @@ impl ArcedNodeBuilder {
 		self.inner.write().expect("lock").set_wallet_recovery_mode();
 	}
 
+	/// Configures background probing.
+	///
+	/// Use [`ProbingConfigBuilder`] to build the configuration.
+	///
+	/// [`ProbingConfigBuilder`]: crate::probing::ProbingConfigBuilder
+	pub fn set_probing_config(&self, config: Arc<ProbingConfig>) {
+		self.inner.write().expect("lock").set_probing_config((*config).clone());
+	}
+
 	/// 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> {
@@ -1252,8 +1295,9 @@ fn build_with_store_internal(
 	gossip_source_config: Option<&GossipSourceConfig>,
 	liquidity_source_config: Option<&LiquiditySourceConfig>,
 	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<&ProbingConfig>, async_payments_role: Option<AsyncPaymentsRole>,
+	recovery_mode: bool, seed_bytes: [u8; 64], runtime: Arc<Runtime>, logger: Arc<Logger>,
+	kv_store: Arc<DynStore>,
 ) -> Result<Node, BuildError> {
 	optionally_install_rustls_cryptoprovider();
 
@@ -1661,7 +1705,10 @@ fn build_with_store_internal(
 		},
 	}
 
-	let scoring_fee_params = ProbabilisticScoringFeeParameters::default();
+	let mut scoring_fee_params = ProbabilisticScoringFeeParameters::default();
+	if let Some(penalty) = probing_config.and_then(|c| c.diversity_penalty_msat) {
+		scoring_fee_params.probing_diversity_penalty_msat = penalty;
+	}
 	let router = Arc::new(DefaultRouter::new(
 		Arc::clone(&network_graph),
 		Arc::clone(&logger),
@@ -2036,6 +2083,38 @@ 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 ProbingStrategy> = match &probing_cfg.kind {
+			ProbingStrategyKind::HighDegree { top_node_count } => {
+				Arc::new(HighDegreeStrategy::new(
+					Arc::clone(&network_graph),
+					Arc::clone(&channel_manager),
+					Arc::clone(&router),
+					*top_node_count,
+					DEFAULT_MIN_PROBE_AMOUNT_MSAT,
+					DEFAULT_MAX_PROBE_AMOUNT_MSAT,
+					probing_cfg.cooldown,
+					config.probing_liquidity_limit_multiplier,
+				))
+			},
+			ProbingStrategyKind::Random { max_hops } => Arc::new(RandomStrategy::new(
+				Arc::clone(&network_graph),
+				Arc::clone(&channel_manager),
+				*max_hops,
+				DEFAULT_MIN_PROBE_AMOUNT_MSAT,
+				DEFAULT_MAX_PROBE_AMOUNT_MSAT,
+			)),
+			ProbingStrategyKind::Custom(s) => Arc::clone(s),
+		};
+		Arc::new(Prober {
+			channel_manager: Arc::clone(&channel_manager),
+			logger: Arc::clone(&logger),
+			strategy,
+			interval: probing_cfg.interval,
+			max_locked_msat: probing_cfg.max_locked_msat,
+		})
+	});
+
 	Ok(Node {
 		runtime,
 		stop_sender,
@@ -2069,6 +2148,7 @@ fn build_with_store_internal(
 		om_mailbox,
 		async_payments_role,
 		hrn_resolver,
+		prober,
 		#[cfg(cycle_tests)]
 		_leak_checker,
 	})

src/config.rs

@@ -28,6 +28,12 @@ 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 MIN_PROBING_INTERVAL: Duration = Duration::from_millis(100);
+pub(crate) const DEFAULT_PROBED_NODE_COOLDOWN_SECS: u64 = 60 * 60; // 1 hour
+pub(crate) const DEFAULT_MAX_PROBE_LOCKED_MSAT: u64 = 100_000_000; // 100k sats
+pub(crate) const DEFAULT_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

@@ -50,6 +50,7 @@ use crate::payment::asynchronous::static_invoice_store::StaticInvoiceStore;
 use crate::payment::store::{
 	PaymentDetails, PaymentDetailsUpdate, PaymentDirection, PaymentKind, PaymentStatus,
 };
+use crate::probing::Prober;
 use crate::runtime::Runtime;
 use crate::types::{
 	CustomTlvRecord, DynStore, KeysManager, OnionMessenger, PaymentStore, Sweeper, Wallet,
@@ -537,12 +538,13 @@ where
 	payment_store: Arc<PaymentStore>,
 	peer_store: Arc<PeerStore<L>>,
 	keys_manager: Arc<KeysManager>,
-	runtime: Arc<Runtime>,
-	logger: L,
-	config: Arc<Config>,
 	static_invoice_store: Option<StaticInvoiceStore>,
 	onion_messenger: Arc<OnionMessenger>,
 	om_mailbox: Option<Arc<OnionMessageMailbox>>,
+	prober: Option<Arc<Prober>>,
+	runtime: Arc<Runtime>,
+	logger: L,
+	config: Arc<Config>,
 }
 
 impl<L: Deref + Clone + Sync + Send + 'static> EventHandler<L>
@@ -558,7 +560,7 @@ where
 		payment_store: Arc<PaymentStore>, peer_store: Arc<PeerStore<L>>,
 		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>,
+		prober: Option<Arc<Prober>>, runtime: Arc<Runtime>, logger: L, config: Arc<Config>,
 	) -> Self {
 		Self {
 			event_queue,
@@ -572,12 +574,13 @@ where
 			payment_store,
 			peer_store,
 			keys_manager,
-			logger,
-			runtime,
-			config,
 			static_invoice_store,
 			onion_messenger,
 			om_mailbox,
+			prober,
+			runtime,
+			logger,
+			config,
 		}
 	}
 
@@ -1186,8 +1189,16 @@ where
 
 			LdkEvent::PaymentPathSuccessful { .. } => {},
 			LdkEvent::PaymentPathFailed { .. } => {},
-			LdkEvent::ProbeSuccessful { .. } => {},
-			LdkEvent::ProbeFailed { .. } => {},
+			LdkEvent::ProbeSuccessful { path, payment_id, .. } => {
+				if let Some(prober) = &self.prober {
+					prober.handle_background_probe_successful(&path, payment_id);
+				}
+			},
+			LdkEvent::ProbeFailed { path, payment_id, .. } => {
+				if let Some(prober) = &self.prober {
+					prober.handle_background_probe_failed(&path, payment_id);
+				}
+			},
 			LdkEvent::HTLCHandlingFailed { failure_type, .. } => {
 				if let Some(liquidity_source) = self.liquidity_source.as_ref() {
 					liquidity_source.handle_htlc_handling_failed(failure_type).await;

src/ffi/types.rs

@@ -148,6 +148,7 @@ pub use crate::entropy::{generate_entropy_mnemonic, NodeEntropy, WordCount};
 use crate::error::Error;
 pub use crate::liquidity::LSPS1OrderStatus;
 pub use crate::logger::{LogLevel, LogRecord, LogWriter};
+pub use crate::probing::ProbingConfig;
 use crate::{hex_utils, SocketAddress, UserChannelId};
 
 uniffi::custom_type!(PublicKey, String, {

src/lib.rs

@@ -101,10 +101,12 @@ pub mod logger;
 mod message_handler;
 pub mod payment;
 mod peer_store;
+pub mod probing;
 mod runtime;
 mod scoring;
 mod tx_broadcaster;
 mod types;
+mod util;
 mod wallet;
 
 use std::default::Default;
@@ -170,6 +172,7 @@ use payment::{
 	UnifiedPayment,
 };
 use peer_store::{PeerInfo, PeerStore};
+use probing::{run_prober, Prober};
 use runtime::Runtime;
 pub use tokio;
 use types::{
@@ -239,6 +242,7 @@ pub struct Node {
 	om_mailbox: Option<Arc<OnionMessageMailbox>>,
 	async_payments_role: Option<AsyncPaymentsRole>,
 	hrn_resolver: HRNResolver,
+	prober: Option<Arc<Prober>>,
 	#[cfg(cycle_tests)]
 	_leak_checker: LeakChecker,
 }
@@ -596,11 +600,19 @@ impl Node {
 			static_invoice_store,
 			Arc::clone(&self.onion_messenger),
 			self.om_mailbox.clone(),
+			self.prober.clone(),
 			Arc::clone(&self.runtime),
 			Arc::clone(&self.logger),
 			Arc::clone(&self.config),
 		));
 
+		if let Some(prober) = self.prober.clone() {
+			let stop_rx = self.stop_sender.subscribe();
+			self.runtime.spawn_cancellable_background_task(async move {
+				run_prober(prober, stop_rx).await;
+			});
+		}
+
 		// Setup background processing
 		let background_persister = Arc::clone(&self.kv_store);
 		let background_event_handler = Arc::clone(&event_handler);
@@ -1081,6 +1093,11 @@ impl Node {
 		))
 	}
 
+	/// Returns a reference to the [`Prober`], or `None` if no probing strategy is configured.
+	pub fn prober(&self) -> Option<&Prober> {
+		self.prober.as_deref()
+	}
+
 	/// 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()