Reload pending payments from ChannelMonitor HTLC data on reload

If we go to send a payment, add the HTLC(s) to the channel(s),
commit the ChannelMonitor updates to disk, and then crash, we'll
come back up with no pending payments but HTLC(s) ready to be
claim/failed.

This makes it rather impractical to write a payment sender/retryer,
as you cannot guarantee atomicity - you cannot guarantee you'll
have retry data persisted even if the HTLC(s) are actually pending.

Because ChannelMonitors are *the* atomically-persisted data in LDK,
we lean on their current HTLC data to figure out what HTLC(s) are a
part of an outbound payment, rebuilding the pending payments list
on reload.

Author: TheBlueMatt

lightning/src/chain/channelmonitor.rs

@@ -1527,6 +1527,77 @@ impl<Signer: Sign> ChannelMonitor<Signer> {
 
 		res
 	}
+
+	/// Gets the set of outbound HTLCs which are pending resolution in this channel.
+	/// This is used to reconstruct pending outbound payments on restart in the ChannelManager.
+	pub(crate) fn get_pending_htlcs(&self) -> HashMap<HTLCSource, HTLCOutputInCommitment> {
+		let mut res = HashMap::new();
+		let us = self.inner.lock().unwrap();
+
+		macro_rules! walk_htlcs {
+			($holder_commitment: expr, $htlc_iter: expr) => {
+				for (htlc, source) in $htlc_iter {
+					if us.htlcs_resolved_on_chain.iter().any(|v| Some(v.input_idx) == htlc.transaction_output_index) {
+						assert!(us.funding_spend_confirmed.is_some());
+					} else if htlc.offered == $holder_commitment {
+						// If the payment was outbound, check if there's an HTLCUpdate
+						// indicating we have spent this HTLC with a timeout, claiming it back
+						// and awaiting confirmations on it.
+						let htlc_update_confd = us.onchain_events_awaiting_threshold_conf.iter().find_map(|event| {
+							if let OnchainEvent::HTLCUpdate { input_idx: Some(input_idx), .. } = event.event {
+								if Some(input_idx) == htlc.transaction_output_index &&
+									us.best_block.height() >= event.height + ANTI_REORG_DELAY - 1
+								{ Some(()) } else { None }
+							} else { None }
+						});
+						if htlc_update_confd.is_none() {
+							res.insert(source.clone(), htlc.clone());
+						}
+					}
+				}
+			}
+		}
+
+		if let Some(txid) = us.funding_spend_confirmed {
+			if Some(txid) == us.current_counterparty_commitment_txid || Some(txid) == us.prev_counterparty_commitment_txid {
+				walk_htlcs!(false, us.counterparty_claimable_outpoints.get(&txid).unwrap().iter().find_map(|(a, b)| {
+					if let &Some(ref source) = b {
+						Some((a, &**source))
+					} else { None }
+				}));
+			} else if txid == us.current_holder_commitment_tx.txid {
+				walk_htlcs!(true, us.current_holder_commitment_tx.htlc_outputs.iter().find_map(|(a, _, c)| {
+					if let Some(source) = c { Some((a, source)) } else { None }
+				}));
+			} else if let Some(prev_commitment) = &us.prev_holder_signed_commitment_tx {
+				if txid == prev_commitment.txid {
+					walk_htlcs!(true, prev_commitment.htlc_outputs.iter().find_map(|(a, _, c)| {
+						if let Some(source) = c { Some((a, source)) } else { None }
+					}));
+				}
+			}
+		} else {
+			macro_rules! check_htlc_fails {
+				($txid: expr, $commitment_tx: expr) => {
+					if let Some(ref latest_outpoints) = us.counterparty_claimable_outpoints.get($txid) {
+						for &(ref htlc, ref source_option) in latest_outpoints.iter() {
+							if let &Some(ref source) = source_option {
+								res.insert((**source).clone(), htlc.clone());
+							}
+						}
+					}
+				}
+			}
+			if let Some(ref txid) = us.current_counterparty_commitment_txid {
+				check_htlc_fails!(txid, "current");
+			}
+			if let Some(ref txid) = us.prev_counterparty_commitment_txid {
+				check_htlc_fails!(txid, "previous");
+			}
+		}
+
+		res
+	}
 }
 
 /// Compares a broadcasted commitment transaction's HTLCs with those in the latest state,

lightning/src/ln/channelmanager.rs

@@ -145,7 +145,7 @@ pub(super) enum HTLCForwardInfo {
 }
 
 /// Tracks the inbound corresponding to an outbound HTLC
-#[derive(Clone, PartialEq)]
+#[derive(Clone, Hash, PartialEq, Eq)]
 pub(crate) struct HTLCPreviousHopData {
 	short_channel_id: u64,
 	htlc_id: u64,
@@ -189,7 +189,7 @@ impl Readable for PaymentId {
 	}
 }
 /// Tracks the inbound corresponding to an outbound HTLC
-#[derive(Clone, PartialEq)]
+#[derive(Clone, PartialEq, Eq)]
 pub(crate) enum HTLCSource {
 	PreviousHopData(HTLCPreviousHopData),
 	OutboundRoute {
@@ -202,6 +202,23 @@ pub(crate) enum HTLCSource {
 		payment_secret: Option<PaymentSecret>,
 	},
 }
+impl core::hash::Hash for HTLCSource {
+	fn hash<H: core::hash::Hasher>(&self, hasher: &mut H) {
+		match self {
+			HTLCSource::PreviousHopData(prev_hop_data) => {
+				0u8.hash(hasher);
+				prev_hop_data.hash(hasher);
+			},
+			HTLCSource::OutboundRoute { path, session_priv, payment_id, payment_secret, first_hop_htlc_msat: _ } => {
+				1u8.hash(hasher);
+				path.hash(hasher);
+				session_priv[..].hash(hasher);
+				payment_id.hash(hasher);
+				payment_secret.hash(hasher);
+			},
+		}
+	}
+}
 #[cfg(test)]
 impl HTLCSource {
 	pub fn dummy() -> Self {
@@ -5833,6 +5850,34 @@ impl<'a, Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 				outbounds.insert(id, PendingOutboundPayment::Legacy { session_privs });
 			}
 			pending_outbound_payments = Some(outbounds);
+		} else {
+			// We only rebuild the pending payments map if we were most recently serialized by
+			// 0.0.102+
+			for (_, monitor) in args.channel_monitors {
+				for (htlc_source, htlc) in monitor.get_pending_htlcs() {
+					if let HTLCSource::OutboundRoute { payment_id, session_priv, path, payment_secret, .. } = htlc_source {
+						if path.is_empty() { return Err(DecodeError::InvalidValue); }
+						let path_amt = path.last().unwrap().fee_msat;
+						let mut session_priv_bytes = [0; 32];
+						session_priv_bytes[..].copy_from_slice(&session_priv[..]);
+						match pending_outbound_payments.as_mut().unwrap().entry(payment_id) {
+							hash_map::Entry::Occupied(mut entry) => {
+								entry.get_mut().insert(session_priv_bytes, path_amt);
+							},
+							hash_map::Entry::Vacant(entry) => {
+								entry.insert(PendingOutboundPayment::Retryable {
+									session_privs: [session_priv_bytes].iter().map(|a| *a).collect(),
+									payment_hash: htlc.payment_hash,
+									payment_secret,
+									pending_amt_msat: path_amt,
+									total_msat: path_amt,
+									starting_block_height: best_block_height,
+								});
+							}
+						}
+					}
+				}
+			}
 		}
 
 		let mut secp_ctx = Secp256k1::new();