f - Make async offer readiness race-free

Return a completed future when an async receive offer is already ready.

Drop redundant public wrappers now covered by lower-level APIs.

Co-Authored-By: HAL 9000

Author: tnull

lightning/src/ln/async_payments_tests.rs

@@ -1093,6 +1093,32 @@ fn ignore_duplicate_invoice() {
 	assert!(nodes[0].node.get_and_clear_pending_events().is_empty());
 }
 
+#[test]
+fn async_receive_offer_ready_future_completes_when_offer_already_ready() {
+	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 recipient_id = vec![42; 32];
+	let inv_server_paths =
+		nodes[1].node.blinded_paths_for_async_recipient(recipient_id.clone(), None).unwrap();
+	nodes[2].node.set_paths_to_static_invoice_server(inv_server_paths).unwrap();
+	expect_offer_paths_requests(&nodes[2], &[&nodes[0], &nodes[1]]);
+
+	pass_static_invoice_server_messages(&nodes[1], &nodes[2], recipient_id.clone());
+
+	assert!(nodes[2].node.get_async_receive_offer_ready_future().poll_is_complete());
+	assert!(nodes[2].node.get_async_receive_offer_ready_future().poll_is_complete());
+}
+
 #[test]
 fn async_receive_flow_success() {
 	// Test that an always-online sender can successfully pay an async receiver.

lightning/src/ln/channelmanager.rs

@@ -5911,16 +5911,6 @@ impl<
 		self.check_refresh_async_receive_offer_cache(false).unwrap();
 	}
 
-	/// Requests fresh async receive offer paths from the configured static invoice server, if any.
-	pub fn refresh_async_receive_offers(&self) -> Result<(), ()> {
-		self.check_refresh_async_receive_offer_cache(false).map_err(|()| {
-			log_error!(
-				self.logger,
-				"Failed to create blinded paths when requesting async receive offer paths"
-			);
-		})
-	}
-
 	/// Requests fresh async receive offer paths from the configured static invoice server, if any,
 	/// and attaches `payment_metadata` to the resulting BOLT 12 payment contexts.
 	///
@@ -5941,30 +5931,6 @@ impl<
 		})
 	}
 
-	/// Returns once an async receive offer is ready after the interactive static-invoice
-	/// protocol completes, or immediately if one is already available.
-	///
-	/// Callers that need a timeout can combine this future with their runtime's timeout
-	/// primitive.
-	#[cfg_attr(
-		feature = "std",
-		doc = "Synchronous callers should instead fetch the underlying [`Future`] via [`Self::get_async_receive_offer_ready_future`] and call [`Future::wait_timeout`] on it."
-	)]
-	///
-	/// [`Future`]: crate::util::wakers::Future
-	#[cfg_attr(
-		feature = "std",
-		doc = "[`Future::wait_timeout`]: crate::util::wakers::Future::wait_timeout"
-	)]
-	pub async fn await_async_receive_offer(&self) -> Result<Offer, ()> {
-		if let Ok(offer) = self.get_async_receive_offer() {
-			return Ok(offer);
-		}
-
-		self.flow.get_async_receive_offer_ready_future().await;
-		self.get_async_receive_offer()
-	}
-
 	/// Returns a [`Future`] that completes when an async receive offer is ready.
 	///
 	/// See [`OffersMessageFlow::get_async_receive_offer_ready_future`] for details.

lightning/src/offers/async_receive_offer_cache.rs

@@ -265,6 +265,15 @@ impl AsyncReceiveOfferCache {
 			.ok_or(())
 	}
 
+	/// Returns whether [`Self::get_async_receive_offer`] would return an offer without marking an
+	/// unused offer as used.
+	pub(super) fn has_async_receive_offer(&self, duration_since_epoch: Duration) -> bool {
+		self.offers_with_idx().any(|(_, offer)| {
+			!offer.offer.is_expired_no_std(duration_since_epoch)
+				&& matches!(offer.status, OfferStatus::Ready { .. } | OfferStatus::Used { .. })
+		})
+	}
+
 	/// Remove expired offers from the cache, returning the first slot number in the cache that needs
 	/// a new offer, if any exist.
 	pub(super) fn prune_expired_offers(

lightning/src/offers/flow.rs

@@ -1755,7 +1755,8 @@ impl<MR: MessageRouter, L: Logger> OffersMessageFlow<MR, L> {
 	}
 
 	/// Returns a [`Future`] that completes when an async receive offer is ready, i.e., after the
-	/// interactive static-invoice protocol completes.
+	/// interactive static-invoice protocol completes. If an offer is already ready, the returned
+	/// [`Future`] will already be complete.
 	///
 	/// Callers can `.await` the returned [`Future`] in an async context.
 	#[cfg_attr(
@@ -1765,7 +1766,12 @@ impl<MR: MessageRouter, L: Logger> OffersMessageFlow<MR, L> {
 	///
 	/// After it completes, use [`Self::get_async_receive_offer`] to retrieve the offer.
 	pub fn get_async_receive_offer_ready_future(&self) -> Future {
-		self.async_receive_offer_ready_notifier.get_future()
+		let cache = self.async_receive_offer_cache.lock().unwrap();
+		if cache.has_async_receive_offer(self.duration_since_epoch()) {
+			Future::completed()
+		} else {
+			self.async_receive_offer_ready_notifier.get_future()
+		}
 	}
 
 	/// Get the encoded [`AsyncReceiveOfferCache`] for persistence.

lightning/src/util/wakers.rs

@@ -162,6 +162,18 @@ pub struct Future {
 }
 
 impl Future {
+	pub(crate) fn completed() -> Self {
+		let state = Arc::new(Mutex::new(FutureState {
+			callbacks: Vec::new(),
+			std_future_callbacks: Vec::new(),
+			callbacks_with_state: Vec::new(),
+			complete: true,
+			callbacks_made: false,
+			next_idx: 1,
+		}));
+		Future { state, self_idx: 0 }
+	}
+
 	/// Registers a callback to be called upon completion of this future. If the future has already
 	/// completed, the callback will be called immediately.
 	///