Merge pull request #3125 from valentinewallace/2024-06-async-payments-prefactor
[rust-lightning] / lightning / src / ln / channelmanager.rs
index 80b98b2b6f5bacaafb31739744ea1357d5a0e1c1..6ba7396ebfe04262904640c6b041d6969e95195f 100644 (file)
@@ -58,7 +58,7 @@ use crate::ln::onion_utils::{HTLCFailReason, INVALID_ONION_BLINDING};
 use crate::ln::msgs::{ChannelMessageHandler, DecodeError, LightningError};
 #[cfg(test)]
 use crate::ln::outbound_payment;
-use crate::ln::outbound_payment::{Bolt12PaymentError, OutboundPayments, PaymentAttempts, PendingOutboundPayment, SendAlongPathArgs, StaleExpiration};
+use crate::ln::outbound_payment::{OutboundPayments, PaymentAttempts, PendingOutboundPayment, SendAlongPathArgs, StaleExpiration};
 use crate::ln::wire::Encode;
 use crate::offers::invoice::{BlindedPayInfo, Bolt12Invoice, DEFAULT_RELATIVE_EXPIRY, DerivedSigningPubkey, ExplicitSigningPubkey, InvoiceBuilder, UnsignedBolt12Invoice};
 use crate::offers::invoice_error::InvoiceError;
@@ -105,7 +105,7 @@ use core::time::Duration;
 use core::ops::Deref;
 
 // Re-export this for use in the public API.
-pub use crate::ln::outbound_payment::{PaymentSendFailure, ProbeSendFailure, Retry, RetryableSendFailure, RecipientOnionFields};
+pub use crate::ln::outbound_payment::{Bolt12PaymentError, PaymentSendFailure, ProbeSendFailure, Retry, RetryableSendFailure, RecipientOnionFields};
 use crate::ln::script::ShutdownScript;
 
 // We hold various information about HTLC relay in the HTLC objects in Channel itself:
@@ -3188,7 +3188,7 @@ where
                                                }
                                        } else {
                                                let mut chan_phase = remove_channel_phase!(self, chan_phase_entry);
-                                               shutdown_result = Some(chan_phase.context_mut().force_shutdown(false, ClosureReason::HolderForceClosed));
+                                               shutdown_result = Some(chan_phase.context_mut().force_shutdown(false, ClosureReason::HolderForceClosed { broadcasted_latest_txn: Some(false) }));
                                        }
                                },
                                hash_map::Entry::Vacant(_) => {
@@ -3357,7 +3357,7 @@ where
                        let closure_reason = if let Some(peer_msg) = peer_msg {
                                ClosureReason::CounterpartyForceClosed { peer_msg: UntrustedString(peer_msg.to_string()) }
                        } else {
-                               ClosureReason::HolderForceClosed
+                               ClosureReason::HolderForceClosed { broadcasted_latest_txn: Some(broadcast) }
                        };
                        let logger = WithContext::from(&self.logger, Some(*peer_node_id), Some(*channel_id), None);
                        if let hash_map::Entry::Occupied(chan_phase_entry) = peer_state.channel_by_id.entry(channel_id.clone()) {
@@ -3996,14 +3996,43 @@ where
                self.pending_outbound_payments.test_set_payment_metadata(payment_id, new_payment_metadata);
        }
 
-       pub(super) fn send_payment_for_bolt12_invoice(&self, invoice: &Bolt12Invoice, payment_id: PaymentId) -> Result<(), Bolt12PaymentError> {
+       /// Pays the [`Bolt12Invoice`] associated with the `payment_id` encoded in its `payer_metadata`.
+       ///
+       /// The invoice's `payer_metadata` is used to authenticate that the invoice was indeed requested
+       /// before attempting a payment. [`Bolt12PaymentError::UnexpectedInvoice`] is returned if this
+       /// fails or if the encoded `payment_id` is not recognized. The latter may happen once the
+       /// payment is no longer tracked because the payment was attempted after:
+       /// - an invoice for the `payment_id` was already paid,
+       /// - one full [timer tick] has elapsed since initially requesting the invoice when paying an
+       ///   offer, or
+       /// - the refund corresponding to the invoice has already expired.
+       ///
+       /// To retry the payment, request another invoice using a new `payment_id`.
+       ///
+       /// Attempting to pay the same invoice twice while the first payment is still pending will
+       /// result in a [`Bolt12PaymentError::DuplicateInvoice`].
+       ///
+       /// Otherwise, either [`Event::PaymentSent`] or [`Event::PaymentFailed`] are used to indicate
+       /// whether or not the payment was successful.
+       ///
+       /// [timer tick]: Self::timer_tick_occurred
+       pub fn send_payment_for_bolt12_invoice(&self, invoice: &Bolt12Invoice) -> Result<(), Bolt12PaymentError> {
+               let secp_ctx = &self.secp_ctx;
+               let expanded_key = &self.inbound_payment_key;
+               match invoice.verify(expanded_key, secp_ctx) {
+                       Ok(payment_id) => self.send_payment_for_verified_bolt12_invoice(invoice, payment_id),
+                       Err(()) => Err(Bolt12PaymentError::UnexpectedInvoice),
+               }
+       }
+
+       fn send_payment_for_verified_bolt12_invoice(&self, invoice: &Bolt12Invoice, payment_id: PaymentId) -> Result<(), Bolt12PaymentError> {
                let best_block_height = self.best_block.read().unwrap().height;
                let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(self);
                self.pending_outbound_payments
                        .send_payment_for_bolt12_invoice(
                                invoice, payment_id, &self.router, self.list_usable_channels(),
-                               || self.compute_inflight_htlcs(), &self.entropy_source, &self.node_signer,
-                               best_block_height, &self.logger, &self.pending_events,
+                               || self.compute_inflight_htlcs(), &self.entropy_source, &self.node_signer, &self,
+                               &self.secp_ctx, best_block_height, &self.logger, &self.pending_events,
                                |args| self.send_payment_along_path(args)
                        )
        }
@@ -4866,8 +4895,8 @@ where
                                if short_chan_id != 0 {
                                        let mut forwarding_counterparty = None;
                                        macro_rules! forwarding_channel_not_found {
-                                               () => {
-                                                       for forward_info in pending_forwards.drain(..) {
+                                               ($forward_infos: expr) => {
+                                                       for forward_info in $forward_infos {
                                                                match forward_info {
                                                                        HTLCForwardInfo::AddHTLC(PendingAddHTLCInfo {
                                                                                prev_short_channel_id, prev_htlc_id, prev_channel_id, prev_funding_outpoint,
@@ -4975,7 +5004,7 @@ where
                                        let (counterparty_node_id, forward_chan_id) = match chan_info_opt {
                                                Some((cp_id, chan_id)) => (cp_id, chan_id),
                                                None => {
-                                                       forwarding_channel_not_found!();
+                                                       forwarding_channel_not_found!(pending_forwards.drain(..));
                                                        continue;
                                                }
                                        };
@@ -4983,96 +5012,148 @@ where
                                        let per_peer_state = self.per_peer_state.read().unwrap();
                                        let peer_state_mutex_opt = per_peer_state.get(&counterparty_node_id);
                                        if peer_state_mutex_opt.is_none() {
-                                               forwarding_channel_not_found!();
+                                               forwarding_channel_not_found!(pending_forwards.drain(..));
                                                continue;
                                        }
                                        let mut peer_state_lock = peer_state_mutex_opt.unwrap().lock().unwrap();
                                        let peer_state = &mut *peer_state_lock;
-                                       if let Some(ChannelPhase::Funded(ref mut chan)) = peer_state.channel_by_id.get_mut(&forward_chan_id) {
-                                               let logger = WithChannelContext::from(&self.logger, &chan.context, None);
-                                               for forward_info in pending_forwards.drain(..) {
-                                                       let queue_fail_htlc_res = match forward_info {
-                                                               HTLCForwardInfo::AddHTLC(PendingAddHTLCInfo {
-                                                                       prev_short_channel_id, prev_htlc_id, prev_channel_id, prev_funding_outpoint,
-                                                                       prev_user_channel_id, forward_info: PendingHTLCInfo {
-                                                                               incoming_shared_secret, payment_hash, outgoing_amt_msat, outgoing_cltv_value,
-                                                                               routing: PendingHTLCRouting::Forward {
-                                                                                       onion_packet, blinded, ..
-                                                                               }, skimmed_fee_msat, ..
+                                       let mut draining_pending_forwards = pending_forwards.drain(..);
+                                       while let Some(forward_info) = draining_pending_forwards.next() {
+                                               let queue_fail_htlc_res = match forward_info {
+                                                       HTLCForwardInfo::AddHTLC(PendingAddHTLCInfo {
+                                                               prev_short_channel_id, prev_htlc_id, prev_channel_id, prev_funding_outpoint,
+                                                               prev_user_channel_id, forward_info: PendingHTLCInfo {
+                                                                       incoming_shared_secret, payment_hash, outgoing_amt_msat, outgoing_cltv_value,
+                                                                       routing: PendingHTLCRouting::Forward {
+                                                                               ref onion_packet, blinded, ..
+                                                                       }, skimmed_fee_msat, ..
+                                                               },
+                                                       }) => {
+                                                               let htlc_source = HTLCSource::PreviousHopData(HTLCPreviousHopData {
+                                                                       short_channel_id: prev_short_channel_id,
+                                                                       user_channel_id: Some(prev_user_channel_id),
+                                                                       channel_id: prev_channel_id,
+                                                                       outpoint: prev_funding_outpoint,
+                                                                       htlc_id: prev_htlc_id,
+                                                                       incoming_packet_shared_secret: incoming_shared_secret,
+                                                                       // Phantom payments are only PendingHTLCRouting::Receive.
+                                                                       phantom_shared_secret: None,
+                                                                       blinded_failure: blinded.map(|b| b.failure),
+                                                               });
+                                                               let next_blinding_point = blinded.and_then(|b| {
+                                                                       let encrypted_tlvs_ss = self.node_signer.ecdh(
+                                                                               Recipient::Node, &b.inbound_blinding_point, None
+                                                                       ).unwrap().secret_bytes();
+                                                                       onion_utils::next_hop_pubkey(
+                                                                               &self.secp_ctx, b.inbound_blinding_point, &encrypted_tlvs_ss
+                                                                       ).ok()
+                                                               });
+
+                                                               // Forward the HTLC over the most appropriate channel with the corresponding peer,
+                                                               // applying non-strict forwarding.
+                                                               // The channel with the least amount of outbound liquidity will be used to maximize the
+                                                               // probability of being able to successfully forward a subsequent HTLC.
+                                                               let maybe_optimal_channel = peer_state.channel_by_id.values_mut().filter_map(|phase| match phase {
+                                                                       ChannelPhase::Funded(chan) => {
+                                                                               let balances = chan.context.get_available_balances(&self.fee_estimator);
+                                                                               if outgoing_amt_msat <= balances.next_outbound_htlc_limit_msat &&
+                                                                                       outgoing_amt_msat >= balances.next_outbound_htlc_minimum_msat &&
+                                                                                       chan.context.is_usable() {
+                                                                                       Some((chan, balances))
+                                                                               } else {
+                                                                                       None
+                                                                               }
                                                                        },
-                                                               }) => {
-                                                                       let logger = WithChannelContext::from(&self.logger, &chan.context, Some(payment_hash));
-                                                                       log_trace!(logger, "Adding HTLC from short id {} with payment_hash {} to channel with short id {} after delay", prev_short_channel_id, &payment_hash, short_chan_id);
-                                                                       let htlc_source = HTLCSource::PreviousHopData(HTLCPreviousHopData {
-                                                                               short_channel_id: prev_short_channel_id,
-                                                                               user_channel_id: Some(prev_user_channel_id),
-                                                                               channel_id: prev_channel_id,
-                                                                               outpoint: prev_funding_outpoint,
-                                                                               htlc_id: prev_htlc_id,
-                                                                               incoming_packet_shared_secret: incoming_shared_secret,
-                                                                               // Phantom payments are only PendingHTLCRouting::Receive.
-                                                                               phantom_shared_secret: None,
-                                                                               blinded_failure: blinded.map(|b| b.failure),
-                                                                       });
-                                                                       let next_blinding_point = blinded.and_then(|b| {
-                                                                               let encrypted_tlvs_ss = self.node_signer.ecdh(
-                                                                                       Recipient::Node, &b.inbound_blinding_point, None
-                                                                               ).unwrap().secret_bytes();
-                                                                               onion_utils::next_hop_pubkey(
-                                                                                       &self.secp_ctx, b.inbound_blinding_point, &encrypted_tlvs_ss
-                                                                               ).ok()
-                                                                       });
-                                                                       if let Err(e) = chan.queue_add_htlc(outgoing_amt_msat,
-                                                                               payment_hash, outgoing_cltv_value, htlc_source.clone(),
-                                                                               onion_packet, skimmed_fee_msat, next_blinding_point, &self.fee_estimator,
-                                                                               &&logger)
-                                                                       {
-                                                                               if let ChannelError::Ignore(msg) = e {
-                                                                                       log_trace!(logger, "Failed to forward HTLC with payment_hash {}: {}", &payment_hash, msg);
+                                                                       _ => None,
+                                                               }).min_by_key(|(_, balances)| balances.next_outbound_htlc_limit_msat).map(|(c, _)| c);
+                                                               let optimal_channel = match maybe_optimal_channel {
+                                                                       Some(chan) => chan,
+                                                                       None => {
+                                                                               // Fall back to the specified channel to return an appropriate error.
+                                                                               if let Some(ChannelPhase::Funded(ref mut chan)) = peer_state.channel_by_id.get_mut(&forward_chan_id) {
+                                                                                       chan
                                                                                } else {
-                                                                                       panic!("Stated return value requirements in send_htlc() were not met");
+                                                                                       forwarding_channel_not_found!(core::iter::once(forward_info).chain(draining_pending_forwards));
+                                                                                       break;
                                                                                }
+                                                                       }
+                                                               };
+
+                                                               let logger = WithChannelContext::from(&self.logger, &optimal_channel.context, Some(payment_hash));
+                                                               let channel_description = if optimal_channel.context.get_short_channel_id() == Some(short_chan_id) {
+                                                                       "specified"
+                                                               } else {
+                                                                       "alternate"
+                                                               };
+                                                               log_trace!(logger, "Forwarding HTLC from SCID {} with payment_hash {} and next hop SCID {} over {} channel {} with corresponding peer {}",
+                                                                       prev_short_channel_id, &payment_hash, short_chan_id, channel_description, optimal_channel.context.channel_id(), &counterparty_node_id);
+                                                               if let Err(e) = optimal_channel.queue_add_htlc(outgoing_amt_msat,
+                                                                               payment_hash, outgoing_cltv_value, htlc_source.clone(),
+                                                                               onion_packet.clone(), skimmed_fee_msat, next_blinding_point, &self.fee_estimator,
+                                                                               &&logger)
+                                                               {
+                                                                       if let ChannelError::Ignore(msg) = e {
+                                                                               log_trace!(logger, "Failed to forward HTLC with payment_hash {} to peer {}: {}", &payment_hash, &counterparty_node_id, msg);
+                                                                       } else {
+                                                                               panic!("Stated return value requirements in send_htlc() were not met");
+                                                                       }
+
+                                                                       if let Some(ChannelPhase::Funded(ref mut chan)) = peer_state.channel_by_id.get_mut(&forward_chan_id) {
                                                                                let (failure_code, data) = self.get_htlc_temp_fail_err_and_data(0x1000|7, short_chan_id, chan);
                                                                                failed_forwards.push((htlc_source, payment_hash,
                                                                                        HTLCFailReason::reason(failure_code, data),
                                                                                        HTLCDestination::NextHopChannel { node_id: Some(chan.context.get_counterparty_node_id()), channel_id: forward_chan_id }
                                                                                ));
-                                                                               continue;
+                                                                       } else {
+                                                                               forwarding_channel_not_found!(core::iter::once(forward_info).chain(draining_pending_forwards));
+                                                                               break;
                                                                        }
-                                                                       None
-                                                               },
-                                                               HTLCForwardInfo::AddHTLC { .. } => {
-                                                                       panic!("short_channel_id != 0 should imply any pending_forward entries are of type Forward");
-                                                               },
-                                                               HTLCForwardInfo::FailHTLC { htlc_id, err_packet } => {
+                                                               }
+                                                               None
+                                                       },
+                                                       HTLCForwardInfo::AddHTLC { .. } => {
+                                                               panic!("short_channel_id != 0 should imply any pending_forward entries are of type Forward");
+                                                       },
+                                                       HTLCForwardInfo::FailHTLC { htlc_id, ref err_packet } => {
+                                                               if let Some(ChannelPhase::Funded(ref mut chan)) = peer_state.channel_by_id.get_mut(&forward_chan_id) {
+                                                                       let logger = WithChannelContext::from(&self.logger, &chan.context, None);
                                                                        log_trace!(logger, "Failing HTLC back to channel with short id {} (backward HTLC ID {}) after delay", short_chan_id, htlc_id);
-                                                                       Some((chan.queue_fail_htlc(htlc_id, err_packet, &&logger), htlc_id))
-                                                               },
-                                                               HTLCForwardInfo::FailMalformedHTLC { htlc_id, failure_code, sha256_of_onion } => {
+                                                                       Some((chan.queue_fail_htlc(htlc_id, err_packet.clone(), &&logger), htlc_id))
+                                                               } else {
+                                                                       forwarding_channel_not_found!(core::iter::once(forward_info).chain(draining_pending_forwards));
+                                                                       break;
+                                                               }
+                                                       },
+                                                       HTLCForwardInfo::FailMalformedHTLC { htlc_id, failure_code, sha256_of_onion } => {
+                                                               if let Some(ChannelPhase::Funded(ref mut chan)) = peer_state.channel_by_id.get_mut(&forward_chan_id) {
+                                                                       let logger = WithChannelContext::from(&self.logger, &chan.context, None);
                                                                        log_trace!(logger, "Failing malformed HTLC back to channel with short id {} (backward HTLC ID {}) after delay", short_chan_id, htlc_id);
                                                                        let res = chan.queue_fail_malformed_htlc(
                                                                                htlc_id, failure_code, sha256_of_onion, &&logger
                                                                        );
                                                                        Some((res, htlc_id))
-                                                               },
-                                                       };
-                                                       if let Some((queue_fail_htlc_res, htlc_id)) = queue_fail_htlc_res {
-                                                               if let Err(e) = queue_fail_htlc_res {
-                                                                       if let ChannelError::Ignore(msg) = e {
+                                                               } else {
+                                                                       forwarding_channel_not_found!(core::iter::once(forward_info).chain(draining_pending_forwards));
+                                                                       break;
+                                                               }
+                                                       },
+                                               };
+                                               if let Some((queue_fail_htlc_res, htlc_id)) = queue_fail_htlc_res {
+                                                       if let Err(e) = queue_fail_htlc_res {
+                                                               if let ChannelError::Ignore(msg) = e {
+                                                                       if let Some(ChannelPhase::Funded(ref mut chan)) = peer_state.channel_by_id.get_mut(&forward_chan_id) {
+                                                                               let logger = WithChannelContext::from(&self.logger, &chan.context, None);
                                                                                log_trace!(logger, "Failed to fail HTLC with ID {} backwards to short_id {}: {}", htlc_id, short_chan_id, msg);
-                                                                       } else {
-                                                                               panic!("Stated return value requirements in queue_fail_{{malformed_}}htlc() were not met");
                                                                        }
-                                                                       // fail-backs are best-effort, we probably already have one
-                                                                       // pending, and if not that's OK, if not, the channel is on
-                                                                       // the chain and sending the HTLC-Timeout is their problem.
-                                                                       continue;
+                                                               } else {
+                                                                       panic!("Stated return value requirements in queue_fail_{{malformed_}}htlc() were not met");
                                                                }
+                                                               // fail-backs are best-effort, we probably already have one
+                                                               // pending, and if not that's OK, if not, the channel is on
+                                                               // the chain and sending the HTLC-Timeout is their problem.
+                                                               continue;
                                                        }
                                                }
-                                       } else {
-                                               forwarding_channel_not_found!();
-                                               continue;
                                        }
                                } else {
                                        'next_forwardable_htlc: for forward_info in pending_forwards.drain(..) {
@@ -5531,7 +5612,7 @@ where
                                        log_error!(logger,
                                                "Force-closing pending channel with ID {} for not establishing in a timely manner", chan_id);
                                        update_maps_on_chan_removal!(self, &context);
-                                       shutdown_channels.push(context.force_shutdown(false, ClosureReason::HolderForceClosed));
+                                       shutdown_channels.push(context.force_shutdown(false, ClosureReason::HolderForceClosed { broadcasted_latest_txn: Some(false) }));
                                        pending_msg_events.push(MessageSendEvent::HandleError {
                                                node_id: counterparty_node_id,
                                                action: msgs::ErrorAction::SendErrorMessage {
@@ -6130,21 +6211,13 @@ where
                }
                if valid_mpp {
                        for htlc in sources.drain(..) {
-                               let prev_hop_chan_id = htlc.prev_hop.channel_id;
-                               if let Err((pk, err)) = self.claim_funds_from_hop(
+                               self.claim_funds_from_hop(
                                        htlc.prev_hop, payment_preimage,
                                        |_, definitely_duplicate| {
                                                debug_assert!(!definitely_duplicate, "We shouldn't claim duplicatively from a payment");
                                                Some(MonitorUpdateCompletionAction::PaymentClaimed { payment_hash })
                                        }
-                               ) {
-                                       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).
-                                               let logger = WithContext::from(&self.logger, None, Some(prev_hop_chan_id), Some(payment_hash));
-                                               log_error!(logger, "Temporary failure claiming HTLC, treating as success: {}", err.err.err);
-                                       } else { errs.push((pk, err)); }
-                               }
+                               );
                        }
                }
                if !valid_mpp {
@@ -6166,9 +6239,10 @@ where
                }
        }
 
-       fn claim_funds_from_hop<ComplFunc: FnOnce(Option<u64>, bool) -> Option<MonitorUpdateCompletionAction>>(&self,
-               prev_hop: HTLCPreviousHopData, payment_preimage: PaymentPreimage, completion_action: ComplFunc)
-       -> Result<(), (PublicKey, MsgHandleErrInternal)> {
+       fn claim_funds_from_hop<ComplFunc: FnOnce(Option<u64>, bool) -> Option<MonitorUpdateCompletionAction>>(
+               &self, prev_hop: HTLCPreviousHopData, payment_preimage: PaymentPreimage,
+               completion_action: ComplFunc,
+       ) {
                //TODO: Delay the claimed_funds relaying just like we do outbound relay!
 
                // If we haven't yet run background events assume we're still deserializing and shouldn't
@@ -6230,7 +6304,7 @@ where
                                                                let action = if let Some(action) = completion_action(None, true) {
                                                                        action
                                                                } else {
-                                                                       return Ok(());
+                                                                       return;
                                                                };
                                                                mem::drop(peer_state_lock);
 
@@ -6246,7 +6320,7 @@ where
                                                                } else {
                                                                        debug_assert!(false,
                                                                                "Duplicate claims should always free another channel immediately");
-                                                                       return Ok(());
+                                                                       return;
                                                                };
                                                                if let Some(peer_state_mtx) = per_peer_state.get(&node_id) {
                                                                        let mut peer_state = peer_state_mtx.lock().unwrap();
@@ -6271,7 +6345,7 @@ where
                                                        }
                                                }
                                        }
-                                       return Ok(());
+                                       return;
                                }
                        }
                }
@@ -6319,7 +6393,6 @@ where
                // generally always allowed to be duplicative (and it's specifically noted in
                // `PaymentForwarded`).
                self.handle_monitor_update_completion_actions(completion_action(None, false));
-               Ok(())
        }
 
        fn finalize_claims(&self, sources: Vec<HTLCSource>) {
@@ -6352,7 +6425,7 @@ where
                                let completed_blocker = RAAMonitorUpdateBlockingAction::from_prev_hop_data(&hop_data);
                                #[cfg(debug_assertions)]
                                let claiming_chan_funding_outpoint = hop_data.outpoint;
-                               let res = self.claim_funds_from_hop(hop_data, payment_preimage,
+                               self.claim_funds_from_hop(hop_data, payment_preimage,
                                        |htlc_claim_value_msat, definitely_duplicate| {
                                                let chan_to_release =
                                                        if let Some(node_id) = next_channel_counterparty_node_id {
@@ -6446,10 +6519,6 @@ where
                                                        })
                                                }
                                        });
-                               if let Err((pk, err)) = res {
-                                       let result: Result<(), _> = Err(err);
-                                       let _ = handle_error!(self, result, pk);
-                               }
                        },
                }
        }
@@ -6633,7 +6702,7 @@ where
                log_trace!(logger, "ChannelMonitor updated to {}. Current highest is {}. {} pending in-flight updates.",
                        highest_applied_update_id, channel.context.get_latest_monitor_update_id(),
                        remaining_in_flight);
-               if !channel.is_awaiting_monitor_update() || channel.context.get_latest_monitor_update_id() != highest_applied_update_id {
+               if !channel.is_awaiting_monitor_update() || remaining_in_flight != 0 {
                        return;
                }
                handle_monitor_update_completion!(self, peer_state_lock, peer_state, per_peer_state, channel);
@@ -8018,7 +8087,7 @@ where
                                                                                let reason = if let MonitorEvent::HolderForceClosedWithInfo { reason, .. } = monitor_event {
                                                                                        reason
                                                                                } else {
-                                                                                       ClosureReason::HolderForceClosed
+                                                                                       ClosureReason::HolderForceClosed { broadcasted_latest_txn: Some(true) }
                                                                                };
                                                                                failed_channels.push(chan.context.force_shutdown(false, reason.clone()));
                                                                                if let Ok(update) = self.get_channel_update_for_broadcast(&chan) {
@@ -10272,42 +10341,56 @@ where
                                };
 
                                match response {
-                                       Ok(invoice) => return responder.respond(OffersMessage::Invoice(invoice)),
-                                       Err(error) => return responder.respond(OffersMessage::InvoiceError(error.into())),
+                                       Ok(invoice) => responder.respond(OffersMessage::Invoice(invoice)),
+                                       Err(error) => responder.respond(OffersMessage::InvoiceError(error.into())),
                                }
                        },
                        OffersMessage::Invoice(invoice) => {
-                               let response = invoice
-                                       .verify(expanded_key, secp_ctx)
-                                       .map_err(|()| InvoiceError::from_string("Unrecognized invoice".to_owned()))
-                                       .and_then(|payment_id| {
+                               let result = match invoice.verify(expanded_key, secp_ctx) {
+                                       Ok(payment_id) => {
                                                let features = self.bolt12_invoice_features();
                                                if invoice.invoice_features().requires_unknown_bits_from(&features) {
                                                        Err(InvoiceError::from(Bolt12SemanticError::UnknownRequiredFeatures))
+                                               } else if self.default_configuration.manually_handle_bolt12_invoices {
+                                                       let event = Event::InvoiceReceived { payment_id, invoice, responder };
+                                                       self.pending_events.lock().unwrap().push_back((event, None));
+                                                       return ResponseInstruction::NoResponse;
                                                } else {
-                                                       self.send_payment_for_bolt12_invoice(&invoice, payment_id)
+                                                       self.send_payment_for_verified_bolt12_invoice(&invoice, payment_id)
                                                                .map_err(|e| {
                                                                        log_trace!(self.logger, "Failed paying invoice: {:?}", e);
                                                                        InvoiceError::from_string(format!("{:?}", e))
                                                                })
                                                }
-                                       });
+                                       },
+                                       Err(()) => Err(InvoiceError::from_string("Unrecognized invoice".to_owned())),
+                               };
 
-                               match (responder, response) {
-                                       (Some(responder), Err(e)) => responder.respond(OffersMessage::InvoiceError(e)),
-                                       (None, Err(_)) => {
-                                               log_trace!(
-                                                       self.logger,
-                                                       "A response was generated, but there is no reply_path specified for sending the response."
-                                               );
-                                               return ResponseInstruction::NoResponse;
-                                       }
-                                       _ => return ResponseInstruction::NoResponse,
+                               match result {
+                                       Ok(()) => ResponseInstruction::NoResponse,
+                                       Err(e) => match responder {
+                                               Some(responder) => responder.respond(OffersMessage::InvoiceError(e)),
+                                               None => {
+                                                       log_trace!(self.logger, "No reply path for sending invoice error: {:?}", e);
+                                                       ResponseInstruction::NoResponse
+                                               },
+                                       },
+                               }
+                       },
+                       #[cfg(async_payments)]
+                       OffersMessage::StaticInvoice(_invoice) => {
+                               match responder {
+                                       Some(responder) => {
+                                               responder.respond(OffersMessage::InvoiceError(
+                                                               InvoiceError::from_string("Static invoices not yet supported".to_string())
+                                               ))
+                                       },
+                                       None => return ResponseInstruction::NoResponse,
                                }
                        },
                        OffersMessage::InvoiceError(invoice_error) => {
                                log_trace!(self.logger, "Received invoice_error: {}", invoice_error);
-                               return ResponseInstruction::NoResponse;
+                               ResponseInstruction::NoResponse
                        },
                }
        }
@@ -12584,7 +12667,7 @@ mod tests {
 
                nodes[0].node.force_close_channel_with_peer(&chan.2, &nodes[1].node.get_our_node_id(), None, true).unwrap();
                check_added_monitors!(nodes[0], 1);
-               check_closed_event!(nodes[0], 1, ClosureReason::HolderForceClosed, [nodes[1].node.get_our_node_id()], 100000);
+               check_closed_event!(nodes[0], 1, ClosureReason::HolderForceClosed { broadcasted_latest_txn: Some(true) }, [nodes[1].node.get_our_node_id()], 100000);
 
                // Confirm that the channel_update was not sent immediately to node[1] but was cached.
                let node_1_events = nodes[1].node.get_and_clear_pending_msg_events();
@@ -12643,7 +12726,7 @@ mod tests {
                nodes[0].node.force_close_broadcasting_latest_txn(&chan.2, &nodes[1].node.get_our_node_id(), error_message.to_string()).unwrap();
                check_closed_broadcast!(nodes[0], true);
                check_added_monitors!(nodes[0], 1);
-               check_closed_event!(nodes[0], 1, ClosureReason::HolderForceClosed, [nodes[1].node.get_our_node_id()], 100000);
+               check_closed_event!(nodes[0], 1, ClosureReason::HolderForceClosed { broadcasted_latest_txn: Some(true) }, [nodes[1].node.get_our_node_id()], 100000);
 
                {
                        // Assert that nodes[1] is awaiting removal for nodes[0] once nodes[1] has been
@@ -13380,7 +13463,7 @@ mod tests {
                nodes[0].node.force_close_broadcasting_latest_txn(&chan_id, &nodes[1].node.get_our_node_id(), error_message.to_string()).unwrap();
                check_closed_broadcast(&nodes[0], 1, true);
                check_added_monitors(&nodes[0], 1);
-               check_closed_event!(nodes[0], 1, ClosureReason::HolderForceClosed, [nodes[1].node.get_our_node_id()], 100000);
+               check_closed_event!(nodes[0], 1, ClosureReason::HolderForceClosed { broadcasted_latest_txn: Some(true) }, [nodes[1].node.get_our_node_id()], 100000);
                {
                        let txn = nodes[0].tx_broadcaster.txn_broadcast();
                        assert_eq!(txn.len(), 1);