Keep funding payment records accurate

src/builder.rs

@@ -58,6 +58,7 @@ use crate::config::BitcoindRestClientConfig;
 use crate::config::{
 	default_user_config, may_announce_channel, AnnounceError, AsyncPaymentsRole, Config,
 	ElectrumSyncConfig, EsploraSyncConfig, HRNResolverConfig, TorConfig,
+	CHANNEL_TX_FACTS_CACHE_CAPACITY, CHANNEL_TX_FACTS_CACHE_WARMUP_COUNT,
 	DEFAULT_ESPLORA_SERVER_URL, DEFAULT_LOG_FILENAME, DEFAULT_LOG_LEVEL,
 	DEFAULT_MAX_PROBE_AMOUNT_MSAT, DEFAULT_MIN_PROBE_AMOUNT_MSAT, PAYMENT_CACHE_CAPACITY,
 	PAYMENT_CACHE_WARMUP_COUNT,
@@ -83,6 +84,8 @@ use crate::io::utils::{
 use crate::io::vss_store::VssStoreBuilder;
 use crate::io::{
 	self, CHANNEL_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE,
+	CHANNEL_TX_FACTS_PERSISTENCE_PRIMARY_NAMESPACE,
+	CHANNEL_TX_FACTS_PERSISTENCE_SECONDARY_NAMESPACE,
 	FORWARDED_PAYMENT_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
 	PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
 	PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
@@ -104,11 +107,12 @@ use crate::probing::{
 use crate::runtime::{Runtime, RuntimeSpawner};
 use crate::tx_broadcaster::TransactionBroadcaster;
 use crate::types::{
-	AsyncPersister, ChainMonitor, ChannelManager, DynStore, DynStoreRef, DynStoreWrapper,
-	GossipSync, Graph, KeysManager, MessageRouter, OnionMessenger, PaymentStore, PeerManager,
-	PendingPaymentStore,
+	AsyncPersister, ChainMonitor, ChannelManager, ChannelTxFactsStore, DynStore, DynStoreRef,
+	DynStoreWrapper, GossipSync, Graph, KeysManager, MessageRouter, OnionMessenger, PaymentStore,
+	PeerManager, PendingPaymentStore,
 };
 use crate::wallet::persist::{read_address_pool, KVStoreWalletPersister};
+use crate::wallet::provenance::NodeChannelLiveness;
 use crate::wallet::Wallet;
 use crate::{Node, NodeMetrics, PersistedNodeMetrics};
 
@@ -1564,6 +1568,7 @@ fn build_with_store_internal(
 		channel_forwarding_stats_res,
 		node_metris_res,
 		pending_payment_store_res,
+		channel_tx_facts_store_res,
 		address_pool_res,
 	) = runtime.block_on(async move {
 		tokio::join!(
@@ -1587,6 +1592,13 @@ fn build_with_store_internal(
 				PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
 				Arc::clone(&logger_ref),
 			),
+			read_n_objects(
+				&*kv_store_ref,
+				CHANNEL_TX_FACTS_PERSISTENCE_PRIMARY_NAMESPACE,
+				CHANNEL_TX_FACTS_PERSISTENCE_SECONDARY_NAMESPACE,
+				CHANNEL_TX_FACTS_CACHE_WARMUP_COUNT,
+				Arc::clone(&logger_ref),
+			),
 			read_address_pool(&*kv_store_ref, &*logger_ref),
 		)
 	});
@@ -1918,6 +1930,24 @@ fn build_with_store_internal(
 		},
 	};
 
+	let channel_tx_facts_store = match channel_tx_facts_store_res {
+		Ok(channel_tx_facts) => Arc::new(ChannelTxFactsStore::new(
+			// The read hands us the newest records first, while the cache treats the objects it
+			// is seeded with as increasingly recently used. Reverse them, so that the newest
+			// record is the last one to be evicted rather than the first.
+			channel_tx_facts.into_iter().rev().collect(),
+			KeepLeastRecentlyUsed::new(CHANNEL_TX_FACTS_CACHE_CAPACITY),
+			CHANNEL_TX_FACTS_PERSISTENCE_PRIMARY_NAMESPACE.to_string(),
+			CHANNEL_TX_FACTS_PERSISTENCE_SECONDARY_NAMESPACE.to_string(),
+			Arc::clone(&kv_store),
+			Arc::clone(&logger),
+		)),
+		Err(e) => {
+			log_error!(logger, "Failed to read channel transaction facts from store: {}", e);
+			return Err(BuildError::ReadFailed);
+		},
+	};
+
 	let persisted_pool_indices = match address_pool_res {
 		Ok(indices) => indices,
 		Err(e) => {
@@ -1938,6 +1968,7 @@ fn build_with_store_internal(
 		Arc::clone(&config),
 		Arc::clone(&logger),
 		Arc::clone(&pending_payment_store),
+		Arc::clone(&channel_tx_facts_store),
 	));
 
 	// Fill the address pool up front so LDK's sync `SignerProvider` callbacks can hand out
@@ -1947,8 +1978,6 @@ fn build_with_store_internal(
 		BuildError::WalletSetupFailed
 	})?;
 
-	tx_broadcaster.set_wallet(Arc::downgrade(&wallet));
-
 	// Initialize the KeysManager
 	let cur_time = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).map_err(|e| {
 		log_error!(logger, "Failed to get current time: {}", e);
@@ -2429,6 +2458,15 @@ fn build_with_store_internal(
 		},
 	};
 
+	// The wallet drops the facts it recorded for a channel once nothing holds that channel
+	// anymore, and records on start what a held channel's producers never reported; both it can
+	// only ask now that the node's channel state exists.
+	wallet.set_channel_liveness(Arc::new(NodeChannelLiveness::new(
+		&channel_manager,
+		&chain_monitor,
+		&output_sweeper,
+	)));
+
 	let event_queue = match event_queue_res {
 		Ok(event_queue) => Arc::new(event_queue),
 		Err(e) => {

src/chain/mod.rs

@@ -35,8 +35,9 @@ use crate::config::ElectrumSyncConfig;
 use crate::config::EsploraSyncConfig;
 use crate::config::{BackgroundSyncConfig, Config, WALLET_SYNC_INTERVAL_MINIMUM_SECS};
 use crate::fee_estimator::OnchainFeeEstimator;
-use crate::logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger};
+use crate::logger::{log_debug, log_info, log_trace, LdkLogger, Logger};
 use crate::runtime::Runtime;
+use crate::tx_broadcaster::BroadcastPackage;
 use crate::types::{Broadcaster, ChainMonitor, ChannelManager, DynStore, Sweeper, Wallet};
 use crate::{Error, PersistedNodeMetrics};
 
@@ -562,52 +563,44 @@ impl ChainSource {
 		}
 	}
 
+	/// Hands the package to the configured chain source, parents before their child so a CPFP
+	/// package a chain source submits one transaction at a time is still accepted.
+	async fn broadcast(&self, package: BroadcastPackage) {
+		let package = package.into_sorted_transactions();
+		match &self.kind {
+			#[cfg(feature = "chain-esplora")]
+			ChainSourceKind::Esplora(esplora_chain_source) => {
+				esplora_chain_source.process_transaction_broadcast(package).await
+			},
+			#[cfg(feature = "chain-electrum")]
+			ChainSourceKind::Electrum(electrum_chain_source) => {
+				electrum_chain_source.process_transaction_broadcast(package).await
+			},
+			#[cfg(feature = "chain-bitcoind")]
+			ChainSourceKind::Bitcoind(bitcoind_chain_source) => {
+				bitcoind_chain_source.process_transaction_broadcast(package).await
+			},
+		}
+	}
+
 	pub(crate) async fn continuously_process_broadcast_queue(
 		&self, mut stop_tx_bcast_receiver: tokio::sync::watch::Receiver<()>,
 	) {
-		let mut receiver = self.tx_broadcaster.get_broadcast_queue().await;
 		loop {
-			let tx_bcast_logger = Arc::clone(&self.logger);
-			tokio::select! {
+			let package = tokio::select! {
+				// A stop request is polled first, so a queue that always has a package ready
+				// cannot starve it.
+				biased;
 				_ = stop_tx_bcast_receiver.changed() => {
 					log_debug!(
-						tx_bcast_logger,
+						self.logger,
 						"Stopping broadcasting transactions.",
 					);
 					return;
 				}
-				Some(next_package) = receiver.recv() => {
-					// Classify funding broadcasts into payment records before sending. If
-					// classification fails we skip the broadcast, since broadcasting a tx we
-					// failed to record would leave it on-chain without a payment.
-					let package = match self.tx_broadcaster.classify_package(next_package).await {
-						Ok(package) => package,
-						Err(e) => {
-							log_error!(
-								tx_bcast_logger,
-								"Skipping broadcast: failed to persist payment records: {:?}",
-								e,
-							);
-							continue;
-						},
-					};
-					let package = package.into_sorted_transactions();
-					match &self.kind {
-						#[cfg(feature = "chain-esplora")]
-						ChainSourceKind::Esplora(esplora_chain_source) => {
-							esplora_chain_source.process_transaction_broadcast(package).await
-						},
-						#[cfg(feature = "chain-electrum")]
-						ChainSourceKind::Electrum(electrum_chain_source) => {
-							electrum_chain_source.process_transaction_broadcast(package).await
-						},
-						#[cfg(feature = "chain-bitcoind")]
-						ChainSourceKind::Bitcoind(bitcoind_chain_source) => {
-							bitcoind_chain_source.process_transaction_broadcast(package).await
-						},
-					}
-				}
-			}
+				package = self.tx_broadcaster.next_package() => package,
+			};
+			self.broadcast(package).await;
 		}
 	}
 }

src/config.rs

@@ -65,6 +65,59 @@ pub(crate) const PAYMENT_CACHE_CAPACITY: NonZeroUsize = NonZeroUsize::new(1000).
 // may displace those entries.
 pub(crate) const PAYMENT_CACHE_WARMUP_COUNT: NonZeroUsize = NonZeroUsize::new(50).unwrap();
 
+// The number of channel transaction provenance records we keep in memory.
+//
+// A record is written when a channel produces a transaction and read back when the wallet meets
+// that transaction, so the working set is a node's recent channel activity rather than its whole
+// history. Records are small — a handful of outpoints, each with a role and a channel reference
+// — so this bounds the store's share of memory well below the payment store's while still
+// covering the channels a node is busy with.
+pub(crate) const CHANNEL_TX_FACTS_CACHE_CAPACITY: NonZeroUsize = NonZeroUsize::new(1000).unwrap();
+
+// The number of channel transaction provenance records we read into the cache when starting up.
+//
+// This matches the built-in storage backends' page size, so warming the cache costs a single page
+// listing and one batch of reads. Later activity may displace those entries, which are then read
+// back individually as they are needed.
+pub(crate) const CHANNEL_TX_FACTS_CACHE_WARMUP_COUNT: NonZeroUsize = NonZeroUsize::new(50).unwrap();
+
+// The number of blocks a channel transaction provenance record outlives the last thing the node
+// learned about its transaction.
+//
+// Roughly a year at ten minutes a block. It is an absolute backstop rather than the usual reason
+// a record goes: a record is dropped only once the channels it names are gone from the node's
+// channel manager, chain monitor and output sweeper, and the funding it records has been spent
+// and settled. Those checks are blind to a transaction of a channel that never reached them, so
+// without the cap such a record would be kept forever.
+pub(crate) const CHANNEL_TX_FACTS_RETENTION_BLOCKS: u32 = 52_560;
+
+// The number of bytes one channel transaction provenance record may take up.
+//
+// A record is written whole and holds one entry per channel-controlled output of its transaction,
+// so a counterparty loading a commitment transaction with HTLCs grows a record this node is
+// obliged to keep. The limit is comfortably above a commitment transaction carrying the most
+// HTLCs LDK allows, and bounds what any single transaction can cost.
+pub(crate) const CHANNEL_TX_FACTS_MAX_RECORD_BYTES: usize = 128 * 1024;
+
+// The number of channel transaction provenance records the node keeps.
+//
+// Records are dropped only once the channels they belong to have resolved, so between prunes a
+// counterparty opening and closing channels, or replacing a negotiated funding again and again,
+// drives the store's growth. Past this many records nothing new is admitted and the transactions
+// it would have described go unclassified, which is bounded loss of detail rather than unbounded
+// storage.
+pub(crate) const CHANNEL_TX_FACTS_MAX_RECORDS: usize = 100_000;
+
+// The number of pages of channel transaction provenance records one chain tip change examines.
+//
+// Pruning shares the pass that graduates payments, so it has to leave promptly; it resumes where
+// it left off on the next tip and so walks the whole store over consecutive blocks. At the
+// built-in backends' page size this is a couple of hundred records a block: a store whose records
+// fit the cache is walked in a single tip and costs the backend nothing beyond listing its keys,
+// while one at the limit above takes a few hundred blocks — which is also how stale the record
+// count that walk maintains can get.
+pub(crate) const CHANNEL_TX_FACTS_PRUNE_PAGES_PER_TIP: usize = 4;
+
 // The default {Esplora,Electrum} client timeout we're using.
 const DEFAULT_PER_REQUEST_TIMEOUT_SECS: u8 = 10;