Re-fail perm-failed HTLCs on startup in case of MonitorEvent loss

`MonitorEvent`s aren't delivered to the `ChannelManager` in a
durable fashion - if the `ChannelManager` fetches the pending
`MonitorEvent`s, then the `ChannelMonitor` gets persisted (i.e. due
to a block update) then the node crashes, prior to persisting the
`ChannelManager` again, the `MonitorEvent` and its effects on the
`ChannelManger` will be lost. This isn't likely in a sync persist
environment, but in an async one this could be an issue.

Note that this is only an issue for closed channels -
`MonitorEvent`s only inform the `ChannelManager` that a channel is
closed (which the `ChannelManager` will learn on startup or when it
next tries to advance the channel state), that
`ChannelMonitorUpdate` writes completed (which the `ChannelManager`
will detect on startup), or that HTLCs resolved on-chain post
closure. Of the three, only the last is problematic to lose prior
to a reload.

In a previous commit we handled the case of claimed HTLCs by
replaying payment preimages on startup to avoid `MonitorEvent` loss
causing us to miss an HTLC claim. Here we handle the HTLC-failed
case similarly.

Unlike with HTLC claims via preimage, we don't already have replay
logic in `ChannelManager` startup, but its easy enough to add one.
Luckily, we already track when an HTLC reaches permanently-failed
state in `ChannelMonitor` (i.e. it has `ANTI_REORG_DELAY`
confirmations on-chain on the failing transaction), so all we need
to do is add the ability to query for that and fail them on
`ChannelManager` startup.

Author: TheBlueMatt

lightning/src/chain/channelmonitor.rs

@@ -2983,6 +2983,125 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 		res
 	}
 
+	/// Gets the set of outbound HTLCs which hit the chain and ultimately were claimed by us via
+	/// the timeout path and reached [`ANTI_REORG_DELAY`] confirmations. This is used to determine
+	/// if an HTLC has failed without the `ChannelManager` having seen it prior to being persisted.
+	pub(crate) fn get_onchain_failed_outbound_htlcs(&self) -> HashMap<HTLCSource, PaymentHash> {
+		let mut res = new_hash_map();
+		let us = self.inner.lock().unwrap();
+
+		// We only want HTLCs with ANTI_REORG_DELAY confirmations, which implies the commitment
+		// transaction has least ANTI_REORG_DELAY confirmations for any dependent HTLC transactions
+		// to have been confirmed.
+		let confirmed_txid = us.funding_spend_confirmed.or_else(|| {
+			us.onchain_events_awaiting_threshold_conf.iter().find_map(|event| {
+				if let OnchainEvent::FundingSpendConfirmation { .. } = event.event {
+					if event.height <= us.best_block.height - ANTI_REORG_DELAY + 1 {
+						Some(event.txid)
+					} else {
+						None
+					}
+				} else {
+					None
+				}
+			})
+		});
+
+		let confirmed_txid = if let Some(txid) = confirmed_txid {
+			txid
+		} else {
+			return res;
+		};
+
+		macro_rules! walk_htlcs {
+			($holder_commitment: expr, $htlc_iter: expr) => {
+				let mut walk_candidate_htlcs = |htlcs| {
+					for &(ref candidate_htlc, ref candidate_source) in htlcs {
+						let candidate_htlc: &HTLCOutputInCommitment = &candidate_htlc;
+						let candidate_source: &Option<Box<HTLCSource>> = &candidate_source;
+
+						let source: &HTLCSource = if let Some(source) = candidate_source {
+							source
+						} else {
+							continue;
+						};
+						let confirmed = $htlc_iter.find(|(_, conf_src)| Some(source) == *conf_src);
+						if let Some((confirmed_htlc, _)) = confirmed {
+							let filter = |v: &&IrrevocablyResolvedHTLC| {
+								v.commitment_tx_output_idx
+									== confirmed_htlc.transaction_output_index
+							};
+
+							// The HTLC was included in the confirmed commitment transaction, so we
+							// need to see if it has been irrevocably failed yet.
+							if confirmed_htlc.transaction_output_index.is_none() {
+								// Dust HTLCs are always implicitly failed once the commitment
+								// transaction reaches ANTI_REORG_DELAY confirmations.
+								res.insert(source.clone(), confirmed_htlc.payment_hash);
+							} else if let Some(state) =
+								us.htlcs_resolved_on_chain.iter().filter(filter).next()
+							{
+								if state.payment_preimage.is_none() {
+									res.insert(source.clone(), confirmed_htlc.payment_hash);
+								}
+							}
+						} else {
+							// The HTLC was not included in the confirmed commitment transaction,
+							// which has now reached ANTI_REORG_DELAY confirmations and thus the
+							// HTLC has been failed.
+							res.insert(source.clone(), candidate_htlc.payment_hash);
+						}
+					}
+				};
+
+				// We walk the set of HTLCs in the unrevoked counterparty commitment transactions (see
+				// `fail_unbroadcast_htlcs` for a description of why).
+				if let Some(ref txid) = us.funding.current_counterparty_commitment_txid {
+					if let Some(htlcs) = us.funding.counterparty_claimable_outpoints.get(txid) {
+						walk_candidate_htlcs(htlcs);
+					}
+				}
+				if let Some(ref txid) = us.funding.prev_counterparty_commitment_txid {
+					if let Some(htlcs) = us.funding.counterparty_claimable_outpoints.get(txid) {
+						walk_candidate_htlcs(htlcs);
+					}
+				}
+			};
+		}
+
+		let funding = get_confirmed_funding_scope!(us);
+
+		if Some(confirmed_txid) == funding.current_counterparty_commitment_txid
+			|| Some(confirmed_txid) == funding.prev_counterparty_commitment_txid
+		{
+			let htlcs = funding.counterparty_claimable_outpoints.get(&confirmed_txid).unwrap();
+			walk_htlcs!(
+				false,
+				htlcs.iter().filter_map(|(a, b)| {
+					if let &Some(ref source) = b {
+						Some((a, Some(&**source)))
+					} else {
+						None
+					}
+				})
+			);
+		} else if confirmed_txid == funding.current_holder_commitment_tx.trust().txid() {
+			walk_htlcs!(true, holder_commitment_htlcs!(us, CURRENT_WITH_SOURCES));
+		} else if let Some(prev_commitment_tx) = &funding.prev_holder_commitment_tx {
+			if confirmed_txid == prev_commitment_tx.trust().txid() {
+				walk_htlcs!(true, holder_commitment_htlcs!(us, PREV_WITH_SOURCES).unwrap());
+			} else {
+				let htlcs_confirmed: &[(&HTLCOutputInCommitment, _)] = &[];
+				walk_htlcs!(false, htlcs_confirmed.iter());
+			}
+		} else {
+			let htlcs_confirmed: &[(&HTLCOutputInCommitment, _)] = &[];
+			walk_htlcs!(false, htlcs_confirmed.iter());
+		}
+
+		res
+	}
+
 	/// Gets the set of outbound HTLCs which are pending resolution in this channel or which were
 	/// resolved with a preimage from our counterparty.
 	///

lightning/src/ln/channelmanager.rs

@@ -15739,7 +15739,7 @@ where
 						log_error!(logger, " The ChannelMonitor for channel {} is at counterparty commitment transaction number {} but the ChannelManager is at counterparty commitment transaction number {}.",
 							&channel.context.channel_id(), monitor.get_cur_counterparty_commitment_number(), channel.get_cur_counterparty_commitment_transaction_number());
 					}
-					let mut shutdown_result =
+					let shutdown_result =
 						channel.force_shutdown(ClosureReason::OutdatedChannelManager);
 					if shutdown_result.unbroadcasted_batch_funding_txid.is_some() {
 						return Err(DecodeError::InvalidValue);
@@ -15771,7 +15771,10 @@ where
 							},
 						);
 					}
-					failed_htlcs.append(&mut shutdown_result.dropped_outbound_htlcs);
+					for (source, hash, cp_id, chan_id) in shutdown_result.dropped_outbound_htlcs {
+						let reason = LocalHTLCFailureReason::ChannelClosed;
+						failed_htlcs.push((source, hash, cp_id, chan_id, reason));
+					}
 					channel_closures.push_back((
 						events::Event::ChannelClosed {
 							channel_id: channel.context.channel_id(),
@@ -15813,6 +15816,7 @@ where
 								*payment_hash,
 								channel.context.get_counterparty_node_id(),
 								channel.context.channel_id(),
+								LocalHTLCFailureReason::ChannelClosed,
 							));
 						}
 					}
@@ -16537,6 +16541,20 @@ where
 							},
 						}
 					}
+					for (htlc_source, payment_hash) in monitor.get_onchain_failed_outbound_htlcs() {
+						log_info!(
+							args.logger,
+							"Failing HTLC with payment hash {} as it was resolved on-chain.",
+							payment_hash
+						);
+						failed_htlcs.push((
+							htlc_source,
+							payment_hash,
+							monitor.get_counterparty_node_id(),
+							monitor.channel_id(),
+							LocalHTLCFailureReason::OnChainTimeout,
+						));
+					}
 				}
 
 				// Whether the downstream channel was closed or not, try to re-apply any payment
@@ -17217,13 +17235,10 @@ where
 			}
 		}
 
-		for htlc_source in failed_htlcs.drain(..) {
-			let (source, payment_hash, counterparty_node_id, channel_id) = htlc_source;
-			let failure_reason = LocalHTLCFailureReason::ChannelClosed;
-			let receiver = HTLCHandlingFailureType::Forward {
-				node_id: Some(counterparty_node_id),
-				channel_id,
-			};
+		for htlc_source in failed_htlcs {
+			let (source, payment_hash, counterparty_id, channel_id, failure_reason) = htlc_source;
+			let receiver =
+				HTLCHandlingFailureType::Forward { node_id: Some(counterparty_id), channel_id };
 			let reason = HTLCFailReason::from_failure_code(failure_reason);
 			channel_manager.fail_htlc_backwards_internal(&source, &payment_hash, &reason, receiver);
 		}