Implement pending claim rebroadcast on force-closed channels

This attempts to rebroadcast/fee-bump each pending claim a monitor is
tracking for a force-closed channel. This is crucial in preventing
certain classes of pinning attacks and ensures reliability if
broadcasting fails. For implementations of `FeeEstimator` that also
support mempool fee estimation, we may broadcast a fee-bumped claim
instead, ensuring we can also react to mempool fee spikes between
blocks.

Author: wpaulino

lightning/src/chain/chainmonitor.rs

@@ -217,8 +217,14 @@ impl<ChannelSigner: WriteableEcdsaChannelSigner> Deref for LockedChannelMonitor<
 /// or used independently to monitor channels remotely. See the [module-level documentation] for
 /// details.
 ///
+/// Note that `ChainMonitor` should regularly trigger rebroadcasts/fee bumps of pending claims from
+/// a force-closed channel. This is crucial in preventing certain classes of pinning attacks and
+/// ensures reliability if broadcasting fails. In order to do so, we recommend invoking
+/// [`rebroadcast_pending_claims`] roughly once per minute, though it doesn't have to be perfect.
+///
 /// [`ChannelManager`]: crate::ln::channelmanager::ChannelManager
 /// [module-level documentation]: crate::chain::chainmonitor
+/// [`rebroadcast_pending_claims`]: Self::rebroadcast_pending_claims
 pub struct ChainMonitor<ChannelSigner: WriteableEcdsaChannelSigner, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref>
 	where C::Target: chain::Filter,
         T::Target: BroadcasterInterface,
@@ -533,6 +539,19 @@ where C::Target: chain::Filter,
 	pub fn get_update_future(&self) -> Future {
 		self.event_notifier.get_future()
 	}
+
+	/// Triggers rebroadcasts/fee-bumps of pending claims from a force-closed channel. This is
+	/// crucial in preventing certain classes of pinning attacks and ensures reliability if
+	/// broadcasting fails. We recommend invoking this roughly once per minute, though it doesn't
+	/// have to be perfect.
+	pub fn rebroadcast_pending_claims(&self) {
+		let monitors = self.monitors.read().unwrap();
+		for (_, monitor_holder) in &*monitors {
+			monitor_holder.monitor.rebroadcast_pending_claims(
+				&*self.broadcaster, &*self.fee_estimator, &*self.logger
+			)
+		}
+	}
 }
 
 impl<ChannelSigner: WriteableEcdsaChannelSigner, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref>

lightning/src/chain/channelmonitor.rs

@@ -1467,6 +1467,26 @@ impl<Signer: WriteableEcdsaChannelSigner> ChannelMonitor<Signer> {
 	pub fn current_best_block(&self) -> BestBlock {
 		self.inner.lock().unwrap().best_block.clone()
 	}
+
+	/// Triggers rebroadcasts/fee-bumps of pending claims from a force-closed channel. This is
+	/// crucial in preventing certain classes of pinning attacks and ensures reliability if
+	/// broadcasting fails. We recommend invoking this roughly once per minute, though it doesn't
+	/// have to be perfect.
+	pub fn rebroadcast_pending_claims<B: Deref, F: Deref, L: Deref>(
+		&self, broadcaster: B, fee_estimator: F, logger: L,
+	)
+	where
+		B::Target: BroadcasterInterface,
+		F::Target: FeeEstimator,
+		L::Target: Logger,
+	{
+		let fee_estimator = LowerBoundedFeeEstimator::new(fee_estimator);
+		let mut inner = self.inner.lock().unwrap();
+		let current_height = inner.best_block.height;
+		inner.onchain_tx_handler.rebroadcast_pending_claims(
+			current_height, &broadcaster, &fee_estimator, &logger,
+		);
+	}
 }
 
 impl<Signer: WriteableEcdsaChannelSigner> ChannelMonitorImpl<Signer> {

lightning/src/chain/onchaintx.rs

@@ -481,6 +481,54 @@ impl<ChannelSigner: WriteableEcdsaChannelSigner> OnchainTxHandler<ChannelSigner>
 		events.into_iter().map(|(_, event)| event).collect()
 	}
 
+	/// Triggers rebroadcasts/fee-bumps of pending claims from a force-closed channel. This is
+	/// crucial in preventing certain classes of pinning attacks and ensures reliability if
+	/// broadcasting fails. We recommend invoking this roughly once per minute, though it doesn't
+	/// have to be perfect.
+	pub(crate) fn rebroadcast_pending_claims<B: Deref, F: Deref, L: Deref>(
+		&mut self, current_height: u32, broadcaster: &B, fee_estimator: &LowerBoundedFeeEstimator<F>,
+		logger: &L,
+	)
+	where
+		B::Target: BroadcasterInterface,
+		F::Target: FeeEstimator,
+		L::Target: Logger,
+	{
+		let mut bump_requests = Vec::with_capacity(self.pending_claim_requests.len());
+		for (package_id, request) in self.pending_claim_requests.iter() {
+			let inputs = request.outpoints();
+			log_info!(logger, "Triggering rebroadcast/fee-bump for request with inputs {:?}", inputs);
+			bump_requests.push((*package_id, request.clone()));
+	   }
+		for (package_id, request) in bump_requests {
+			self.generate_claim(current_height, &request, false /* manually_bump_feerate */, fee_estimator, logger)
+				.map(|(_, new_feerate, claim)| {
+					if let Some(mut_request) = self.pending_claim_requests.get_mut(&package_id) {
+						mut_request.set_feerate(new_feerate);
+					}
+					match claim {
+						OnchainClaim::Tx(tx) => {
+							log_info!(logger, "Broadcasting RBF-bumped onchain {}", log_tx!(tx));
+							broadcaster.broadcast_transaction(&tx);
+						},
+						#[cfg(anchors)]
+						OnchainClaim::Event(event) => {
+							log_info!(logger, "Yielding fee-bumped onchain event to spend inputs {:?}",
+								request.outpoints());
+							#[cfg(debug_assertions)] {
+								debug_assert!(request.requires_external_funding());
+								let num_existing = self.pending_claim_events.iter()
+									.filter(|entry| entry.0 == package_id).count();
+								assert!(num_existing == 0 || num_existing == 1);
+							}
+							self.pending_claim_events.retain(|event| event.0 != package_id);
+							self.pending_claim_events.push((package_id, event));
+						}
+					}
+				});
+		}
+	}
+
 	/// Lightning security model (i.e being able to redeem/timeout HTLC or penalize counterparty
 	/// onchain) lays on the assumption of claim transactions getting confirmed before timelock
 	/// expiration (CSV or CLTV following cases). In case of high-fee spikes, claim tx may get stuck

lightning/src/ln/functional_test_utils.rs

@@ -151,6 +151,20 @@ impl ConnectStyle {
 		}
 	}
 
+	pub fn updates_best_block_first(&self) -> bool {
+		match self {
+			ConnectStyle::BestBlockFirst => true,
+			ConnectStyle::BestBlockFirstSkippingBlocks => true,
+			ConnectStyle::BestBlockFirstReorgsOnlyTip => true,
+			ConnectStyle::TransactionsFirst => false,
+			ConnectStyle::TransactionsFirstSkippingBlocks => false,
+			ConnectStyle::TransactionsDuplicativelyFirstSkippingBlocks => false,
+			ConnectStyle::HighlyRedundantTransactionsFirstSkippingBlocks => false,
+			ConnectStyle::TransactionsFirstReorgsOnlyTip => false,
+			ConnectStyle::FullBlockViaListen => false,
+		}
+	}
+
 	fn random_style() -> ConnectStyle {
 		#[cfg(feature = "std")] {
 			use core::hash::{BuildHasher, Hasher};

lightning/src/ln/monitor_tests.rs

@@ -1778,6 +1778,130 @@ fn test_restored_packages_retry() {
 	do_test_restored_packages_retry(true);
 }
 
+fn do_test_monitor_rebroadcast_pending_claims(anchors: bool) {
+	// Test that we will retry broadcasting pending claims for a force-closed channel on every
+	// `ChainMonitor::rebroadcast_pending_claims` call.
+	if anchors {
+		assert!(cfg!(anchors));
+	}
+	let mut chanmon_cfgs = create_chanmon_cfgs(2);
+	let node_cfgs = create_node_cfgs(2, &chanmon_cfgs);
+	let mut config = test_default_channel_config();
+	if anchors {
+		#[cfg(anchors)] {
+			config.channel_handshake_config.negotiate_anchors_zero_fee_htlc_tx = true;
+		}
+	}
+	let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[Some(config), Some(config)]);
+	let nodes = create_network(2, &node_cfgs, &node_chanmgrs);
+
+	let (_, _, _, chan_id, funding_tx) = create_chan_between_nodes_with_value(
+		&nodes[0], &nodes[1], 1_000_000, 500_000_000
+	);
+	const HTLC_AMT_MSAT: u64 = 1_000_000;
+	const HTLC_AMT_SAT: u64 = HTLC_AMT_MSAT / 1000;
+	route_payment(&nodes[0], &[&nodes[1]], HTLC_AMT_MSAT);
+
+	let htlc_expiry = nodes[0].best_block_info().1 + TEST_FINAL_CLTV + 1;
+
+	let commitment_txn = get_local_commitment_txn!(&nodes[0], &chan_id);
+	assert_eq!(commitment_txn.len(), if anchors { 1 /* commitment tx only */} else { 2 /* commitment and htlc timeout tx */ });
+	check_spends!(&commitment_txn[0], &funding_tx);
+	mine_transaction(&nodes[0], &commitment_txn[0]);
+	check_closed_broadcast!(&nodes[0], true);
+	check_closed_event(&nodes[0], 1, ClosureReason::CommitmentTxConfirmed, false);
+	check_added_monitors(&nodes[0], 1);
+
+	// Set up a helper closure we'll use throughout our test. We should only expect retries without
+	// bumps if fees have not increased after a block has been connected (assuming the height timer
+	// re-evaluates at every block) or after `ChainMonitor::rebroadcast_pending_claims` is called.
+	let mut prev_htlc_tx_feerate = None;
+	let mut check_htlc_retry = |should_retry: bool, should_bump: bool| {
+		let htlc_tx_feerate = if anchors {
+			assert!(nodes[0].tx_broadcaster.txn_broadcast().is_empty());
+			let mut events = nodes[0].chain_monitor.chain_monitor.get_and_clear_pending_events();
+			assert_eq!(events.len(), if should_retry { 1 } else { 0 });
+			if !should_retry {
+				return;
+			}
+			#[allow(unused_assignments)]
+			let mut feerate = 0;
+			#[cfg(anchors)] {
+				let (htlcs, htlc_tx_feerate) = if let Event::BumpTransaction(BumpTransactionEvent::HTLCResolution {
+					target_feerate_sat_per_1000_weight, htlc_descriptors, ..
+				}) = events.pop().unwrap() {
+					(htlc_descriptors, target_feerate_sat_per_1000_weight)
+				} else { panic!("unexpected event"); };
+				assert_eq!(htlcs.len(), 1);
+				assert_eq!(htlcs[0].commitment_txid, commitment_txn[0].txid());
+				assert!((htlcs[0].htlc.transaction_output_index.unwrap() as usize) < commitment_txn[0].output.len());
+				feerate = htlc_tx_feerate as u64;
+			}
+			feerate
+		} else {
+			assert!(nodes[0].chain_monitor.chain_monitor.get_and_clear_pending_events().is_empty());
+			let txn = nodes[0].tx_broadcaster.txn_broadcast();
+			assert_eq!(txn.len(), if should_retry { 1 } else { 0 });
+			if !should_retry {
+				return;
+			}
+			check_spends!(txn[0], commitment_txn[0]);
+			let htlc_tx_fee = HTLC_AMT_SAT - txn[0].output[0].value;
+			htlc_tx_fee * 1000 / txn[0].weight() as u64
+		};
+		if should_bump {
+			assert!(htlc_tx_feerate > prev_htlc_tx_feerate.unwrap());
+		} else if let Some(prev_feerate) = prev_htlc_tx_feerate.take() {
+			assert!(htlc_tx_feerate <= prev_feerate);
+		}
+		prev_htlc_tx_feerate = Some(htlc_tx_feerate);
+	};
+
+	// Connect blocks up to one before the HTLC expires. This should not result in a claim/retry.
+	connect_blocks(&nodes[0], htlc_expiry - nodes[0].best_block_info().1 - 2);
+	check_htlc_retry(false, false);
+
+	// Connect one more block, producing our first claim.
+	connect_blocks(&nodes[0], 1);
+	check_htlc_retry(true, false);
+
+	// Connect one more block, expecting a retry with a fee bump. Unfortunately, we cannot bump HTLC
+	// transactions pre-anchors.
+	connect_blocks(&nodes[0], 1);
+	check_htlc_retry(true, anchors);
+
+	// Trigger a call and we should have another retry, but without a bump.
+	nodes[0].chain_monitor.chain_monitor.rebroadcast_pending_claims();
+	check_htlc_retry(true, false);
+
+	// Double the feerate and trigger a call, expecting a fee-bumped retry.
+	*nodes[0].fee_estimator.sat_per_kw.lock().unwrap() *= 2;
+	nodes[0].chain_monitor.chain_monitor.rebroadcast_pending_claims();
+	check_htlc_retry(true, anchors);
+
+	// Connect one more block, expecting a retry with a fee bump. Unfortunately, we cannot bump HTLC
+	// transactions pre-anchors.
+	connect_blocks(&nodes[0], 1);
+	check_htlc_retry(true, anchors);
+
+	if !anchors {
+		// Mine the HTLC transaction to ensure we don't retry claims while they're confirmed.
+		mine_transaction(&nodes[0], &commitment_txn[1]);
+		if nodes[0].connect_style.borrow().updates_best_block_first() {
+			check_htlc_retry(true, false);
+		}
+		nodes[0].chain_monitor.chain_monitor.rebroadcast_pending_claims();
+		check_htlc_retry(false, false);
+	}
+}
+
+#[test]
+fn test_monitor_timer_based_claim() {
+	do_test_monitor_rebroadcast_pending_claims(false);
+	#[cfg(anchors)]
+	do_test_monitor_rebroadcast_pending_claims(true);
+}
+
 #[cfg(anchors)]
 #[test]
 fn test_yield_anchors_events() {