Initiate and refresh async receive offers

As an async recipient, we need to interactively build static invoices that an
always-online node will serve to payers on our behalf.

At the start of this process, we send a requests for paths to include in our
offers to the always-online node on startup and refresh the cached offers when
they start to get stale.

Author: valentinewallace

lightning/src/blinded_path/message.rs

@@ -404,6 +404,19 @@ pub enum OffersContext {
 /// [`AsyncPaymentsMessage`]: crate::onion_message::async_payments::AsyncPaymentsMessage
 #[derive(Clone, Debug)]
 pub enum AsyncPaymentsContext {
+	/// Context used by a reply path to an [`OfferPathsRequest`], provided back to us as an async
+	/// recipient in corresponding [`OfferPaths`] messages from the static invoice server.
+	///
+	/// [`OfferPathsRequest`]: crate::onion_message::async_payments::OfferPathsRequest
+	/// [`OfferPaths`]: crate::onion_message::async_payments::OfferPaths
+	OfferPaths {
+		/// The time as duration since the Unix epoch at which this path expires and messages sent over
+		/// it should be ignored.
+		///
+		/// As an async recipient we use this field to time out a static invoice server from sending us
+		/// offer paths if we are no longer configured to accept paths from them.
+		path_absolute_expiry: core::time::Duration,
+	},
 	/// Context contained within the reply [`BlindedMessagePath`] we put in outbound
 	/// [`HeldHtlcAvailable`] messages, provided back to us in corresponding [`ReleaseHeldHtlc`]
 	/// messages.
@@ -486,6 +499,9 @@ impl_writeable_tlv_based_enum!(AsyncPaymentsContext,
 		(2, hmac, required),
 		(4, path_absolute_expiry, required),
 	},
+	(2, OfferPaths) => {
+		(0, path_absolute_expiry, required),
+	},
 );
 
 /// Contains a simple nonce for use in a blinded path's context.

lightning/src/ln/channelmanager.rs

@@ -5139,6 +5139,20 @@ where
 			)
 	}
 
+	#[cfg(async_payments)]
+	fn check_refresh_async_receive_offers(&self, timer_tick_occurred: bool) {
+		let peers = self.get_peers_for_blinded_path();
+		match self.flow.check_refresh_async_receive_offers(peers, timer_tick_occurred) {
+			Err(()) => {
+				log_error!(
+					self.logger,
+					"Failed to create blinded paths when requesting async receive offer paths"
+				);
+			},
+			Ok(()) => {},
+		}
+	}
+
 	#[cfg(async_payments)]
 	fn initiate_async_payment(
 		&self, invoice: &StaticInvoice, payment_id: PaymentId,
@@ -7133,6 +7147,9 @@ where
 				duration_since_epoch, &self.pending_events
 			);
 
+			#[cfg(async_payments)]
+			self.check_refresh_async_receive_offers(true);
+
 			// Technically we don't need to do this here, but if we have holding cell entries in a
 			// channel that need freeing, it's better to do that here and block a background task
 			// than block the message queueing pipeline.
@@ -11664,6 +11681,13 @@ where
 			return NotifyOption::SkipPersistHandleEvents;
 			//TODO: Also re-broadcast announcement_signatures
 		});
+
+		// While we usually refresh the AsyncReceiveOfferCache on a timer, we also want to start
+		// interactively building offers as soon as we can after startup. We can't start building offers
+		// until we have some peer connection(s) to send onion messages over, so as a minor optimization
+		// refresh the cache when a peer connects.
+		#[cfg(async_payments)]
+		self.check_refresh_async_receive_offers(false);
 		res
 	}
 

lightning/src/offers/async_receive_offer_cache.rs

@@ -171,6 +171,131 @@ impl AsyncReceiveOfferCache {
 #[cfg(async_payments)]
 const MAX_CACHED_OFFERS_TARGET: usize = 10;
 
+// The max number of times we'll attempt to request offer paths per timer tick.
+#[cfg(async_payments)]
+const MAX_UPDATE_ATTEMPTS: u8 = 3;
+
+// If we have an offer that is replaceable and its invoice was confirmed as persisted more than 2
+// hours ago, we can go ahead and refresh it because we always want to have the freshest offer
+// possible when a user goes to retrieve a cached offer.
+//
+// We avoid replacing unused offers too quickly -- this prevents the case where we send multiple
+// invoices from different offers competing for the same slot to the server, messages are received
+// delayed or out-of-order, and we end up providing an offer to the user that the server just
+// deleted and replaced.
+#[cfg(async_payments)]
+const OFFER_REFRESH_THRESHOLD: Duration = Duration::from_secs(2 * 60 * 60);
+
+#[cfg(async_payments)]
+impl AsyncReceiveOfferCache {
+	/// Remove expired offers from the cache, returning whether new offers are needed.
+	pub(super) fn prune_expired_offers(
+		&mut self, duration_since_epoch: Duration, timer_tick_occurred: bool,
+	) -> bool {
+		// Remove expired offers from the cache.
+		let mut offer_was_removed = false;
+		for offer_opt in self.offers.iter_mut() {
+			let offer_is_expired = offer_opt
+				.as_ref()
+				.map_or(false, |offer| offer.offer.is_expired_no_std(duration_since_epoch));
+			if offer_is_expired {
+				offer_opt.take();
+				offer_was_removed = true;
+			}
+		}
+
+		// Allow more offer paths requests to be sent out in a burst roughly once per minute, or if an
+		// offer was removed.
+		if timer_tick_occurred || offer_was_removed {
+			self.reset_offer_paths_request_attempts()
+		}
+
+		self.needs_new_offer_idx(duration_since_epoch).is_some()
+			&& self.offer_paths_request_attempts < MAX_UPDATE_ATTEMPTS
+	}
+
+	/// If we have any empty slots in the cache or offers where the offer can and should be replaced
+	/// with a fresh offer, here we return the index of the slot that needs a new offer. The index is
+	/// used for setting [`ServeStaticInvoice::invoice_slot`] when sending the corresponding new
+	/// static invoice to the server, so the server knows which existing persisted invoice is being
+	/// replaced, if any.
+	///
+	/// Returns `None` if the cache is full and no offers can currently be replaced.
+	///
+	/// [`ServeStaticInvoice::invoice_slot`]: crate::onion_message::async_payments::ServeStaticInvoice::invoice_slot
+	fn needs_new_offer_idx(&self, duration_since_epoch: Duration) -> Option<usize> {
+		// If we have any empty offer slots, return the first one we find
+		let mut offers_opt_iter = self.offers.iter().enumerate();
+		let empty_slot_idx_opt =
+			offers_opt_iter.find_map(|(idx, offer_opt)| offer_opt.is_none().then(|| idx));
+		if empty_slot_idx_opt.is_some() {
+			return empty_slot_idx_opt;
+		}
+
+		// If all of our offers are already used or pending, then none are available to be replaced
+		let no_replaceable_offers = self.offers_with_idx().all(|(_, offer)| {
+			matches!(offer.status, OfferStatus::Used)
+				|| matches!(offer.status, OfferStatus::Pending)
+		});
+		if no_replaceable_offers {
+			return None;
+		}
+
+		// If we only have 1 offer that is available for payments, then none are available to be
+		// replaced
+		let num_payable_offers = self
+			.offers_with_idx()
+			.filter(|(_, offer)| {
+				matches!(offer.status, OfferStatus::Used)
+					|| matches!(offer.status, OfferStatus::Ready { .. })
+			})
+			.count();
+		if num_payable_offers <= 1 {
+			return None;
+		}
+
+		// Filter for unused offers that were last updated more than two hours ago, so they are stale
+		// enough to warrant replacement.
+		let two_hours_ago = duration_since_epoch.saturating_sub(OFFER_REFRESH_THRESHOLD);
+		self.offers_with_idx()
+			.filter_map(|(idx, offer)| match offer.status {
+				OfferStatus::Ready { invoice_confirmed_persisted_at } => {
+					Some((idx, offer, invoice_confirmed_persisted_at))
+				},
+				_ => None,
+			})
+			.filter(|(_, _, invoice_confirmed_persisted_at)| {
+				*invoice_confirmed_persisted_at < two_hours_ago
+			})
+			// Get the stalest offer and return its index
+			.min_by(|a, b| a.2.cmp(&b.2))
+			.map(|(idx, _, _)| idx)
+	}
+
+	/// Returns an iterator over (offer_idx, offer)
+	fn offers_with_idx(&self) -> impl Iterator<Item = (usize, &AsyncReceiveOffer)> {
+		self.offers.iter().enumerate().filter_map(|(idx, offer_opt)| {
+			if let Some(offer) = offer_opt {
+				Some((idx, offer))
+			} else {
+				None
+			}
+		})
+	}
+
+	// Indicates that onion messages requesting new offer paths have been sent to the static invoice
+	// server. Calling this method allows the cache to self-limit how many requests are sent.
+	pub(super) fn new_offers_requested(&mut self) {
+		self.offer_paths_request_attempts += 1;
+	}
+
+	/// Called on timer tick (roughly once per minute) to allow another MAX_UPDATE_ATTEMPTS offer
+	/// paths requests to go out.
+	fn reset_offer_paths_request_attempts(&mut self) {
+		self.offer_paths_request_attempts = 0;
+	}
+}
+
 impl Writeable for AsyncReceiveOfferCache {
 	fn write<W: Writer>(&self, w: &mut W) -> Result<(), io::Error> {
 		write_tlv_fields!(w, {

lightning/src/offers/flow.rs

@@ -66,7 +66,7 @@ use {
 	crate::offers::offer::Amount,
 	crate::offers::signer,
 	crate::offers::static_invoice::{StaticInvoice, StaticInvoiceBuilder},
-	crate::onion_message::async_payments::HeldHtlcAvailable,
+	crate::onion_message::async_payments::{HeldHtlcAvailable, OfferPathsRequest},
 };
 
 #[cfg(feature = "dnssec")]
@@ -235,6 +235,11 @@ where
 /// even if multiple invoices are received.
 const OFFERS_MESSAGE_REQUEST_LIMIT: usize = 10;
 
+/// The default relative expiry for reply paths where a quick response is expected and the reply
+/// path is single-use.
+#[cfg(async_payments)]
+const TEMP_REPLY_PATH_RELATIVE_EXPIRY: Duration = Duration::from_secs(7200);
+
 impl<MR: Deref> OffersMessageFlow<MR>
 where
 	MR::Target: MessageRouter,
@@ -1123,6 +1128,67 @@ where
 		core::mem::take(&mut self.pending_dns_onion_messages.lock().unwrap())
 	}
 
+	/// Sends out [`OfferPathsRequest`] onion messages if we are an often-offline recipient and are
+	/// configured to interactively build offers and static invoices with a static invoice server.
+	///
+	/// # Usage
+	///
+	/// This method should be called on peer connection and once per minute or so, to keep the offers
+	/// cache updated.
+	///
+	/// SHOULD NOT be called much more than once per minute or too many messages may be sent to the
+	/// server node.
+	///
+	/// Errors if we failed to create blinded reply paths when sending an [`OfferPathsRequest`] message.
+	#[cfg(async_payments)]
+	pub(crate) fn check_refresh_async_receive_offers(
+		&self, peers: Vec<MessageForwardNode>, timer_tick_occurred: bool,
+	) -> Result<(), ()> {
+		// Terminate early if this node does not intend to receive async payments.
+		if self.paths_to_static_invoice_server.lock().unwrap().is_empty() {
+			return Ok(());
+		}
+
+		let duration_since_epoch = self.duration_since_epoch();
+
+		// Update the cache to remove expired offers, and check to see whether we need new offers to be
+		// interactively built with the static invoice server.
+		let needs_new_offers = self
+			.async_receive_offer_cache
+			.lock()
+			.unwrap()
+			.prune_expired_offers(duration_since_epoch, timer_tick_occurred);
+
+		// If we need new offers, send out offer paths request messages to the static invoice server.
+		if needs_new_offers {
+			let context = MessageContext::AsyncPayments(AsyncPaymentsContext::OfferPaths {
+				path_absolute_expiry: duration_since_epoch
+					.saturating_add(TEMP_REPLY_PATH_RELATIVE_EXPIRY),
+			});
+			let reply_paths = match self.create_blinded_paths(peers, context) {
+				Ok(paths) => paths,
+				Err(()) => {
+					return Err(());
+				},
+			};
+
+			// We can't fail past this point, so indicate to the cache that we've requested new offers.
+			self.async_receive_offer_cache.lock().unwrap().new_offers_requested();
+
+			let mut pending_async_payments_messages =
+				self.pending_async_payments_messages.lock().unwrap();
+			let message = AsyncPaymentsMessage::OfferPathsRequest(OfferPathsRequest {});
+			enqueue_onion_message_with_reply_paths(
+				message,
+				&self.paths_to_static_invoice_server.lock().unwrap()[..],
+				reply_paths,
+				&mut pending_async_payments_messages,
+			);
+		}
+
+		Ok(())
+	}
+
 	/// Get the `AsyncReceiveOfferCache` for persistence.
 	pub(crate) fn writeable_async_receive_offer_cache(&self) -> impl Writeable + '_ {
 		&self.async_receive_offer_cache