Handle claim result event generation in claim_funds_from_hop

Currently `claim_funds` and `claim_funds_internal` call
`claim_funds_from_hop` and then surface and `Event` to the user
informing them of the forwarded/claimed payment based on it's
result. In both places we assume that a claim "completed" even if
a monitor update is being done async.

Instead, here we push that event generation through a
`MonitorUpdateCompletionAction` and a call to
`handle_monitor_update_completion_action`. This will allow us to
hold the event(s) until async monitor updates complete in the
future.

Author: TheBlueMatt

lightning/src/ln/channelmanager.rs

@@ -4294,7 +4294,6 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 		let mut expected_amt_msat = None;
 		let mut valid_mpp = true;
 		let mut errs = Vec::new();
-		let mut claimed_any_htlcs = false;
 		let mut channel_state_lock = self.channel_state.lock().unwrap();
 		let channel_state = &mut *channel_state_lock;
 		for htlc in sources.iter() {
@@ -4344,13 +4343,14 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 		}
 		if valid_mpp {
 			for htlc in sources.drain(..) {
-				match self.claim_funds_from_hop(&mut channel_state_lock, htlc.prev_hop, payment_preimage) {
+				match self.claim_funds_from_hop(&mut channel_state_lock, htlc.prev_hop, payment_preimage,
+					|_| Some(MonitorUpdateCompletionAction::PaymentClaimed { payment_hash }))
+				{
 					ClaimFundsFromHop::MonitorUpdateFail(pk, err, _) => {
 						if let msgs::ErrorAction::IgnoreError = err.err.action {
 							// We got a temporary failure updating monitor, but will claim the
 							// HTLC when the monitor updating is restored (or on chain).
 							log_error!(self.logger, "Temporary failure claiming HTLC, treating as success: {}", err.err.err);
-							claimed_any_htlcs = true;
 						} else { errs.push((pk, err)); }
 					},
 					ClaimFundsFromHop::PrevHopForceClosed => unreachable!("We already checked for channel existence, we can't fail here!"),
@@ -4360,7 +4360,7 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 						// available to be claimed. Thus, it does not make sense to set
 						// `claimed_any_htlcs`.
 					},
-					ClaimFundsFromHop::Success(_) => claimed_any_htlcs = true,
+					ClaimFundsFromHop::Success(_) => {},
 				}
 			}
 		}
@@ -4377,22 +4377,17 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 			}
 		}
 
-		let PendingClaimingPayment { amount_msat, payment_purpose: purpose, receiver_node_id } =
-			self.pending_claimed_payments.lock().unwrap().remove(&payment_hash).unwrap();
-		if claimed_any_htlcs {
-			self.pending_events.lock().unwrap().push(events::Event::PaymentClaimed {
-				payment_hash, purpose, amount_msat, receiver_node_id: Some(receiver_node_id),
-			});
-		}
-
 		// Now we can handle any errors which were generated.
 		for (counterparty_node_id, err) in errs.drain(..) {
 			let res: Result<(), _> = Err(err);
 			let _ = handle_error!(self, res, counterparty_node_id);
 		}
 	}
 
-	fn claim_funds_from_hop(&self, channel_state_lock: &mut MutexGuard<ChannelHolder<<K::Target as KeysInterface>::Signer>>, prev_hop: HTLCPreviousHopData, payment_preimage: PaymentPreimage) -> ClaimFundsFromHop {
+	fn claim_funds_from_hop<ComplFunc: FnOnce(Option<u64>) -> Option<MonitorUpdateCompletionAction>>(&self,
+		channel_state_lock: &mut MutexGuard<ChannelHolder<<K::Target as KeysInterface>::Signer>>,
+		prev_hop: HTLCPreviousHopData, payment_preimage: PaymentPreimage, completion_action: ComplFunc)
+	-> ClaimFundsFromHop {
 		//TODO: Delay the claimed_funds relaying just like we do outbound relay!
 
 		let chan_id = prev_hop.outpoint.to_channel_id();
@@ -4407,6 +4402,7 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 								log_given_level!(self.logger, if e == ChannelMonitorUpdateStatus::PermanentFailure { Level::Error } else { Level::Debug },
 									"Failed to update channel monitor with preimage {:?}: {:?}",
 									payment_preimage, e);
+								self.handle_monitor_update_completion_actions(completion_action(Some(htlc_value_msat)));
 								return ClaimFundsFromHop::MonitorUpdateFail(
 									chan.get().get_counterparty_node_id(),
 									handle_monitor_update_res!(self, e, chan, RAACommitmentOrder::CommitmentFirst, false, msgs.is_some()).unwrap_err(),
@@ -4429,6 +4425,7 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 								}
 							});
 						}
+						self.handle_monitor_update_completion_actions(completion_action(Some(htlc_value_msat)));
 						return ClaimFundsFromHop::Success(htlc_value_msat);
 					} else {
 						return ClaimFundsFromHop::DuplicateClaim;
@@ -4448,10 +4445,14 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 					if drop {
 						chan.remove_entry();
 					}
+					self.handle_monitor_update_completion_actions(completion_action(None));
 					return ClaimFundsFromHop::MonitorUpdateFail(counterparty_node_id, res, None);
 				},
 			}
-		} else { return ClaimFundsFromHop::PrevHopForceClosed }
+		} else {
+			self.handle_monitor_update_completion_actions(completion_action(None));
+			return ClaimFundsFromHop::PrevHopForceClosed
+		}
 	}
 
 	fn finalize_claims(&self, mut sources: Vec<HTLCSource>) {
@@ -4524,13 +4525,24 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 			},
 			HTLCSource::PreviousHopData(hop_data) => {
 				let prev_outpoint = hop_data.outpoint;
-				let res = self.claim_funds_from_hop(&mut channel_state_lock, hop_data, payment_preimage);
-				let claimed_htlc = if let ClaimFundsFromHop::DuplicateClaim = res { false } else { true };
-				let htlc_claim_value_msat = match res {
-					ClaimFundsFromHop::MonitorUpdateFail(_, _, amt_opt) => amt_opt,
-					ClaimFundsFromHop::Success(amt) => Some(amt),
-					_ => None,
-				};
+				let res = self.claim_funds_from_hop(&mut channel_state_lock, hop_data, payment_preimage,
+					|htlc_claim_value_msat| {
+						if let Some(forwarded_htlc_value) = forwarded_htlc_value_msat {
+							let fee_earned_msat = if let Some(claimed_htlc_value) = htlc_claim_value_msat {
+								Some(claimed_htlc_value - forwarded_htlc_value)
+							} else { None };
+
+							let prev_channel_id = Some(prev_outpoint.to_channel_id());
+							let next_channel_id = Some(next_channel_id);
+
+							Some(MonitorUpdateCompletionAction::SurfaceEvent { event: events::Event::PaymentForwarded {
+								fee_earned_msat,
+								claim_from_onchain_tx: from_onchain,
+								prev_channel_id,
+								next_channel_id,
+							}})
+						} else { None }
+					});
 				if let ClaimFundsFromHop::PrevHopForceClosed = res {
 					let preimage_update = ChannelMonitorUpdate {
 						update_id: CLOSED_CHANNEL_UPDATE_ID,
@@ -4561,25 +4573,6 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 					let result: Result<(), _> = Err(err);
 					let _ = handle_error!(self, result, pk);
 				}
-
-				if claimed_htlc {
-					if let Some(forwarded_htlc_value) = forwarded_htlc_value_msat {
-						let fee_earned_msat = if let Some(claimed_htlc_value) = htlc_claim_value_msat {
-							Some(claimed_htlc_value - forwarded_htlc_value)
-						} else { None };
-
-						let mut pending_events = self.pending_events.lock().unwrap();
-						let prev_channel_id = Some(prev_outpoint.to_channel_id());
-						let next_channel_id = Some(next_channel_id);
-
-						pending_events.push(events::Event::PaymentForwarded {
-							fee_earned_msat,
-							claim_from_onchain_tx: from_onchain,
-							prev_channel_id,
-							next_channel_id,
-						});
-					}
-				}
 			},
 		}
 	}