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,21 @@ 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.
+		///
+		/// This avoids the situation where the [`OfferPaths`] message is very delayed and thus
+		/// outdated.
+		///
+		/// [`OfferPaths`]: crate::onion_message::async_payments::OfferPaths
+		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 +501,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

@@ -5248,6 +5248,20 @@ where
 		)
 	}
 
+	#[cfg(async_payments)]
+	fn check_refresh_async_receive_offer_cache(&self, timer_tick_occurred: bool) {
+		let peers = self.get_peers_for_blinded_path();
+		match self.flow.check_refresh_async_receive_offer_cache(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,
@@ -7242,6 +7256,9 @@ where
 				duration_since_epoch, &self.pending_events
 			);
 
+			#[cfg(async_payments)]
+			self.check_refresh_async_receive_offer_cache(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.
@@ -11978,6 +11995,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_offer_cache(false);
 		res
 	}
 

lightning/src/offers/async_receive_offer_cache.rs

@@ -172,6 +172,133 @@ 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, force_reset_request_attempts: 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 up to `MAX_UPDATE_ATTEMPTS` offer paths requests to be sent out roughly once per
+		// minute, or if an offer was removed.
+		if force_reset_request_attempts || 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 that 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 empty_slot_idx_opt = self.offers.iter().position(|offer_opt| offer_opt.is_none());
+		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 | OfferStatus::Pending));
+		if no_replaceable_offers {
+			return None;
+		}
+
+		// All offers are pending except for one, so we shouldn't request an update of the only usable
+		// offer
+		let num_payable_offers = self
+			.offers_with_idx()
+			.filter(|(_, offer)| {
+				matches!(offer.status, OfferStatus::Used | OfferStatus::Ready { .. })
+			})
+			.count();
+		if num_payable_offers <= 1 {
+			return None;
+		}
+
+		// Filter for unused offers where longer than OFFER_REFRESH_THRESHOLD time has passed since they
+		// were last updated, so they are stale enough to warrant replacement.
+		let awhile_ago = duration_since_epoch.saturating_sub(OFFER_REFRESH_THRESHOLD);
+		self.unused_offers()
+			.filter(|(_, _, invoice_confirmed_persisted_at)| {
+				*invoice_confirmed_persisted_at < awhile_ago
+			})
+			// Get the stalest offer and return its index
+			.min_by(|(_, _, persisted_at_a), (_, _, persisted_at_b)| {
+				persisted_at_a.cmp(&persisted_at_b)
+			})
+			.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
+			}
+		})
+	}
+
+	/// Returns an iterator over (offer_idx, offer, invoice_confirmed_persisted_at)
+	/// where all returned offers are [`OfferStatus::Ready`]
+	fn unused_offers(&self) -> impl Iterator<Item = (usize, &AsyncReceiveOffer, Duration)> {
+		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,
+		})
+	}
+
+	// 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")]
@@ -238,6 +238,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,
@@ -1126,6 +1131,65 @@ 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. When calling this method once per minute, SHOULD set `timer_tick_occurred` so
+	/// the cache can self-regulate the number of messages sent out.
+	///
+	/// Errors if we failed to create blinded reply paths when sending an [`OfferPathsRequest`] message.
+	#[cfg(async_payments)]
+	pub(crate) fn check_refresh_async_receive_offer_cache(
+		&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