Introduce LatestCounterpartyCommitment monitor update variant

This new variant is a backwards-incompatible successor to
`LatestCounterpartyCommitmentTXInfo` that is capable of handling
counterparty commitment updates while a splice is pending. Since all
counterparty commitment candidates share the same set of non-dust and
dust HTLCs (though each non-dust HTLC can have differing output
indices), we can share the same set of HTLC source data across all
candidates. Unfortunately, each `FundingScope` tracks its own set of
`counterparty_claimable_outpoints`, which includes the HTLC source data
(though only for the current and previous counterparty commitments), so
we end up duplicating it (both in memory and on disk) temporarily.

Author: wpaulino

lightning/src/chain/channelmonitor.rs

@@ -649,11 +649,9 @@ pub(crate) enum ChannelMonitorUpdateStep {
 		to_broadcaster_value_sat: Option<u64>,
 		to_countersignatory_value_sat: Option<u64>,
 	},
-	LatestCounterpartyCommitmentTX {
-		// The dust and non-dust htlcs for that commitment
-		htlc_outputs: Vec<(HTLCOutputInCommitment, Option<Box<HTLCSource>>)>,
-		// Contains only the non-dust htlcs
-		commitment_tx: CommitmentTransaction,
+	LatestCounterpartyCommitment {
+		commitment_txs: Vec<CommitmentTransaction>,
+		htlc_data: CommitmentHTLCData,
 	},
 	PaymentPreimage {
 		payment_preimage: PaymentPreimage,
@@ -689,7 +687,7 @@ impl ChannelMonitorUpdateStep {
 			ChannelMonitorUpdateStep::LatestHolderCommitmentTXInfo { .. } => "LatestHolderCommitmentTXInfo",
 			ChannelMonitorUpdateStep::LatestHolderCommitment { .. } => "LatestHolderCommitment",
 			ChannelMonitorUpdateStep::LatestCounterpartyCommitmentTXInfo { .. } => "LatestCounterpartyCommitmentTXInfo",
-			ChannelMonitorUpdateStep::LatestCounterpartyCommitmentTX { .. } => "LatestCounterpartyCommitmentTX",
+			ChannelMonitorUpdateStep::LatestCounterpartyCommitment { .. } => "LatestCounterpartyCommitment",
 			ChannelMonitorUpdateStep::PaymentPreimage { .. } => "PaymentPreimage",
 			ChannelMonitorUpdateStep::CommitmentSecret { .. } => "CommitmentSecret",
 			ChannelMonitorUpdateStep::ChannelForceClosed { .. } => "ChannelForceClosed",
@@ -729,9 +727,9 @@ impl_writeable_tlv_based_enum_upgradable!(ChannelMonitorUpdateStep,
 	(5, ShutdownScript) => {
 		(0, scriptpubkey, required),
 	},
-	(6, LatestCounterpartyCommitmentTX) => {
-		(0, htlc_outputs, required_vec),
-		(2, commitment_tx, required),
+	(6, LatestCounterpartyCommitment) => {
+		(1, commitment_txs, required_vec),
+		(3, htlc_data, required),
 	},
 	(8, LatestHolderCommitment) => {
 		(1, commitment_txs, required_vec),
@@ -1847,34 +1845,29 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 	///
 	/// This is used to provide the counterparty commitment transaction directly to the monitor
 	/// before the initial persistence of a new channel.
-	pub(crate) fn provide_initial_counterparty_commitment_tx<L: Deref>(
-		&self, commitment_tx: CommitmentTransaction, logger: &L,
-	) where
-		L::Target: Logger,
-	{
+	pub(crate) fn provide_initial_counterparty_commitment_tx(
+		&self, commitment_tx: CommitmentTransaction,
+	) {
 		let mut inner = self.inner.lock().unwrap();
-		let logger = WithChannelMonitor::from_impl(logger, &*inner, None);
-		inner.provide_initial_counterparty_commitment_tx(commitment_tx, &logger);
+		inner.provide_initial_counterparty_commitment_tx(commitment_tx);
 	}
 
 	/// Informs this monitor of the latest counterparty (ie non-broadcastable) commitment transaction.
 	/// The monitor watches for it to be broadcasted and then uses the HTLC information (and
 	/// possibly future revocation/preimage information) to claim outputs where possible.
 	/// We cache also the mapping hash:commitment number to lighten pruning of old preimages by watchtowers.
 	#[cfg(test)]
-	#[rustfmt::skip]
-	fn provide_latest_counterparty_commitment_tx<L: Deref>(
-		&self,
-		txid: Txid,
-		htlc_outputs: Vec<(HTLCOutputInCommitment, Option<Box<HTLCSource>>)>,
-		commitment_number: u64,
-		their_per_commitment_point: PublicKey,
-		logger: &L,
-	) where L::Target: Logger {
+	fn provide_latest_counterparty_commitment_tx(
+		&self, txid: Txid, htlc_outputs: Vec<(HTLCOutputInCommitment, Option<Box<HTLCSource>>)>,
+		commitment_number: u64, their_per_commitment_point: PublicKey,
+	) {
 		let mut inner = self.inner.lock().unwrap();
-		let logger = WithChannelMonitor::from_impl(logger, &*inner, None);
 		inner.provide_latest_counterparty_commitment_tx(
-			txid, htlc_outputs, commitment_number, their_per_commitment_point, &logger)
+			txid,
+			htlc_outputs,
+			commitment_number,
+			their_per_commitment_point,
+		)
 	}
 
 	#[cfg(test)]
@@ -3261,9 +3254,9 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 	}
 
 	#[rustfmt::skip]
-	fn provide_initial_counterparty_commitment_tx<L: Deref>(
-		&mut self, commitment_tx: CommitmentTransaction, logger: &WithChannelMonitor<L>,
-	) where L::Target: Logger {
+	fn provide_initial_counterparty_commitment_tx(
+		&mut self, commitment_tx: CommitmentTransaction,
+	) {
 		// We populate this field for downgrades
 		self.initial_counterparty_commitment_info = Some((commitment_tx.per_commitment_point(),
 			commitment_tx.feerate_per_kw(), commitment_tx.to_broadcaster_value_sat(), commitment_tx.to_countersignatory_value_sat()));
@@ -3274,17 +3267,16 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 		}
 
 		self.provide_latest_counterparty_commitment_tx(commitment_tx.trust().txid(), Vec::new(), commitment_tx.commitment_number(),
-				commitment_tx.per_commitment_point(), logger);
+				commitment_tx.per_commitment_point());
 		// Soon, we will only populate this field
 		self.initial_counterparty_commitment_tx = Some(commitment_tx);
 	}
 
 	#[rustfmt::skip]
-	fn provide_latest_counterparty_commitment_tx<L: Deref>(
-		&mut self, txid: Txid,
-		htlc_outputs: Vec<(HTLCOutputInCommitment, Option<Box<HTLCSource>>)>,
-		commitment_number: u64, their_per_commitment_point: PublicKey, logger: &WithChannelMonitor<L>,
-	) where L::Target: Logger {
+	fn provide_latest_counterparty_commitment_tx(
+		&mut self, txid: Txid, htlc_outputs: Vec<(HTLCOutputInCommitment, Option<Box<HTLCSource>>)>,
+		commitment_number: u64, their_per_commitment_point: PublicKey,
+	) {
 		// TODO: Encrypt the htlc_outputs data with the single-hash of the commitment transaction
 		// so that a remote monitor doesn't learn anything unless there is a malicious close.
 		// (only maybe, sadly we cant do the same for local info, as we need to be aware of
@@ -3293,11 +3285,11 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 			self.counterparty_hash_commitment_number.insert(htlc.payment_hash, commitment_number);
 		}
 
-		log_trace!(logger, "Tracking new counterparty commitment transaction with txid {} at commitment number {} with {} HTLC outputs", txid, commitment_number, htlc_outputs.len());
 		self.funding.prev_counterparty_commitment_txid = self.funding.current_counterparty_commitment_txid.take();
 		self.funding.current_counterparty_commitment_txid = Some(txid);
-		self.funding.counterparty_claimable_outpoints.insert(txid, htlc_outputs.clone());
+		self.funding.counterparty_claimable_outpoints.insert(txid, htlc_outputs);
 		self.current_counterparty_commitment_number = commitment_number;
+
 		//TODO: Merge this into the other per-counterparty-transaction output storage stuff
 		match self.their_cur_per_commitment_points {
 			Some(old_points) => {
@@ -3319,6 +3311,57 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 		}
 	}
 
+	fn update_counterparty_commitment_data(
+		&mut self, commitment_txs: &[CommitmentTransaction], htlc_data: &CommitmentHTLCData,
+	) -> Result<(), &'static str> {
+		self.verify_matching_commitment_transactions(commitment_txs.iter())?;
+
+		let htlcs_for_commitment = |commitment: &CommitmentTransaction| {
+			debug_assert!(htlc_data.nondust_htlc_sources.len() <= commitment.nondust_htlcs().len());
+			let mut nondust_htlcs = commitment.nondust_htlcs().iter();
+			let mut sources = htlc_data.nondust_htlc_sources.iter();
+			let nondust_htlcs = core::iter::from_fn(move || {
+				let htlc = nondust_htlcs.next()?.clone();
+				let source = (!htlc.offered).then(|| {
+					let source = sources
+						.next()
+						.expect("Every inbound non-dust HTLC should have a corresponding source")
+						.clone();
+					Box::new(source)
+				});
+				Some((htlc, source))
+			});
+
+			let dust_htlcs = htlc_data.dust_htlcs.iter().map(|(htlc, source)| {
+				(htlc.clone(), source.as_ref().map(|source| Box::new(source.clone())))
+			});
+
+			nondust_htlcs.chain(dust_htlcs).collect::<Vec<_>>()
+		};
+
+		let current_funding_commitment_tx = commitment_txs.first().unwrap();
+		self.provide_latest_counterparty_commitment_tx(
+			current_funding_commitment_tx.trust().txid(),
+			htlcs_for_commitment(current_funding_commitment_tx),
+			current_funding_commitment_tx.commitment_number(),
+			current_funding_commitment_tx.per_commitment_point(),
+		);
+
+		for (pending_funding, commitment_tx) in
+			self.pending_funding.iter_mut().zip(commitment_txs.iter().skip(1))
+		{
+			let commitment_txid = commitment_tx.trust().txid();
+			pending_funding.prev_counterparty_commitment_txid =
+				pending_funding.current_counterparty_commitment_txid.take();
+			pending_funding.current_counterparty_commitment_txid = Some(commitment_txid);
+			pending_funding
+				.counterparty_claimable_outpoints
+				.insert(commitment_txid, htlcs_for_commitment(commitment_tx));
+		}
+
+		Ok(())
+	}
+
 	/// Informs this monitor of the latest holder (ie broadcastable) commitment transaction. The
 	/// monitor watches for timeouts and may broadcast it if we approach such a timeout. Thus, it
 	/// is important that any clones of this channel monitor (including remote clones) by kept
@@ -3851,14 +3894,24 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 				},
 				// Soon we will drop the `LatestCounterpartyCommitmentTXInfo` variant in favor of `LatestCounterpartyCommitment`.
 				// For now we just add the code to handle the new updates.
-				// Next step: in channel, switch channel monitor updates to use the `LatestCounterpartyCommitmentTX` variant.
+				// Next step: in channel, switch channel monitor updates to use the `LatestCounterpartyCommitment` variant.
 				ChannelMonitorUpdateStep::LatestCounterpartyCommitmentTXInfo { commitment_txid, htlc_outputs, commitment_number, their_per_commitment_point, .. } => {
 					log_trace!(logger, "Updating ChannelMonitor with latest counterparty commitment transaction info");
-					self.provide_latest_counterparty_commitment_tx(*commitment_txid, htlc_outputs.clone(), *commitment_number, *their_per_commitment_point, logger)
+					if self.pending_funding.is_empty() {
+						self.provide_latest_counterparty_commitment_tx(*commitment_txid, htlc_outputs.clone(), *commitment_number, *their_per_commitment_point)
+					} else {
+						log_error!(logger, "Received unexpected non-splice counterparty commitment monitor update");
+						ret = Err(());
+					}
 				},
-				ChannelMonitorUpdateStep::LatestCounterpartyCommitmentTX { htlc_outputs, commitment_tx } => {
-					log_trace!(logger, "Updating ChannelMonitor with latest counterparty commitment transaction info");
-					self.provide_latest_counterparty_commitment_tx(commitment_tx.trust().txid(), htlc_outputs.clone(), commitment_tx.commitment_number(), commitment_tx.per_commitment_point(), logger)
+				ChannelMonitorUpdateStep::LatestCounterpartyCommitment {
+					commitment_txs, htlc_data,
+				} => {
+					log_trace!(logger, "Updating ChannelMonitor with {} latest counterparty commitments", commitment_txs.len());
+					if let Err(e) = self.update_counterparty_commitment_data(commitment_txs, htlc_data) {
+						log_error!(logger, "Failed updating latest counterparty commitment state: {}", e);
+						ret = Err(());
+					}
 				},
 				ChannelMonitorUpdateStep::PaymentPreimage { payment_preimage, payment_info } => {
 					log_trace!(logger, "Updating ChannelMonitor with payment preimage");
@@ -3934,7 +3987,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 				ChannelMonitorUpdateStep::LatestHolderCommitmentTXInfo { .. }
 					|ChannelMonitorUpdateStep::LatestHolderCommitment { .. }
 					|ChannelMonitorUpdateStep::LatestCounterpartyCommitmentTXInfo { .. }
-					|ChannelMonitorUpdateStep::LatestCounterpartyCommitmentTX { .. }
+					|ChannelMonitorUpdateStep::LatestCounterpartyCommitment { .. }
 					|ChannelMonitorUpdateStep::ShutdownScript { .. }
 					|ChannelMonitorUpdateStep::CommitmentSecret { .. }
 					|ChannelMonitorUpdateStep::RenegotiatedFunding { .. } =>
@@ -4094,7 +4147,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 		update.updates.iter().filter_map(|update| {
 			// Soon we will drop the first branch here in favor of the second.
 			// In preparation, we just add the second branch without deleting the first.
-			// Next step: in channel, switch channel monitor updates to use the `LatestCounterpartyCommitmentTX` variant.
+			// Next step: in channel, switch channel monitor updates to use the `LatestCounterpartyCommitment` variant.
 			match update {
 				&ChannelMonitorUpdateStep::LatestCounterpartyCommitmentTXInfo { commitment_txid,
 					ref htlc_outputs, commitment_number, their_per_commitment_point,
@@ -4112,19 +4165,17 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 
 					debug_assert_eq!(commitment_tx.trust().txid(), commitment_txid);
 
-					Some(commitment_tx)
+					Some(vec![commitment_tx])
 				},
-				&ChannelMonitorUpdateStep::LatestCounterpartyCommitmentTX {
-					htlc_outputs: _, ref commitment_tx,
-				} => {
-					Some(commitment_tx.clone())
+				&ChannelMonitorUpdateStep::LatestCounterpartyCommitment { ref commitment_txs, .. } => {
+					Some(commitment_txs.clone())
 				},
 				&ChannelMonitorUpdateStep::RenegotiatedFunding { ref counterparty_commitment_tx, .. } => {
-					Some(counterparty_commitment_tx.clone())
+					Some(vec![counterparty_commitment_tx.clone()])
 				},
 				_ => None,
 			}
-		}).collect()
+		}).flatten().collect()
 	}
 
 	#[rustfmt::skip]
@@ -6248,9 +6299,9 @@ mod tests {
 		monitor.provide_latest_holder_commitment_tx(dummy_commitment_tx.clone(),
 			&nondust_htlcs.iter().map(|htlc| (htlc.clone(), Some(dummy_sig), Some(dummy_source.clone()))).collect::<Vec<_>>());
 		monitor.provide_latest_counterparty_commitment_tx(Txid::from_byte_array(Sha256::hash(b"1").to_byte_array()),
-			preimages_slice_to_htlc_outputs!(preimages[5..15]), 281474976710655, dummy_key, &logger);
+			preimages_slice_to_htlc_outputs!(preimages[5..15]), 281474976710655, dummy_key);
 		monitor.provide_latest_counterparty_commitment_tx(Txid::from_byte_array(Sha256::hash(b"2").to_byte_array()),
-			preimages_slice_to_htlc_outputs!(preimages[15..20]), 281474976710654, dummy_key, &logger);
+			preimages_slice_to_htlc_outputs!(preimages[15..20]), 281474976710654, dummy_key);
 		for &(ref preimage, ref hash) in preimages.iter() {
 			let bounded_fee_estimator = LowerBoundedFeeEstimator::new(&fee_estimator);
 			monitor.provide_payment_preimage_unsafe_legacy(
@@ -6267,7 +6318,7 @@ mod tests {
 		test_preimages_exist!(&preimages[15..20], monitor);
 
 		monitor.provide_latest_counterparty_commitment_tx(Txid::from_byte_array(Sha256::hash(b"3").to_byte_array()),
-			preimages_slice_to_htlc_outputs!(preimages[17..20]), 281474976710653, dummy_key, &logger);
+			preimages_slice_to_htlc_outputs!(preimages[17..20]), 281474976710653, dummy_key);
 
 		// Now provide a further secret, pruning preimages 15-17
 		secret[0..32].clone_from_slice(&<Vec<u8>>::from_hex("c7518c8ae4660ed02894df8976fa1a3659c1a8b4b5bec0c4b872abeba4cb8964").unwrap());
@@ -6277,7 +6328,7 @@ mod tests {
 		test_preimages_exist!(&preimages[17..20], monitor);
 
 		monitor.provide_latest_counterparty_commitment_tx(Txid::from_byte_array(Sha256::hash(b"4").to_byte_array()),
-			preimages_slice_to_htlc_outputs!(preimages[18..20]), 281474976710652, dummy_key, &logger);
+			preimages_slice_to_htlc_outputs!(preimages[18..20]), 281474976710652, dummy_key);
 
 		// Now update holder commitment tx info, pruning only element 18 as we still care about the
 		// previous commitment tx's preimages too

lightning/src/ln/channel.rs

@@ -2647,7 +2647,7 @@ where
 			holder_commitment_tx, best_block, context.counterparty_node_id, context.channel_id(),
 		);
 		channel_monitor.provide_initial_counterparty_commitment_tx(
-			counterparty_initial_commitment_tx.clone(), logger,
+			counterparty_initial_commitment_tx.clone(),
 		);
 
 		self.context_mut().cur_counterparty_commitment_transaction_number -= 1;
@@ -10614,32 +10614,54 @@ where
 		}
 		self.context.resend_order = RAACommitmentOrder::RevokeAndACKFirst;
 
-		let mut updates = Vec::with_capacity(self.pending_funding.len() + 1);
-		for funding in core::iter::once(&self.funding).chain(self.pending_funding.iter()) {
+		let update = if self.pending_funding.is_empty() {
 			let (htlcs_ref, counterparty_commitment_tx) =
-				self.build_commitment_no_state_update(funding, logger);
-			let htlc_outputs: Vec<(HTLCOutputInCommitment, Option<Box<HTLCSource>>)> =
-				htlcs_ref.into_iter().map(|(htlc, htlc_source)| (htlc, htlc_source.map(|source_ref| Box::new(source_ref.clone())))).collect();
-
-			if self.pending_funding.is_empty() {
-				// Soon, we will switch this to `LatestCounterpartyCommitmentTX`,
-				// and provide the full commit tx instead of the information needed to rebuild it.
-				updates.push(ChannelMonitorUpdateStep::LatestCounterpartyCommitmentTXInfo {
-					commitment_txid: counterparty_commitment_tx.trust().txid(),
-					htlc_outputs,
-					commitment_number: self.context.cur_counterparty_commitment_transaction_number,
-					their_per_commitment_point: self.context.counterparty_cur_commitment_point.unwrap(),
-					feerate_per_kw: Some(counterparty_commitment_tx.feerate_per_kw()),
-					to_broadcaster_value_sat: Some(counterparty_commitment_tx.to_broadcaster_value_sat()),
-					to_countersignatory_value_sat: Some(counterparty_commitment_tx.to_countersignatory_value_sat()),
-				});
-			} else {
-				updates.push(ChannelMonitorUpdateStep::LatestCounterpartyCommitmentTX {
-					htlc_outputs,
-					commitment_tx: counterparty_commitment_tx,
-				});
+				self.build_commitment_no_state_update(&self.funding, logger);
+			let htlc_outputs = htlcs_ref.into_iter()
+				.map(|(htlc, htlc_source)| (
+					htlc, htlc_source.map(|source_ref| Box::new(source_ref.clone()))
+				))
+				.collect();
+
+			// Soon, we will switch this to `LatestCounterpartyCommitment`,
+			// and provide the full commit tx instead of the information needed to rebuild it.
+			ChannelMonitorUpdateStep::LatestCounterpartyCommitmentTXInfo {
+				commitment_txid: counterparty_commitment_tx.trust().txid(),
+				htlc_outputs,
+				commitment_number: self.context.cur_counterparty_commitment_transaction_number,
+				their_per_commitment_point: self.context.counterparty_cur_commitment_point.unwrap(),
+				feerate_per_kw: Some(counterparty_commitment_tx.feerate_per_kw()),
+				to_broadcaster_value_sat: Some(counterparty_commitment_tx.to_broadcaster_value_sat()),
+				to_countersignatory_value_sat: Some(counterparty_commitment_tx.to_countersignatory_value_sat()),
 			}
-		}
+		} else {
+			let mut htlc_data = None;
+			let commitment_txs = core::iter::once(&self.funding)
+				.chain(self.pending_funding.iter())
+				.map(|funding| {
+					let (htlcs_ref, counterparty_commitment_tx) =
+						self.build_commitment_no_state_update(funding, logger);
+					if htlc_data.is_none() {
+						let nondust_htlc_sources = htlcs_ref.iter()
+							// We check !offered as this is the HTLC from the counterparty's point of view.
+							.filter(|(htlc, _)| !htlc.offered && htlc.transaction_output_index.is_some())
+							.map(|(_, source)| source.expect("Outbound HTLC must have a source").clone())
+							.collect();
+						let dust_htlcs = htlcs_ref.into_iter()
+							.filter(|(htlc, _)| htlc.transaction_output_index.is_none())
+							.map(|(htlc, source)| (htlc, source.cloned()))
+							.collect();
+						htlc_data = Some(CommitmentHTLCData {
+							nondust_htlc_sources,
+							dust_htlcs,
+						});
+					}
+					counterparty_commitment_tx
+				})
+				.collect();
+			let htlc_data = htlc_data.unwrap();
+			ChannelMonitorUpdateStep::LatestCounterpartyCommitment { commitment_txs, htlc_data }
+		};
 
 		if self.context.announcement_sigs_state == AnnouncementSigsState::MessageSent {
 			self.context.announcement_sigs_state = AnnouncementSigsState::Committed;
@@ -10648,7 +10670,7 @@ where
 		self.context.latest_monitor_update_id += 1;
 		let monitor_update = ChannelMonitorUpdate {
 			update_id: self.context.latest_monitor_update_id,
-			updates,
+			updates: vec![update],
 			channel_id: Some(self.context.channel_id()),
 		};
 		self.context.channel_state.set_awaiting_remote_revoke();