Track in-flight splices for failure reporting and crash recovery

src/builder.rs

@@ -53,11 +53,13 @@ use lightning_dns_resolver::OMDomainResolver;
 use vss_client::headers::VssHeaderProvider;
 
 use crate::chain::ChainSource;
+use crate::channel::SpliceTracker;
 #[cfg(feature = "chain-bitcoind")]
 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 +85,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 +108,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 +1569,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 +1593,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 +1931,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 +1969,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 +1979,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 +2459,15 @@ fn build_with_store_internal(
 		},
 	};
 
+	// The wallet drops the facts it recorded for a channel once that channel is not live
+	// anymore, and records on start what a live channel's producers never reported. Both need the
+	// node's channel state, which exists from this point on.
+	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) => {
@@ -2512,6 +2551,13 @@ fn build_with_store_internal(
 		})
 	});
 
+	let splice_tracker = Arc::new(SpliceTracker::new(
+		Arc::clone(&channel_manager),
+		Arc::clone(&wallet),
+		Arc::clone(&pending_payment_store),
+		Arc::clone(&logger),
+	));
+
 	#[cfg(cycle_tests)]
 	let mut _leak_checker = crate::LeakChecker(Vec::new());
 	#[cfg(cycle_tests)]
@@ -2563,6 +2609,7 @@ fn build_with_store_internal(
 		payment_store,
 		forwarding_store,
 		forwarded_payment_aggregation_retention_secs,
+		splice_tracker,
 		lnurl_auth,
 		is_running,
 		node_metrics,

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;
 		}
 	}
 }