Refresh async static invoices after channel changes

lightning/src/ln/async_payments_tests.rs

@@ -416,6 +416,34 @@ fn extract_static_invoice_om<'a>(
 	(peer_id, om, static_invoice.unwrap())
 }
 
+fn extract_serve_static_invoice_om<'a>(
+	recipient: &'a Node, next_hop_nodes: &[&'a Node],
+) -> (PublicKey, msgs::OnionMessage, StaticInvoice) {
+	let mut static_invoice = None;
+	let mut expected_msg_type = |peeled_onion: &_| match peeled_onion {
+		PeeledOnion::AsyncPayments(AsyncPaymentsMessage::ServeStaticInvoice(msg), _, _) => {
+			static_invoice = Some(msg.invoice.clone());
+			true
+		},
+		_ => false,
+	};
+	let expected_msg_type_to_ignore = |peeled_onion: &_| {
+		matches!(
+			peeled_onion,
+			&PeeledOnion::AsyncPayments(AsyncPaymentsMessage::OfferPathsRequest(_), _, _)
+		)
+	};
+	let (peer_id, om) = extract_expected_om(
+		recipient,
+		next_hop_nodes,
+		expected_msg_type,
+		expected_msg_type_to_ignore,
+	)
+	.pop()
+	.unwrap();
+	(peer_id, om, static_invoice.unwrap())
+}
+
 fn extract_held_htlc_available_oms<'a>(
 	payer: &'a Node, next_hop_nodes: &[&'a Node],
 ) -> Vec<(PublicKey, msgs::OnionMessage)> {
@@ -2507,6 +2535,62 @@ fn refresh_static_invoices_for_used_offers() {
 	assert_eq!(res.0, Some(PaidBolt12Invoice::StaticInvoice(updated_invoice)));
 }
 
+/// Checks that a used async receive offer gets a fresh server-side static invoice when a new
+/// channel becomes usable. Used offers may already be published, so they should not wait for the
+/// normal invoice refresh threshold after local payment paths change.
+#[test]
+fn refresh_static_invoices_for_used_offers_when_channel_opens() {
+	let chanmon_cfgs = create_chanmon_cfgs(3);
+	let node_cfgs = create_node_cfgs(3, &chanmon_cfgs);
+
+	let mut allow_priv_chan_fwds_cfg = test_default_channel_config();
+	allow_priv_chan_fwds_cfg.accept_forwards_to_priv_channels = true;
+	let node_chanmgrs =
+		create_node_chanmgrs(3, &node_cfgs, &[None, Some(allow_priv_chan_fwds_cfg), None]);
+
+	let nodes = create_network(3, &node_cfgs, &node_chanmgrs);
+	create_announced_chan_between_nodes_with_value(&nodes, 0, 1, 1_000_000, 0);
+	create_unannounced_chan_between_nodes_with_value(&nodes, 1, 2, 1_000_000, 0);
+	let server = &nodes[1];
+	let recipient = &nodes[2];
+
+	let recipient_id = vec![42; 32];
+	let inv_server_paths =
+		server.node.blinded_paths_for_async_recipient(recipient_id.clone(), None).unwrap();
+	recipient.node.set_paths_to_static_invoice_server(inv_server_paths).unwrap();
+	expect_offer_paths_requests(recipient, &[&nodes[0], server]);
+
+	let flow_res = pass_static_invoice_server_messages(server, recipient, recipient_id.clone());
+	let original_invoice = flow_res.invoice;
+
+	// Mark the offer as used so the cache treats it as potentially published by the application.
+	let _offer = recipient.node.get_async_receive_offer().unwrap();
+
+	// Keep the forced refresh direct so the test only checks that ChannelReady triggers the update.
+	server.message_router.peers_override.lock().unwrap().push(recipient.node.get_our_node_id());
+	recipient.message_router.peers_override.lock().unwrap().push(server.node.get_our_node_id());
+
+	create_unannounced_chan_between_nodes_with_value(&nodes, 1, 2, 1_000_000, 0);
+	let (peer_node_id, serve_static_invoice_om, updated_invoice) =
+		extract_serve_static_invoice_om(recipient, &[server]);
+	assert_eq!(peer_node_id, server.node.get_our_node_id());
+	assert_ne!(original_invoice, updated_invoice);
+
+	server
+		.onion_messenger
+		.handle_onion_message(recipient.node.get_our_node_id(), &serve_static_invoice_om);
+	let mut events = server.node.get_and_clear_pending_events();
+	assert_eq!(events.len(), 1);
+	match events.pop().unwrap() {
+		Event::PersistStaticInvoice { invoice, invoice_slot, recipient_id: ev_id, .. } => {
+			assert_eq!(invoice, updated_invoice);
+			assert_eq!(recipient_id, ev_id);
+			assert_eq!(invoice_slot, flow_res.invoice_slot);
+		},
+		_ => panic!(),
+	}
+}
+
 #[cfg_attr(feature = "std", ignore)]
 #[test]
 fn ignore_expired_static_invoice() {

lightning/src/ln/channelmanager.rs

@@ -3006,6 +3006,8 @@ pub struct ChannelManager<
 	funding_batch_states: Mutex<BTreeMap<Txid, Vec<(ChannelId, PublicKey, bool)>>>,
 
 	background_events_processed_since_startup: AtomicBool,
+	/// Set when a channel change may have made cached async receive static invoices stale.
+	async_receive_static_invoice_refresh_pending: AtomicBool,
 
 	event_persist_notifier: Notifier,
 	needs_persist_flag: AtomicBool,
@@ -3427,7 +3429,7 @@ macro_rules! emit_channel_pending_event {
 }
 
 macro_rules! emit_initial_channel_ready_event {
-	($locked_events: expr, $channel: expr) => {
+	($self: expr, $locked_events: expr, $channel: expr) => {
 		if $channel.context.should_emit_initial_channel_ready_event() {
 			debug_assert!($channel.context.channel_pending_event_emitted());
 			$locked_events.push_back((
@@ -3444,6 +3446,7 @@ macro_rules! emit_initial_channel_ready_event {
 				None,
 			));
 			$channel.context.set_initial_channel_ready_event_emitted();
+			$self.mark_async_receive_static_invoice_refresh_pending();
 		}
 	};
 }
@@ -3766,6 +3769,7 @@ impl<
 			pending_background_events: Mutex::new(Vec::new()),
 			total_consistency_lock: RwLock::new(()),
 			background_events_processed_since_startup: AtomicBool::new(false),
+			async_receive_static_invoice_refresh_pending: AtomicBool::new(false),
 			event_persist_notifier: Notifier::new(),
 			needs_persist_flag: AtomicBool::new(false),
 			funding_batch_states: Mutex::new(BTreeMap::new()),
@@ -4564,6 +4568,8 @@ impl<
 				));
 			}
 		}
+		self.mark_async_receive_static_invoice_refresh_pending();
+
 		for (err, counterparty_node_id) in shutdown_results.drain(..) {
 			let _ = self.handle_error(err, counterparty_node_id);
 		}
@@ -4693,6 +4699,7 @@ impl<
 				log_error!(logger, "Closing channel: {}", err_internal.err.err);
 
 				self.finish_close_channel(shutdown_res);
+				self.process_pending_async_receive_static_invoice_refresh();
 				if let Some((update, node_id_1, node_id_2)) = update_option {
 					let mut pending_broadcast_messages =
 						self.pending_broadcast_messages.lock().unwrap();
@@ -5905,6 +5912,30 @@ impl<
 		}
 	}
 
+	fn force_refresh_async_receive_static_invoices(&self) {
+		let peers = self.get_peers_for_blinded_path();
+		let channels = self.list_usable_channels();
+		let router = &self.router;
+
+		// Static invoices carry payment paths built from our current channels. Force-refreshing lets
+		// cached async receive offers keep their server-side invoices aligned after those channels
+		// change.
+		self.flow.force_refresh_async_receive_static_invoices(peers, channels, router);
+	}
+
+	fn mark_async_receive_static_invoice_refresh_pending(&self) {
+		self.async_receive_static_invoice_refresh_pending.store(true, Ordering::Release);
+	}
+
+	fn process_pending_async_receive_static_invoice_refresh(&self) {
+		// Channel state transitions often happen while a peer's channel lock is held. Defer the
+		// actual refresh until after those locks are released, because rebuilding static invoices
+		// needs a fresh snapshot of usable channels.
+		if self.async_receive_static_invoice_refresh_pending.swap(false, Ordering::AcqRel) {
+			self.force_refresh_async_receive_static_invoices();
+		}
+	}
+
 	#[cfg(test)]
 	pub(crate) fn test_check_refresh_async_receive_offers(&self) {
 		self.check_refresh_async_receive_offer_cache(false);
@@ -8903,6 +8934,7 @@ impl<
 			let mut timed_out_mpp_htlcs = Vec::new();
 			let mut pending_peers_awaiting_removal = Vec::new();
 			let mut feerate_cache = new_hash_map();
+			let mut should_refresh_static_invoices = false;
 
 			{
 				let per_peer_state = self.per_peer_state.read().unwrap();
@@ -8921,7 +8953,10 @@ impl<
 									Some(feerate)
 								}).unwrap();
 								let chan_needs_persist = self.update_channel_fee(chan_id, funded_chan, new_feerate);
-								if chan_needs_persist == NotifyOption::DoPersist { should_persist = NotifyOption::DoPersist; }
+								if chan_needs_persist == NotifyOption::DoPersist {
+									should_persist = NotifyOption::DoPersist;
+									should_refresh_static_invoices = true;
+								}
 
 								if let Err(e) = funded_chan.timer_check_closing_negotiation_progress() {
 									let (needs_close, err) = self.locked_handle_funded_force_close(&mut peer_state.closed_channel_monitor_update_ids, &mut peer_state.in_flight_monitor_updates, e, funded_chan);
@@ -9130,6 +9165,10 @@ impl<
 				.remove_stale_payments(duration_since_epoch, &self.pending_events);
 
 			self.check_refresh_async_receive_offer_cache(true);
+			if should_refresh_static_invoices {
+				self.mark_async_receive_static_invoice_refresh_pending();
+			}
+			self.process_pending_async_receive_static_invoice_refresh();
 
 			if self.check_free_holding_cells() {
 				// While we try to ensure we clear holding cells immediately, its possible we miss
@@ -11194,7 +11233,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 		{
 			let mut pending_events = self.pending_events.lock().unwrap();
 			emit_channel_pending_event!(pending_events, channel);
-			emit_initial_channel_ready_event!(pending_events, channel);
+			emit_initial_channel_ready_event!(self, pending_events, channel);
 			if let Some(splice_negotiated) = funding_tx_signed
 				.as_mut()
 				.and_then(|v| v.splice_negotiated.take())
@@ -12476,7 +12515,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 
 		{
 			let mut pending_events = self.pending_events.lock().unwrap();
-			emit_initial_channel_ready_event!(pending_events, chan);
+			emit_initial_channel_ready_event!(self, pending_events, chan);
 		}
 
 		Ok(())
@@ -13750,6 +13789,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 					},
 					None,
 				));
+				self.mark_async_receive_static_invoice_refresh_pending();
 				splice_promotion.discarded_funding.into_iter().for_each(|funding_info| {
 					let event = Event::DiscardFunding {
 						channel_id: chan.context.channel_id(),
@@ -16453,6 +16493,7 @@ impl<
 												funding_txo: Some(funding_txo.into_bitcoin_outpoint()),
 												channel_type: funded_channel.funding.get_channel_type().clone(),
 											}, None));
+											self.mark_async_receive_static_invoice_refresh_pending();
 											discarded_funding.into_iter().for_each(|funding_info| {
 												let event = Event::DiscardFunding {
 													channel_id: funded_channel.context.channel_id(),
@@ -16474,7 +16515,7 @@ impl<
 
 								{
 									let mut pending_events = self.pending_events.lock().unwrap();
-									emit_initial_channel_ready_event!(pending_events, funded_channel);
+									emit_initial_channel_ready_event!(self, pending_events, funded_channel);
 								}
 
 								if let Some(height) = height_opt {
@@ -16789,9 +16830,12 @@ impl<
 	}
 
 	fn handle_funding_signed(&self, counterparty_node_id: PublicKey, msg: &msgs::FundingSigned) {
-		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(self);
-		let res = self.internal_funding_signed(&counterparty_node_id, msg);
-		let _ = self.handle_error(res, counterparty_node_id);
+		{
+			let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(self);
+			let res = self.internal_funding_signed(&counterparty_node_id, msg);
+			let _ = self.handle_error(res, counterparty_node_id);
+		}
+		self.process_pending_async_receive_static_invoice_refresh();
 	}
 
 	fn handle_peer_storage(&self, counterparty_node_id: PublicKey, msg: msgs::PeerStorage) {
@@ -16815,15 +16859,18 @@ impl<
 		// channel_ready message - while the channel's state will change, any channel_ready message
 		// will ultimately be re-sent on startup and the `ChannelMonitor` won't be updated so we
 		// will not force-close the channel on startup.
-		let _persistence_guard = PersistenceNotifierGuard::optionally_notify(self, || {
-			let res = self.internal_channel_ready(&counterparty_node_id, msg);
-			let persist = match &res {
-				Err(e) if e.closes_channel() => NotifyOption::DoPersist,
-				_ => NotifyOption::SkipPersistHandleEvents,
-			};
-			let _ = self.handle_error(res, counterparty_node_id);
-			persist
-		});
+		{
+			let _persistence_guard = PersistenceNotifierGuard::optionally_notify(self, || {
+				let res = self.internal_channel_ready(&counterparty_node_id, msg);
+				let persist = match &res {
+					Err(e) if e.closes_channel() => NotifyOption::DoPersist,
+					_ => NotifyOption::SkipPersistHandleEvents,
+				};
+				let _ = self.handle_error(res, counterparty_node_id);
+				persist
+			});
+		}
+		self.process_pending_async_receive_static_invoice_refresh();
 	}
 
 	fn handle_stfu(&self, counterparty_node_id: PublicKey, msg: &msgs::Stfu) {
@@ -16873,16 +16920,19 @@ impl<
 
 	#[rustfmt::skip]
 	fn handle_splice_locked(&self, counterparty_node_id: PublicKey, msg: &msgs::SpliceLocked) {
-		let _persistence_guard = PersistenceNotifierGuard::optionally_notify(self, || {
-			let res = self.internal_splice_locked(&counterparty_node_id, msg);
-			let persist = match &res {
-				Err(e) if e.closes_channel() => NotifyOption::DoPersist,
-				Err(_) => NotifyOption::SkipPersistHandleEvents,
-				Ok(()) => NotifyOption::DoPersist,
-			};
-			let _ = self.handle_error(res, counterparty_node_id);
-			persist
-		});
+		{
+			let _persistence_guard = PersistenceNotifierGuard::optionally_notify(self, || {
+				let res = self.internal_splice_locked(&counterparty_node_id, msg);
+				let persist = match &res {
+					Err(e) if e.closes_channel() => NotifyOption::DoPersist,
+					Err(_) => NotifyOption::SkipPersistHandleEvents,
+					Ok(()) => NotifyOption::DoPersist,
+				};
+				let _ = self.handle_error(res, counterparty_node_id);
+				persist
+			});
+		}
+		self.process_pending_async_receive_static_invoice_refresh();
 	}
 
 	fn handle_shutdown(&self, counterparty_node_id: PublicKey, msg: &msgs::Shutdown) {
@@ -20513,6 +20563,7 @@ impl<
 			pending_background_events: Mutex::new(pending_background_events),
 			total_consistency_lock: RwLock::new(()),
 			background_events_processed_since_startup: AtomicBool::new(false),
+			async_receive_static_invoice_refresh_pending: AtomicBool::new(false),
 
 			event_persist_notifier: Notifier::new(),
 			needs_persist_flag: AtomicBool::new(false),

lightning/src/offers/async_receive_offer_cache.rs

@@ -469,6 +469,32 @@ impl AsyncReceiveOfferCache {
 		})
 	}
 
+	/// Returns cached offers whose static invoices should be refreshed after a local channel change.
+	pub(super) fn offers_needing_forced_invoice_refresh(
+		&self,
+	) -> impl Iterator<Item = (&Offer, Nonce, &Responder)> {
+		self.offers_with_idx().filter_map(move |(_, offer)| {
+			let needs_invoice_update = match offer.status {
+				// Used offers may already be published by the application. Keep their server-side
+				// invoices aligned with our current channels instead of waiting for the timer
+				// threshold.
+				OfferStatus::Used { .. } => true,
+				// Pending offers have already been sent to the server, but are not confirmed yet.
+				// Re-sending them is safe and matches the normal timer retry behavior.
+				OfferStatus::Pending => true,
+				// Ready offers have not been handed to the application yet. They are rotated by the
+				// offer-refresh path, so forcing invoice updates for them would mostly create extra
+				// server churn without helping published offers.
+				OfferStatus::Ready { .. } => false,
+			};
+			if needs_invoice_update {
+				Some((&offer.offer, offer.offer_nonce, &offer.update_static_invoice_path))
+			} else {
+				None
+			}
+		})
+	}
+
 	/// Should be called when we receive a [`StaticInvoicePersisted`] message from the static invoice
 	/// server, which indicates that a new offer was persisted by the server and they are ready to
 	/// serve the corresponding static invoice to payers on our behalf.