Move monitor<>outbound_payments startup htlc syncing code.

Move the code that ensures that HTLCs locked into ChannelMonitors are
synchronized with the ChannelManager's OutboundPayments store to the
outbound_payments module.

This is useful both because ChannelManager::read is very long/confusing method,
so it's nice to encapsulate some of its functionality, and because we need to
fix an existing bug in this logic where we may risk double-paying an offer due
to outbound_payments being stale on startup. See the next commit for this
bugfix.

lightning/src/ln/channelmanager.rs

@@ -61,7 +61,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::{OutboundPayments, PaymentAttempts, PendingOutboundPayment, RetryableInvoiceRequest, SendAlongPathArgs, StaleExpiration};
+use crate::ln::outbound_payment::{OutboundPayments, PendingOutboundPayment, RetryableInvoiceRequest, SendAlongPathArgs, StaleExpiration};
 use crate::ln::wire::Encode;
 use crate::offers::invoice::{Bolt12Invoice, DEFAULT_RELATIVE_EXPIRY, DerivedSigningPubkey, ExplicitSigningPubkey, InvoiceBuilder, UnsignedBolt12Invoice};
 use crate::offers::invoice_error::InvoiceError;
@@ -12569,37 +12569,11 @@ where
 								return Err(DecodeError::InvalidValue);
 							}
 
-							let path_amt = path.final_value_msat();
 							let mut session_priv_bytes = [0; 32];
 							session_priv_bytes[..].copy_from_slice(&session_priv[..]);
-							match pending_outbounds.pending_outbound_payments.lock().unwrap().entry(payment_id) {
+							pending_outbounds.insert_from_monitor_on_startup(
-								hash_map::Entry::Occupied(mut entry) => {
+								payment_id, htlc.payment_hash, session_priv_bytes, &path, best_block_height, logger
-									let newly_added = entry.get_mut().insert(session_priv_bytes, &path);
+							);
-									log_info!(logger, "{} a pending payment path for {} msat for session priv {} on an existing pending payment with payment hash {}",
-										if newly_added { "Added" } else { "Had" }, path_amt, log_bytes!(session_priv_bytes), htlc.payment_hash);
-								},
-								hash_map::Entry::Vacant(entry) => {
-									let path_fee = path.fee_msat();
-									entry.insert(PendingOutboundPayment::Retryable {
-										retry_strategy: None,
-										attempts: PaymentAttempts::new(),
-										payment_params: None,
-										session_privs: hash_set_from_iter([session_priv_bytes]),
-										payment_hash: htlc.payment_hash,
-										payment_secret: None, // only used for retries, and we'll never retry on startup
-										payment_metadata: None, // only used for retries, and we'll never retry on startup
-										keysend_preimage: None, // only used for retries, and we'll never retry on startup
-										custom_tlvs: Vec::new(), // only used for retries, and we'll never retry on startup
-										pending_amt_msat: path_amt,
-										pending_fee_msat: Some(path_fee),
-										total_msat: path_amt,
-										starting_block_height: best_block_height,
-										remaining_max_total_routing_fee_msat: None, // only used for retries, and we'll never retry on startup
-									});
-									log_info!(logger, "Added a pending payment for {} msat with payment hash {} for path with session priv {}",
-										path_amt, &htlc.payment_hash,  log_bytes!(session_priv_bytes));
-								}
-							}
 						}
 					}
 					for (htlc_source, (htlc, preimage_opt)) in monitor.get_all_current_outbound_htlcs() {

lightning/src/ln/outbound_payment.rs

@@ -2121,6 +2121,41 @@ impl OutboundPayments {
 		self.awaiting_invoice.store(false, Ordering::Release);
 		invoice_requests
 	}
+
+	pub(super) fn insert_from_monitor_on_startup<L: Logger>(
+		&self, payment_id: PaymentId, payment_hash: PaymentHash, session_priv_bytes: [u8; 32],
+		path: &Path, best_block_height: u32, logger: L
+	) {
+		let path_amt = path.final_value_msat();
+		match self.pending_outbound_payments.lock().unwrap().entry(payment_id) {
+			hash_map::Entry::Occupied(mut entry) => {
+				let newly_added = entry.get_mut().insert(session_priv_bytes, &path);
+				log_info!(logger, "{} a pending payment path for {} msat for session priv {} on an existing pending payment with payment hash {}",
+					if newly_added { "Added" } else { "Had" }, path_amt, log_bytes!(session_priv_bytes), payment_hash);
+			},
+			hash_map::Entry::Vacant(entry) => {
+				let path_fee = path.fee_msat();
+				entry.insert(PendingOutboundPayment::Retryable {
+					retry_strategy: None,
+					attempts: PaymentAttempts::new(),
+					payment_params: None,
+					session_privs: hash_set_from_iter([session_priv_bytes]),
+					payment_hash,
+					payment_secret: None, // only used for retries, and we'll never retry on startup
+					payment_metadata: None, // only used for retries, and we'll never retry on startup
+					keysend_preimage: None, // only used for retries, and we'll never retry on startup
+					custom_tlvs: Vec::new(), // only used for retries, and we'll never retry on startup
+					pending_amt_msat: path_amt,
+					pending_fee_msat: Some(path_fee),
+					total_msat: path_amt,
+					starting_block_height: best_block_height,
+					remaining_max_total_routing_fee_msat: None, // only used for retries, and we'll never retry on startup
+				});
+				log_info!(logger, "Added a pending payment for {} msat with payment hash {} for path with session priv {}",
+					path_amt, payment_hash,  log_bytes!(session_priv_bytes));
+			}
+		}
+	}
 }
 
 /// Returns whether a payment with the given [`PaymentHash`] and [`PaymentId`] is, in fact, a