Generate a PaymentForwarded event when a forwarded HTLC is claimed

It is useful for accounting and informational reasons for users to
be informed when a payment has been successfully forwarded. Thus,
when an HTLC which represents a forwarded leg is claimed, we
generate a new `PaymentForwarded` event.

This requires some additional plumbing to return HTLC values from
`OnchainEvent`s. Further, when we have to go on-chain to claim the
inbound side of the payment, we do not inform the user of the fee
reward, as we cannot calculate it until we see what is confirmed
on-chain.

Substantial code structure rewrites by:
Valentine Wallace <[email protected]>

Author: TheBlueMatt

CHANGELOG.md

@@ -1,3 +1,17 @@
+# 0.0.100 - WIP
+
+## Serialization Compatibility
+ * HTLCs which were in the process of being claimed on-chain when a pre-0.0.100
+   `ChannelMonitor` was serialized may generate `PaymentForwarded` events with
+   spurious `fee_earned_msat` values. This only applies to payments which were
+   unresolved at the time of the upgrade.
+ * 0.0.100 clients with pending PaymentForwarded events at serialization-time
+   will generate serialized `ChannelManager` objects which 0.0.99 and earlier
+   clients cannot read. The likelihood of this can be reduced by ensuring you
+   process all pending events immediately before serialization (as is done by
+   the `lightning-background-processor` crate).
+
+
 # 0.0.99 - 2021-07-09
 
 ## API Updates

fuzz/src/chanmon_consistency.rs

@@ -805,6 +805,7 @@ pub fn do_test<Out: test_logger::Output>(data: &[u8], out: Out) {
 						},
 						events::Event::PaymentSent { .. } => {},
 						events::Event::PaymentFailed { .. } => {},
+						events::Event::PaymentForwarded { .. } if $node == 1 => {},
 						events::Event::PendingHTLCsForwardable { .. } => {
 							nodes[$node].process_pending_htlc_forwards();
 						},

fuzz/src/full_stack.rs

@@ -596,12 +596,10 @@ pub fn do_test(data: &[u8], logger: &Arc<dyn Logger>) {
 					//TODO: enhance by fetching random amounts from fuzz input?
 					payments_received.push(payment_hash);
 				},
-				Event::PaymentSent {..} => {},
-				Event::PaymentFailed {..} => {},
 				Event::PendingHTLCsForwardable {..} => {
 					should_forward = true;
 				},
-				Event::SpendableOutputs {..} => {},
+				_ => {},
 			}
 		}
 	}

lightning/src/chain/channelmonitor.rs

@@ -199,10 +199,12 @@ pub enum MonitorEvent {
 pub struct HTLCUpdate {
 	pub(crate) payment_hash: PaymentHash,
 	pub(crate) payment_preimage: Option<PaymentPreimage>,
-	pub(crate) source: HTLCSource
+	pub(crate) source: HTLCSource,
+	pub(crate) onchain_value_satoshis: Option<u64>,
 }
 impl_writeable_tlv_based!(HTLCUpdate, {
 	(0, payment_hash, required),
+	(1, onchain_value_satoshis, option),
 	(2, source, required),
 	(4, payment_preimage, option),
 });
@@ -385,6 +387,7 @@ enum OnchainEvent {
 	HTLCUpdate {
 		source: HTLCSource,
 		payment_hash: PaymentHash,
+		onchain_value_satoshis: Option<u64>,
 	},
 	MaturingOutput {
 		descriptor: SpendableOutputDescriptor,
@@ -400,6 +403,7 @@ impl_writeable_tlv_based!(OnchainEventEntry, {
 impl_writeable_tlv_based_enum!(OnchainEvent,
 	(0, HTLCUpdate) => {
 		(0, source, required),
+		(1, onchain_value_satoshis, option),
 		(2, payment_hash, required),
 	},
 	(1, MaturingOutput) => {
@@ -1574,6 +1578,7 @@ impl<Signer: Sign> ChannelMonitorImpl<Signer> {
 										event: OnchainEvent::HTLCUpdate {
 											source: (**source).clone(),
 											payment_hash: htlc.payment_hash.clone(),
+											onchain_value_satoshis: Some(htlc.amount_msat / 1000),
 										},
 									};
 									log_info!(logger, "Failing HTLC with payment_hash {} from {} counterparty commitment tx due to broadcast of revoked counterparty commitment transaction, waiting for confirmation (at height {})", log_bytes!(htlc.payment_hash.0), $commitment_tx, entry.confirmation_threshold());
@@ -1641,6 +1646,7 @@ impl<Signer: Sign> ChannelMonitorImpl<Signer> {
 									event: OnchainEvent::HTLCUpdate {
 										source: (**source).clone(),
 										payment_hash: htlc.payment_hash.clone(),
+										onchain_value_satoshis: Some(htlc.amount_msat / 1000),
 									},
 								});
 							}
@@ -1825,6 +1831,7 @@ impl<Signer: Sign> ChannelMonitorImpl<Signer> {
 								height,
 								event: OnchainEvent::HTLCUpdate {
 									source: source.clone(), payment_hash: htlc.payment_hash,
+									onchain_value_satoshis: Some(htlc.amount_msat / 1000)
 								},
 							};
 							log_trace!(logger, "Failing HTLC with payment_hash {} from {} holder commitment tx due to broadcast of transaction, waiting confirmation (at height{})",
@@ -2087,7 +2094,7 @@ impl<Signer: Sign> ChannelMonitorImpl<Signer> {
 		// Produce actionable events from on-chain events having reached their threshold.
 		for entry in onchain_events_reaching_threshold_conf.drain(..) {
 			match entry.event {
-				OnchainEvent::HTLCUpdate { ref source, payment_hash } => {
+				OnchainEvent::HTLCUpdate { ref source, payment_hash, onchain_value_satoshis } => {
 					// Check for duplicate HTLC resolutions.
 					#[cfg(debug_assertions)]
 					{
@@ -2106,9 +2113,10 @@ impl<Signer: Sign> ChannelMonitorImpl<Signer> {
 
 					log_debug!(logger, "HTLC {} failure update has got enough confirmations to be passed upstream", log_bytes!(payment_hash.0));
 					self.pending_monitor_events.push(MonitorEvent::HTLCEvent(HTLCUpdate {
-						payment_hash: payment_hash,
+						payment_hash,
 						payment_preimage: None,
 						source: source.clone(),
+						onchain_value_satoshis,
 					}));
 				},
 				OnchainEvent::MaturingOutput { descriptor } => {
@@ -2325,7 +2333,7 @@ impl<Signer: Sign> ChannelMonitorImpl<Signer> {
 							if pending_htlc.payment_hash == $htlc_output.payment_hash && pending_htlc.amount_msat == $htlc_output.amount_msat {
 								if let &Some(ref source) = pending_source {
 									log_claim!("revoked counterparty commitment tx", false, pending_htlc, true);
-									payment_data = Some(((**source).clone(), $htlc_output.payment_hash));
+									payment_data = Some(((**source).clone(), $htlc_output.payment_hash, $htlc_output.amount_msat));
 									break;
 								}
 							}
@@ -2345,7 +2353,7 @@ impl<Signer: Sign> ChannelMonitorImpl<Signer> {
 								// transaction. This implies we either learned a preimage, the HTLC
 								// has timed out, or we screwed up. In any case, we should now
 								// resolve the source HTLC with the original sender.
-								payment_data = Some(((*source).clone(), htlc_output.payment_hash));
+								payment_data = Some(((*source).clone(), htlc_output.payment_hash, htlc_output.amount_msat));
 							} else if !$holder_tx {
 									check_htlc_valid_counterparty!(self.current_counterparty_commitment_txid, htlc_output);
 								if payment_data.is_none() {
@@ -2378,7 +2386,7 @@ impl<Signer: Sign> ChannelMonitorImpl<Signer> {
 
 			// Check that scan_commitment, above, decided there is some source worth relaying an
 			// HTLC resolution backwards to and figure out whether we learned a preimage from it.
-			if let Some((source, payment_hash)) = payment_data {
+			if let Some((source, payment_hash, amount_msat)) = payment_data {
 				let mut payment_preimage = PaymentPreimage([0; 32]);
 				if accepted_preimage_claim {
 					if !self.pending_monitor_events.iter().any(
@@ -2387,7 +2395,8 @@ impl<Signer: Sign> ChannelMonitorImpl<Signer> {
 						self.pending_monitor_events.push(MonitorEvent::HTLCEvent(HTLCUpdate {
 							source,
 							payment_preimage: Some(payment_preimage),
-							payment_hash
+							payment_hash,
+							onchain_value_satoshis: Some(amount_msat / 1000),
 						}));
 					}
 				} else if offered_preimage_claim {
@@ -2399,7 +2408,8 @@ impl<Signer: Sign> ChannelMonitorImpl<Signer> {
 						self.pending_monitor_events.push(MonitorEvent::HTLCEvent(HTLCUpdate {
 							source,
 							payment_preimage: Some(payment_preimage),
-							payment_hash
+							payment_hash,
+							onchain_value_satoshis: Some(amount_msat / 1000),
 						}));
 					}
 				} else {
@@ -2415,7 +2425,10 @@ impl<Signer: Sign> ChannelMonitorImpl<Signer> {
 					let entry = OnchainEventEntry {
 						txid: tx.txid(),
 						height,
-						event: OnchainEvent::HTLCUpdate { source: source, payment_hash: payment_hash },
+						event: OnchainEvent::HTLCUpdate {
+							source, payment_hash,
+							onchain_value_satoshis: Some(amount_msat / 1000),
+						},
 					};
 					log_info!(logger, "Failing HTLC with payment_hash {} timeout by a spend tx, waiting for confirmation (at height {})", log_bytes!(payment_hash.0), entry.confirmation_threshold());
 					self.onchain_events_awaiting_threshold_conf.push(entry);

lightning/src/ln/chanmon_update_fail_tests.rs

@@ -1159,6 +1159,7 @@ fn test_monitor_update_fail_reestablish() {
 	assert!(updates.update_fee.is_none());
 	assert_eq!(updates.update_fulfill_htlcs.len(), 1);
 	nodes[1].node.handle_update_fulfill_htlc(&nodes[2].node.get_our_node_id(), &updates.update_fulfill_htlcs[0]);
+	expect_payment_forwarded!(nodes[1], Some(1000), false);
 	check_added_monitors!(nodes[1], 1);
 	assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
 	commitment_signed_dance!(nodes[1], nodes[2], updates.commitment_signed, false);
@@ -2317,6 +2318,7 @@ fn do_test_reconnect_dup_htlc_claims(htlc_status: HTLCStatusAtDupClaim, second_f
 		assert_eq!(fulfill_msg, cs_updates.update_fulfill_htlcs[0]);
 	}
 	nodes[1].node.handle_update_fulfill_htlc(&nodes[2].node.get_our_node_id(), &fulfill_msg);
+	expect_payment_forwarded!(nodes[1], Some(1000), false);
 	check_added_monitors!(nodes[1], 1);
 
 	let mut bs_updates = None;

lightning/src/ln/channel.rs

@@ -306,6 +306,7 @@ pub struct CounterpartyForwardingInfo {
 enum UpdateFulfillFetch {
 	NewClaim {
 		monitor_update: ChannelMonitorUpdate,
+		htlc_value_msat: u64,
 		msg: Option<msgs::UpdateFulfillHTLC>,
 	},
 	DuplicateClaim {},
@@ -319,6 +320,8 @@ pub enum UpdateFulfillCommitFetch {
 	NewClaim {
 		/// The ChannelMonitorUpdate which places the new payment preimage in the channel monitor
 		monitor_update: ChannelMonitorUpdate,
+		/// The value of the HTLC which was claimed, in msat.
+		htlc_value_msat: u64,
 		/// The update_fulfill message and commitment_signed message (if the claim was not placed
 		/// in the holding cell).
 		msgs: Option<(msgs::UpdateFulfillHTLC, msgs::CommitmentSigned)>,
@@ -336,6 +339,9 @@ pub enum UpdateFulfillCommitFetch {
 // Holder designates channel data owned for the benefice of the user client.
 // Counterparty designates channel data owned by the another channel participant entity.
 pub(super) struct Channel<Signer: Sign> {
+	#[cfg(any(test, feature = "_test_utils"))]
+	pub(crate) config: ChannelConfig,
+	#[cfg(not(any(test, feature = "_test_utils")))]
 	config: ChannelConfig,
 
 	user_id: u64,
@@ -1275,6 +1281,7 @@ impl<Signer: Sign> Channel<Signer> {
 		// these, but for now we just have to treat them as normal.
 
 		let mut pending_idx = core::usize::MAX;
+		let mut htlc_value_msat = 0;
 		for (idx, htlc) in self.pending_inbound_htlcs.iter().enumerate() {
 			if htlc.htlc_id == htlc_id_arg {
 				assert_eq!(htlc.payment_hash, payment_hash_calc);
@@ -1294,6 +1301,7 @@ impl<Signer: Sign> Channel<Signer> {
 					}
 				}
 				pending_idx = idx;
+				htlc_value_msat = htlc.amount_msat;
 				break;
 			}
 		}
@@ -1335,7 +1343,7 @@ impl<Signer: Sign> Channel<Signer> {
 							// TODO: We may actually be able to switch to a fulfill here, though its
 							// rare enough it may not be worth the complexity burden.
 							debug_assert!(false, "Tried to fulfill an HTLC that was already failed");
-							return UpdateFulfillFetch::NewClaim { monitor_update, msg: None };
+							return UpdateFulfillFetch::NewClaim { monitor_update, htlc_value_msat, msg: None };
 						}
 					},
 					_ => {}
@@ -1347,7 +1355,7 @@ impl<Signer: Sign> Channel<Signer> {
 			});
 			#[cfg(any(test, feature = "fuzztarget"))]
 			self.historical_inbound_htlc_fulfills.insert(htlc_id_arg);
-			return UpdateFulfillFetch::NewClaim { monitor_update, msg: None };
+			return UpdateFulfillFetch::NewClaim { monitor_update, htlc_value_msat, msg: None };
 		}
 		#[cfg(any(test, feature = "fuzztarget"))]
 		self.historical_inbound_htlc_fulfills.insert(htlc_id_arg);
@@ -1357,14 +1365,15 @@ impl<Signer: Sign> Channel<Signer> {
 			if let InboundHTLCState::Committed = htlc.state {
 			} else {
 				debug_assert!(false, "Have an inbound HTLC we tried to claim before it was fully committed to");
-				return UpdateFulfillFetch::NewClaim { monitor_update, msg: None };
+				return UpdateFulfillFetch::NewClaim { monitor_update, htlc_value_msat, msg: None };
 			}
 			log_trace!(logger, "Upgrading HTLC {} to LocalRemoved with a Fulfill in channel {}!", log_bytes!(htlc.payment_hash.0), log_bytes!(self.channel_id));
 			htlc.state = InboundHTLCState::LocalRemoved(InboundHTLCRemovalReason::Fulfill(payment_preimage_arg.clone()));
 		}
 
 		UpdateFulfillFetch::NewClaim {
 			monitor_update,
+			htlc_value_msat,
 			msg: Some(msgs::UpdateFulfillHTLC {
 				channel_id: self.channel_id(),
 				htlc_id: htlc_id_arg,
@@ -1375,7 +1384,7 @@ impl<Signer: Sign> Channel<Signer> {
 
 	pub fn get_update_fulfill_htlc_and_commit<L: Deref>(&mut self, htlc_id: u64, payment_preimage: PaymentPreimage, logger: &L) -> Result<UpdateFulfillCommitFetch, (ChannelError, ChannelMonitorUpdate)> where L::Target: Logger {
 		match self.get_update_fulfill_htlc(htlc_id, payment_preimage, logger) {
-			UpdateFulfillFetch::NewClaim { mut monitor_update, msg: Some(update_fulfill_htlc) } => {
+			UpdateFulfillFetch::NewClaim { mut monitor_update, htlc_value_msat, msg: Some(update_fulfill_htlc) } => {
 				let (commitment, mut additional_update) = match self.send_commitment_no_status_check(logger) {
 					Err(e) => return Err((e, monitor_update)),
 					Ok(res) => res
@@ -1384,9 +1393,10 @@ impl<Signer: Sign> Channel<Signer> {
 				// strictly increasing by one, so decrement it here.
 				self.latest_monitor_update_id = monitor_update.update_id;
 				monitor_update.updates.append(&mut additional_update.updates);
-				Ok(UpdateFulfillCommitFetch::NewClaim { monitor_update, msgs: Some((update_fulfill_htlc, commitment)) })
+				Ok(UpdateFulfillCommitFetch::NewClaim { monitor_update, htlc_value_msat, msgs: Some((update_fulfill_htlc, commitment)) })
 			},
-			UpdateFulfillFetch::NewClaim { monitor_update, msg: None } => Ok(UpdateFulfillCommitFetch::NewClaim { monitor_update, msgs: None }),
+			UpdateFulfillFetch::NewClaim { monitor_update, htlc_value_msat, msg: None } =>
+				Ok(UpdateFulfillCommitFetch::NewClaim { monitor_update, htlc_value_msat, msgs: None }),
 			UpdateFulfillFetch::DuplicateClaim {} => Ok(UpdateFulfillCommitFetch::DuplicateClaim {}),
 		}
 	}
@@ -2163,7 +2173,7 @@ impl<Signer: Sign> Channel<Signer> {
 
 	/// Marks an outbound HTLC which we have received update_fail/fulfill/malformed
 	#[inline]
-	fn mark_outbound_htlc_removed(&mut self, htlc_id: u64, check_preimage: Option<PaymentHash>, fail_reason: Option<HTLCFailReason>) -> Result<&HTLCSource, ChannelError> {
+	fn mark_outbound_htlc_removed(&mut self, htlc_id: u64, check_preimage: Option<PaymentHash>, fail_reason: Option<HTLCFailReason>) -> Result<&OutboundHTLCOutput, ChannelError> {
 		for htlc in self.pending_outbound_htlcs.iter_mut() {
 			if htlc.htlc_id == htlc_id {
 				match check_preimage {
@@ -2182,13 +2192,13 @@ impl<Signer: Sign> Channel<Signer> {
 					OutboundHTLCState::AwaitingRemoteRevokeToRemove(_) | OutboundHTLCState::AwaitingRemovedRemoteRevoke(_) | OutboundHTLCState::RemoteRemoved(_) =>
 						return Err(ChannelError::Close(format!("Remote tried to fulfill/fail HTLC ({}) that they'd already fulfilled/failed", htlc_id))),
 				}
-				return Ok(&htlc.source);
+				return Ok(htlc);
 			}
 		}
 		Err(ChannelError::Close("Remote tried to fulfill/fail an HTLC we couldn't find".to_owned()))
 	}
 
-	pub fn update_fulfill_htlc(&mut self, msg: &msgs::UpdateFulfillHTLC) -> Result<HTLCSource, ChannelError> {
+	pub fn update_fulfill_htlc(&mut self, msg: &msgs::UpdateFulfillHTLC) -> Result<(HTLCSource, u64), ChannelError> {
 		if (self.channel_state & (ChannelState::ChannelFunded as u32)) != (ChannelState::ChannelFunded as u32) {
 			return Err(ChannelError::Close("Got fulfill HTLC message when channel was not in an operational state".to_owned()));
 		}
@@ -2197,7 +2207,7 @@ impl<Signer: Sign> Channel<Signer> {
 		}
 
 		let payment_hash = PaymentHash(Sha256::hash(&msg.payment_preimage.0[..]).into_inner());
-		self.mark_outbound_htlc_removed(msg.htlc_id, Some(payment_hash), None).map(|source| source.clone())
+		self.mark_outbound_htlc_removed(msg.htlc_id, Some(payment_hash), None).map(|htlc| (htlc.source.clone(), htlc.amount_msat))
 	}
 
 	pub fn update_fail_htlc(&mut self, msg: &msgs::UpdateFailHTLC, fail_reason: HTLCFailReason) -> Result<(), ChannelError> {
@@ -2496,7 +2506,7 @@ impl<Signer: Sign> Channel<Signer> {
 						// in it hitting the holding cell again and we cannot change the state of a
 						// holding cell HTLC from fulfill to anything else.
 						let (update_fulfill_msg_option, mut additional_monitor_update) =
-							if let UpdateFulfillFetch::NewClaim { msg, monitor_update } = self.get_update_fulfill_htlc(htlc_id, *payment_preimage, logger) {
+							if let UpdateFulfillFetch::NewClaim { msg, monitor_update, .. } = self.get_update_fulfill_htlc(htlc_id, *payment_preimage, logger) {
 								(msg, monitor_update)
 							} else { unreachable!() };
 						update_fulfill_htlcs.push(update_fulfill_msg_option.unwrap());