Retry user-initiated splices across restarts and disconnects

bindings/ldk_node.udl

@@ -124,6 +124,8 @@ interface Node {
 	[Throws=NodeError]
 	void splice_out([ByRef]UserChannelId user_channel_id, PublicKey counterparty_node_id, [ByRef]Address address, u64 splice_amount_sats);
 	[Throws=NodeError]
+	void bump_channel_funding_fee([ByRef]UserChannelId user_channel_id, PublicKey counterparty_node_id);
+	[Throws=NodeError]
 	void close_channel([ByRef]UserChannelId user_channel_id, PublicKey counterparty_node_id);
 	[Throws=NodeError]
 	void force_close_channel([ByRef]UserChannelId user_channel_id, PublicKey counterparty_node_id, string? reason);

src/builder.rs

@@ -64,7 +64,9 @@ use crate::io::utils::{
 };
 use crate::io::vss_store::VssStoreBuilder;
 use crate::io::{
-	self, PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
+	self, CHANNEL_RECORD_PERSISTENCE_PRIMARY_NAMESPACE,
+	CHANNEL_RECORD_PERSISTENCE_SECONDARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
+	PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
 	PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
 	PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
 };
@@ -77,9 +79,9 @@ use crate::peer_store::PeerStore;
 use crate::runtime::{Runtime, RuntimeSpawner};
 use crate::tx_broadcaster::TransactionBroadcaster;
 use crate::types::{
-	AsyncPersister, ChainMonitor, ChannelManager, DynStore, DynStoreRef, DynStoreWrapper,
-	GossipSync, Graph, HRNResolver, KeysManager, MessageRouter, OnionMessenger, PaymentStore,
-	PeerManager, PendingPaymentStore,
+	AsyncPersister, ChainMonitor, ChannelManager, ChannelRecordStore, DynStore, DynStoreRef,
+	DynStoreWrapper, GossipSync, Graph, HRNResolver, KeysManager, MessageRouter, OnionMessenger,
+	PaymentStore, PeerManager, PendingPaymentStore,
 };
 use crate::wallet::persist::KVStoreWalletPersister;
 use crate::wallet::Wallet;
@@ -1379,7 +1381,7 @@ fn build_with_store_internal(
 
 	let kv_store_ref = Arc::clone(&kv_store);
 	let logger_ref = Arc::clone(&logger);
-	let (payment_store_res, node_metris_res, pending_payment_store_res) =
+	let (payment_store_res, node_metris_res, pending_payment_store_res, channel_record_store_res) =
 		runtime.block_on(async move {
 			tokio::join!(
 				read_all_objects(
@@ -1394,6 +1396,12 @@ fn build_with_store_internal(
 					PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
 					PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
 					Arc::clone(&logger_ref),
+				),
+				read_all_objects(
+					&*kv_store_ref,
+					CHANNEL_RECORD_PERSISTENCE_PRIMARY_NAMESPACE,
+					CHANNEL_RECORD_PERSISTENCE_SECONDARY_NAMESPACE,
+					Arc::clone(&logger_ref),
 				)
 			)
 		});
@@ -1605,6 +1613,20 @@ fn build_with_store_internal(
 		},
 	};
 
+	let channel_record_store = match channel_record_store_res {
+		Ok(channel_records) => Arc::new(ChannelRecordStore::new(
+			channel_records,
+			CHANNEL_RECORD_PERSISTENCE_PRIMARY_NAMESPACE.to_string(),
+			CHANNEL_RECORD_PERSISTENCE_SECONDARY_NAMESPACE.to_string(),
+			Arc::clone(&kv_store),
+			Arc::clone(&logger),
+		)),
+		Err(e) => {
+			log_error!(logger, "Failed to read channel record data from store: {}", e);
+			return Err(BuildError::ReadFailed);
+		},
+	};
+
 	let wallet = Arc::new(Wallet::new(
 		bdk_wallet,
 		wallet_persister,
@@ -1618,6 +1640,8 @@ fn build_with_store_internal(
 		Arc::clone(&pending_payment_store),
 	));
 
+	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);
@@ -2149,6 +2173,7 @@ fn build_with_store_internal(
 		scorer,
 		peer_store,
 		payment_store,
+		channel_record_store,
 		lnurl_auth,
 		is_running,
 		node_metrics,

src/chain/bitcoind.rs

@@ -571,16 +571,18 @@ impl BitcoindChainSource {
 		Ok(())
 	}
 
-	pub(crate) async fn process_broadcast_package(&self, package: Vec<Transaction>) {
+	pub(crate) async fn process_broadcast_package(
+		&self, txs: impl IntoIterator<Item = Transaction>,
+	) {
 		// While it's a bit unclear when we'd be able to lean on Bitcoin Core >v28
 		// features, we should eventually switch to use `submitpackage` via the
 		// `rust-bitcoind-json-rpc` crate rather than just broadcasting individual
 		// transactions.
-		for tx in &package {
+		for tx in txs {
 			let txid = tx.compute_txid();
 			let timeout_fut = tokio::time::timeout(
 				Duration::from_secs(DEFAULT_TX_BROADCAST_TIMEOUT_SECS),
-				self.api_client.broadcast_transaction(tx),
+				self.api_client.broadcast_transaction(&tx),
 			);
 			match timeout_fut.await {
 				Ok(res) => match res {

src/chain/electrum.rs

@@ -294,7 +294,9 @@ impl ElectrumChainSource {
 		Ok(())
 	}
 
-	pub(crate) async fn process_broadcast_package(&self, package: Vec<Transaction>) {
+	pub(crate) async fn process_broadcast_package(
+		&self, txs: impl IntoIterator<Item = Transaction>,
+	) {
 		let electrum_client: Arc<ElectrumRuntimeClient> = if let Some(client) =
 			self.electrum_runtime_status.read().expect("lock").client().as_ref()
 		{
@@ -304,7 +306,7 @@ impl ElectrumChainSource {
 			return;
 		};
 
-		for tx in package {
+		for tx in txs {
 			electrum_client.broadcast(tx).await;
 		}
 	}

src/chain/esplora.rs

@@ -355,12 +355,14 @@ impl EsploraChainSource {
 		Ok(())
 	}
 
-	pub(crate) async fn process_broadcast_package(&self, package: Vec<Transaction>) {
-		for tx in &package {
+	pub(crate) async fn process_broadcast_package(
+		&self, txs: impl IntoIterator<Item = Transaction>,
+	) {
+		for tx in txs {
 			let txid = tx.compute_txid();
 			let timeout_fut = tokio::time::timeout(
 				Duration::from_secs(self.sync_config.timeouts_config.tx_broadcast_timeout_secs),
-				self.esplora_client.broadcast(tx),
+				self.esplora_client.broadcast(&tx),
 			);
 			match timeout_fut.await {
 				Ok(res) => match res {

src/chain/mod.rs

@@ -24,7 +24,7 @@ use crate::config::{
 	WALLET_SYNC_INTERVAL_MINIMUM_SECS,
 };
 use crate::fee_estimator::OnchainFeeEstimator;
-use crate::logger::{log_debug, log_info, log_trace, LdkLogger, Logger};
+use crate::logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger};
 use crate::runtime::Runtime;
 use crate::types::{Broadcaster, ChainMonitor, ChannelManager, DynStore, Sweeper, Wallet};
 use crate::{Error, PersistedNodeMetrics};
@@ -453,15 +453,27 @@ impl ChainSource {
 					return;
 				}
 				Some(next_package) = receiver.recv() => {
+					let package = match self.tx_broadcaster.classify_package(next_package).await {
+						Ok(p) => p,
+						Err(e) => {
+							log_error!(
+								tx_bcast_logger,
+								"Skipping broadcast: failed to persist payment records: {:?}",
+								e,
+							);
+							continue;
+						},
+					};
+					let txs = package.into_iter().map(|(tx, _)| tx);
 					match &self.kind {
 						ChainSourceKind::Esplora(esplora_chain_source) => {
-							esplora_chain_source.process_broadcast_package(next_package).await
+							esplora_chain_source.process_broadcast_package(txs).await
 						},
 						ChainSourceKind::Electrum(electrum_chain_source) => {
-							electrum_chain_source.process_broadcast_package(next_package).await
+							electrum_chain_source.process_broadcast_package(txs).await
 						},
 						ChainSourceKind::Bitcoind(bitcoind_chain_source) => {
-							bitcoind_chain_source.process_broadcast_package(next_package).await
+							bitcoind_chain_source.process_broadcast_package(txs).await
 						},
 					}
 				}