Rename MultiFuturePoller and let it return concrete results

Author: tnull

lightning/src/events/mod.rs

@@ -2334,6 +2334,7 @@ pub trait EventsProvider {
 /// currently succeed (e.g., due to a persistence failure).
 ///
 /// LDK will ensure the event is persisted and will eventually be replayed.
+#[derive(Clone, Copy, Debug)]
 pub struct ReplayEvent();
 
 /// A trait implemented for objects handling events from [`EventsProvider`].

lightning/src/onion_message/messenger.rs

@@ -18,7 +18,7 @@ use bitcoin::secp256k1::{self, PublicKey, Scalar, Secp256k1, SecretKey};
 use crate::blinded_path::{BlindedPath, IntroductionNode, NextMessageHop, NodeIdLookUp};
 use crate::blinded_path::message::{advance_path_by_one, ForwardNode, ForwardTlvs, MessageContext, OffersContext, ReceiveTlvs};
 use crate::blinded_path::utils;
-use crate::events::{Event, EventHandler, EventsProvider};
+use crate::events::{Event, EventHandler, EventsProvider, ReplayEvent};
 use crate::sign::{EntropySource, NodeSigner, Recipient};
 use crate::ln::features::{InitFeatures, NodeFeatures};
 use crate::ln::msgs::{self, OnionMessage, OnionMessageHandler, SocketAddress};
@@ -1328,7 +1328,7 @@ where
 	/// have an ordering requirement.
 	///
 	/// See the trait-level documentation of [`EventsProvider`] for requirements.
-	pub async fn process_pending_events_async<Future: core::future::Future<Output = Result<(), ()>> + core::marker::Unpin, H: Fn(Event) -> Future>(
+	pub async fn process_pending_events_async<Future: core::future::Future<Output = Result<(), ReplayEvent>> + core::marker::Unpin, H: Fn(Event) -> Future>(
 		&self, handler: H
 	) {
 		let mut intercepted_msgs = Vec::new();
@@ -1346,26 +1346,26 @@ where
 		for (node_id, recipient) in self.message_recipients.lock().unwrap().iter_mut() {
 			if let OnionMessageRecipient::PendingConnection(_, addresses, _) = recipient {
 				if let Some(addresses) = addresses.take() {
-					futures.push(Some(handler(Event::ConnectionNeeded { node_id: *node_id, addresses })));
+					futures.push(handler(Event::ConnectionNeeded { node_id: *node_id, addresses }));
 				}
 			}
 		}
 
 		for ev in intercepted_msgs {
 			if let Event::OnionMessageIntercepted { .. } = ev {} else { debug_assert!(false); }
-			futures.push(Some(handler(ev)));
+			futures.push(handler(ev));
 		}
 		// Let the `OnionMessageIntercepted` events finish before moving on to peer_connecteds
-		crate::util::async_poll::MultiFuturePoller(futures).await;
+		crate::util::async_poll::MultiResultFuturePoller::new(futures).await;
 
 		if peer_connecteds.len() <= 1 {
 			for event in peer_connecteds { handler(event).await; }
 		} else {
 			let mut futures = Vec::new();
 			for event in peer_connecteds {
-				futures.push(Some(handler(event)));
+				futures.push(handler(event));
 			}
-			crate::util::async_poll::MultiFuturePoller(futures).await;
+			crate::util::async_poll::MultiResultFuturePoller::new(futures).await;
 		}
 	}
 }

lightning/src/util/async_poll.rs

@@ -15,29 +15,64 @@ use core::marker::Unpin;
 use core::pin::Pin;
 use core::task::{Context, Poll};
 
-pub(crate) struct MultiFuturePoller<F: Future<Output = ()> + Unpin>(pub Vec<Option<F>>);
+enum ResultFuture<F: Future<Output = Result<(), E>>, E: Copy + Unpin> {
+	Pending(F),
+	Ready(Result<(), E>),
+}
+
+pub(crate) struct MultiResultFuturePoller<
+	F: Future<Output = Result<(), E>> + Unpin,
+	E: Copy + Unpin,
+> {
+	futures_state: Vec<ResultFuture<F, E>>,
+}
 
-impl<F: Future<Output = ()> + Unpin> Future for MultiFuturePoller<F> {
-	type Output = ();
-	fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
+impl<F: Future<Output = Result<(), E>> + Unpin, E: Copy + Unpin> MultiResultFuturePoller<F, E> {
+	pub fn new(futures: Vec<F>) -> Self {
+		let futures_state = futures.into_iter().map(|f| ResultFuture::Pending(f)).collect();
+		Self { futures_state }
+	}
+}
+
+impl<F: Future<Output = Result<(), E>> + Unpin, E: Copy + Unpin> Future
+	for MultiResultFuturePoller<F, E>
+{
+	type Output = Vec<Result<(), E>>;
+	fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Vec<Result<(), E>>> {
 		let mut have_pending_futures = false;
-		for fut_option in self.get_mut().0.iter_mut() {
-			let mut fut = match fut_option.take() {
-				None => continue,
-				Some(fut) => fut,
-			};
-			match Pin::new(&mut fut).poll(cx) {
-				Poll::Ready(()) => {},
-				Poll::Pending => {
-					have_pending_futures = true;
-					*fut_option = Some(fut);
+		let futures_state = &mut self.get_mut().futures_state;
+		for state in futures_state.iter_mut() {
+			match state {
+				ResultFuture::Pending(ref mut fut) => match Pin::new(fut).poll(cx) {
+					Poll::Ready(res) => {
+						*state = ResultFuture::Ready(res);
+					},
+					Poll::Pending => {
+						have_pending_futures = true;
+					},
 				},
+				ResultFuture::Ready(_) => continue,
 			}
 		}
+
 		if have_pending_futures {
 			Poll::Pending
 		} else {
-			Poll::Ready(())
+			let results = futures_state
+				.iter()
+				.filter_map(|e| match e {
+					ResultFuture::Ready(res) => Some(res),
+					ResultFuture::Pending(_) => {
+						debug_assert!(
+							false,
+							"All futures are expected to be ready if none are pending"
+						);
+						None
+					},
+				})
+				.copied()
+				.collect();
+			Poll::Ready(results)
 		}
 	}
 }