Cache pending offer in specific invoice slot

diff --git a/lightning/src/blinded_path/message.rs b/lightning/src/blinded_path/message.rs
index 142fe99..37499f2 100644
--- a/lightning/src/blinded_path/message.rs
+++ b/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),
diff --git a/lightning/src/offers/async_receive_offer_cache.rs b/lightning/src/offers/async_receive_offer_cache.rs
index b3e6b83..5b9ea26 100644
--- a/lightning/src/offers/async_receive_offer_cache.rs
+++ b/lightning/src/offers/async_receive_offer_cache.rs
@@ -64,6 +64,18 @@ struct AsyncReceiveOffer {
 	update_static_invoice_path: Responder,
 }
 
+impl AsyncReceiveOffer {
+	/// An offer needs to be refreshed if it is unused and has been cached longer than
+	/// `OFFER_REFRESH_THRESHOLD`.
+	fn needs_refresh(&self, duration_since_epoch: Duration) -> bool {
+		let awhile_ago = duration_since_epoch.saturating_sub(OFFER_REFRESH_THRESHOLD);
+		match self.status {
+			OfferStatus::Ready { .. } => self.created_at < awhile_ago,
+			_ => false,
+		}
+	}
+}
+
 impl_writeable_tlv_based_enum!(OfferStatus,
 	(0, Used) => {
 		(0, invoice_created_at, required),
@@ -283,9 +295,9 @@ impl AsyncReceiveOfferCache {
 	/// to build a new offer.
 	pub(super) fn should_build_offer_with_paths(
 		&self, offer_paths: &[BlindedMessagePath], offer_paths_absolute_expiry_secs: Option<u64>,
-		duration_since_epoch: Duration,
+		slot: u16, duration_since_epoch: Duration,
 	) -> bool {
-		if self.needs_new_offer_idx(duration_since_epoch).is_none() {
+		if !self.slot_needs_offer(slot, duration_since_epoch) {
 			return false;
 		}
 
@@ -307,37 +319,51 @@ 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, slot: u16,
+	) -> Result<(), ()> {
 		self.prune_expired_offers(duration_since_epoch, false);
 
 		if !self.should_build_offer_with_paths(
 			offer.paths(),
 			offer_paths_absolute_expiry_secs,
+			slot,
 			duration_since_epoch,
 		) {
 			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 {
+		match self.offers.get_mut(slot as usize) {
+			Some(slot) => {
+				*slot = Some(AsyncReceiveOffer {
 					offer,
 					created_at: duration_since_epoch,
 					offer_nonce,
 					status: OfferStatus::Pending,
 					update_static_invoice_path,
-				});
+				})
+			},
+			None => {
+				debug_assert!(false, "Slot in cache was invalid but should'be been checked above");
+				return Err(());
 			},
-			None => return Err(()),
 		}
 
-		Ok(idx.try_into().map_err(|_| ())?)
+		Ok(())
+	}
+
+	fn slot_needs_offer(&self, slot: u16, duration_since_epoch: Duration) -> bool {
+		match self.offers.get(slot as usize) {
+			Some(Some(offer)) => offer.needs_refresh(duration_since_epoch),
+			// This slot in the cache was pre-allocated as needing an offer in
+			// `set_paths_to_static_invoice_server` and is currently vacant
+			Some(None) => true,
+			// `slot` is out-of-range. Note that the cache only has `MAX_CACHED_OFFERS_TARGET` slots
+			// total, so any slots outside of that range are invalid.
+			None => {
+				debug_assert!(false, "Got offer paths for a non-existent slot in the cache");
+				false
+			},
+		}
 	}
 
 	/// If we have any empty slots in the cache or offers that can and should be replaced with a fresh
@@ -377,12 +403,11 @@ 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.
-		let awhile_ago = duration_since_epoch.saturating_sub(OFFER_REFRESH_THRESHOLD);
-		self.unused_ready_offers()
-			.filter(|(_, offer, _)| offer.created_at < awhile_ago)
+		self.offers_with_idx()
+			.filter(|(_, offer)| offer.needs_refresh(duration_since_epoch))
 			// 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)
+			.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)
diff --git a/lightning/src/offers/flow.rs b/lightning/src/offers/flow.rs
index 7c809ad..39fdd4f 100644
--- a/lightning/src/offers/flow.rs
+++ b/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_slot,
 		});
 		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.
@@ -1460,6 +1462,7 @@ where
 			if !cache.should_build_offer_with_paths(
 				&message.paths[..],
 				message.paths_absolute_expiry,
+				invoice_slot,
 				duration_since_epoch,
 			) {
 				return None;
@@ -1495,18 +1498,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 {