Cache pending offer in specific invoice slot

When we as an async recipient receive offer paths from the static invoice server,
we create an offer and cache it, retrying persisting a corresponding invoice with
the server until it succeeds.

In the initially-merged version of this protocol, we would put this cached
offer in any slot in the cache that needed an offer at the time the offer paths
were received. However, in the last commit we started requesting offer paths
for a specific slot in the cache, as part of eliminating the use of the
invoice_id field in the overall protocol.

As a result, here we put the cached offer in the specific cache slot that the
original OfferPathsRequest indicated, rather than any slot that could use a new
offer.

Author: valentinewallace

lightning/src/blinded_path/message.rs

@@ -448,6 +448,14 @@ pub enum AsyncPaymentsContext {
 	/// [`OfferPathsRequest`]: crate::onion_message::async_payments::OfferPathsRequest
 	/// [`OfferPaths`]: crate::onion_message::async_payments::OfferPaths
 	OfferPaths {
+		/// The "slot" in the static invoice server's database that the invoice corresponding to these
+		/// offer paths should go into, originally set by us in [`OfferPathsRequest::invoice_slot`]. This
+		/// value allows us as the recipient to replace a specific invoice that is stored by the server,
+		/// which is useful for limiting the number of invoices stored by the server while also keeping
+		/// all the invoices persisted with the server fresh.
+		///
+		/// [`OfferPathsRequest::invoice_slot`]: crate::onion_message::async_payments::OfferPathsRequest::invoice_slot
+		invoice_slot: u16,
 		/// The time as duration since the Unix epoch at which this path expires and messages sent over
 		/// it should be ignored.
 		///
@@ -573,6 +581,7 @@ impl_writeable_tlv_based_enum!(AsyncPaymentsContext,
 	},
 	(2, OfferPaths) => {
 		(0, path_absolute_expiry, required),
+		(2, invoice_slot, required),
 	},
 	(3, StaticInvoicePersisted) => {
 		(0, offer_id, required),

lightning/src/offers/async_receive_offer_cache.rs

@@ -46,7 +46,7 @@ enum OfferStatus {
 	Pending,
 }
 
-#[derive(Clone)]
+#[derive(Clone, PartialEq)]
 struct AsyncReceiveOffer {
 	offer: Offer,
 	/// The time as duration since the Unix epoch at which this offer was created. Useful when
@@ -304,8 +304,8 @@ impl AsyncReceiveOfferCache {
 	/// until it succeeds, see [`AsyncReceiveOfferCache`] docs.
 	pub(super) fn cache_pending_offer(
 		&mut self, offer: Offer, offer_paths_absolute_expiry_secs: Option<u64>, offer_nonce: Nonce,
-		update_static_invoice_path: Responder, duration_since_epoch: Duration,
-	) -> Result<u16, ()> {
+		update_static_invoice_path: Responder, duration_since_epoch: Duration, cache_slot: u16,
+	) -> Result<(), ()> {
 		self.prune_expired_offers(duration_since_epoch, false);
 
 		if !self.should_build_offer_with_paths(
@@ -316,25 +316,23 @@ impl AsyncReceiveOfferCache {
 			return Err(());
 		}
 
-		let idx = match self.needs_new_offer_idx(duration_since_epoch) {
-			Some(idx) => idx,
-			None => return Err(()),
-		};
-
-		match self.offers.get_mut(idx) {
-			Some(offer_opt) => {
-				*offer_opt = Some(AsyncReceiveOffer {
-					offer,
-					created_at: duration_since_epoch,
-					offer_nonce,
-					status: OfferStatus::Pending,
-					update_static_invoice_path,
-				});
-			},
-			None => return Err(()),
+		let slot_needs_new_offer = self.offers.get(cache_slot as usize) == Some(&None)
+			|| self
+				.unused_offers_needing_refresh(duration_since_epoch)
+				.find(|(idx, _)| *idx == cache_slot as usize)
+				.is_some();
+		if !slot_needs_new_offer {
+			return Err(());
 		}
 
-		Ok(idx.try_into().map_err(|_| ())?)
+		self.offers[cache_slot as usize] = Some(AsyncReceiveOffer {
+			offer,
+			created_at: duration_since_epoch,
+			offer_nonce,
+			status: OfferStatus::Pending,
+			update_static_invoice_path,
+		});
+		Ok(())
 	}
 
 	/// If we have any empty slots in the cache or offers that can and should be replaced with a fresh
@@ -372,14 +370,9 @@ impl AsyncReceiveOfferCache {
 			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_ready_offers()
-			.filter(|(_, offer, _)| offer.created_at < awhile_ago)
-			// Get the stalest offer and return its index
-			.min_by(|(_, offer_a, _), (_, offer_b, _)| offer_a.created_at.cmp(&offer_b.created_at))
-			.map(|(idx, _, _)| idx)
+		self.unused_offers_needing_refresh(duration_since_epoch)
+			.min_by(|(_, offer_a), (_, offer_b)| offer_a.created_at.cmp(&offer_b.created_at))
+			.map(|(idx, _)| idx)
 	}
 
 	/// Returns an iterator over (offer_idx, offer)
@@ -446,6 +439,21 @@ impl AsyncReceiveOfferCache {
 		})
 	}
 
+	// 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.
+	fn unused_offers_needing_refresh(
+		&self, duration_since_epoch: Duration,
+	) -> impl Iterator<Item = (usize, &AsyncReceiveOffer)> {
+		let awhile_ago = duration_since_epoch.saturating_sub(OFFER_REFRESH_THRESHOLD);
+		self.unused_ready_offers().filter_map(move |(idx, offer, _)| {
+			if offer.created_at < awhile_ago {
+				Some((idx, offer))
+			} 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.

lightning/src/offers/flow.rs

@@ -1276,6 +1276,7 @@ where
 		let context = MessageContext::AsyncPayments(AsyncPaymentsContext::OfferPaths {
 			path_absolute_expiry: duration_since_epoch
 				.saturating_add(TEMP_REPLY_PATH_RELATIVE_EXPIRY),
+			invoice_slot: needs_new_offer_idx,
 		});
 		let reply_paths = match self.create_blinded_paths(peers, context) {
 			Ok(paths) => paths,
@@ -1444,14 +1445,15 @@ where
 		R::Target: Router,
 	{
 		let duration_since_epoch = self.duration_since_epoch();
-		match context {
-			AsyncPaymentsContext::OfferPaths { path_absolute_expiry } => {
+		let invoice_slot = match context {
+			AsyncPaymentsContext::OfferPaths { invoice_slot, path_absolute_expiry } => {
 				if duration_since_epoch > path_absolute_expiry {
 					return None;
 				}
+				invoice_slot
 			},
 			_ => return None,
-		}
+		};
 
 		{
 			// Only respond with `ServeStaticInvoice` if we actually need a new offer built.
@@ -1495,18 +1497,16 @@ where
 			Err(()) => return None,
 		};
 
-		let res = self.async_receive_offer_cache.lock().unwrap().cache_pending_offer(
+		if let Err(()) = self.async_receive_offer_cache.lock().unwrap().cache_pending_offer(
 			offer,
 			message.paths_absolute_expiry,
 			offer_nonce,
 			responder,
 			duration_since_epoch,
-		);
-
-		let invoice_slot = match res {
-			Ok(idx) => idx,
-			Err(()) => return None,
-		};
+			invoice_slot,
+		) {
+			return None;
+		}
 
 		let reply_path_context = {
 			MessageContext::AsyncPayments(AsyncPaymentsContext::StaticInvoicePersisted {