more cleanup

Author: patrick-ogrady

broadcast/src/buffered/engine.rs

@@ -158,7 +158,10 @@ where
             network.0,
             network.1,
         );
-        let peer_set_subscription = &mut self.peer_provider.subscribe().await;
+        let Ok(mut peer_set_subscription) = self.peer_provider.subscribe().await else {
+            debug!("peer set subscription failed");
+            return;
+        };
 
         select_loop! {
             self.context,

collector/fuzz/fuzz_targets/collector.rs

@@ -151,7 +151,7 @@ impl Handler for FuzzHandler {
     type Request = FuzzRequest;
     type Response = FuzzResponse;
 
-    async fn process(
+    fn process(
         &mut self,
         _origin: Self::PublicKey,
         request: Self::Request,
@@ -186,12 +186,7 @@ impl Monitor for FuzzMonitor {
     type PublicKey = PublicKey;
     type Response = FuzzResponse;
 
-    async fn collected(
-        &mut self,
-        _handler: Self::PublicKey,
-        _response: Self::Response,
-        _count: usize,
-    ) {
+    fn collected(&mut self, _handler: Self::PublicKey, _response: Self::Response, _count: usize) {
         self.collected_count += 1;
     }
 }
@@ -379,9 +374,7 @@ fn fuzz(input: FuzzInput) {
 
                     if let Some(handler) = handlers.get_mut(&handler_idx) {
                         let (tx, rx) = oneshot::channel();
-                        handler
-                            .process(peers[origin_idx].public_key(), request.clone(), tx)
-                            .await;
+                        handler.process(peers[origin_idx].public_key(), request.clone(), tx);
 
                         if should_respond {
                             if let Ok(response) = rx.await {
@@ -400,9 +393,7 @@ fn fuzz(input: FuzzInput) {
                     let handler_idx = (peer_idx as usize) % peers.len();
 
                     if let Some(monitor) = monitors.get_mut(&monitor_idx) {
-                        monitor
-                            .collected(peers[handler_idx].public_key(), response, count)
-                            .await;
+                        monitor.collected(peers[handler_idx].public_key(), response, count);
                     }
                 }
 

collector/src/lib.rs

@@ -15,7 +15,6 @@ commonware_macros::stability_scope!(ALPHA {
     use commonware_cryptography::{Committable, Digestible, PublicKey};
     use commonware_p2p::Recipients;
     use commonware_utils::channel::oneshot;
-    use std::future::Future;
 
     pub mod p2p;
 
@@ -53,15 +52,17 @@ commonware_macros::stability_scope!(ALPHA {
             + Digestible<Digest = <Self::Request as Digestible>::Digest>
             + Codec;
 
-        /// Processes a `request` from an [Originator] and (optionally) send a response.
+        /// Processes a `request` from an [Originator] and (optionally) sends a response.
         ///
-        /// If no response is needed, the `responder` should be dropped.
+        /// Implementations should return promptly. If processing requires async work,
+        /// enqueue it and use `response` to send the result later. If no response is
+        /// needed, the `response` should be dropped.
         fn process(
             &mut self,
             origin: Self::PublicKey,
             request: Self::Request,
             response: oneshot::Sender<Self::Response>,
-        ) -> impl Future<Output = ()> + Send;
+        );
     }
 
     /// A [Monitor] collects responses from [Handler]s.
@@ -75,12 +76,15 @@ commonware_macros::stability_scope!(ALPHA {
         /// Called for each response collected with the number of responses collected so far for
         /// the same commitment.
         ///
+        /// Implementations should return promptly. If observing a collection event
+        /// requires async work, enqueue it before returning.
+        ///
         /// [Monitor::collected] is only called once per `handler`.
         fn collected(
             &mut self,
             handler: Self::PublicKey,
             response: Self::Response,
             count: usize,
-        ) -> impl Future<Output = ()> + Send;
+        );
     }
 });

collector/src/p2p/engine.rs

@@ -125,7 +125,6 @@ where
 
         // Create futures pools
         let mut processed: Pool<Result<(P, Rs), oneshot::error::RecvError>> = Pool::default();
-        let mut monitored: Pool<()> = Pool::default();
         select_loop! {
             self.context,
             on_stopped => {
@@ -162,7 +161,6 @@ where
                 // Send the response
                 let _ = res_tx.send(Recipients::One(peer), reply, self.priority_response);
             },
-            _ = monitored.next_completed() => {},
 
             // Request from an originator
             message = req_rx.recv() => {
@@ -186,9 +184,8 @@ where
 
                 // Handle the request
                 let (tx, rx) = oneshot::channel();
-                let mut handler = self.handler.clone();
+                self.handler.process(peer.clone(), msg, tx);
                 processed.push(async move {
-                    handler.process(peer.clone(), msg, tx).await;
                     Ok((peer, rx.await?))
                 });
             },
@@ -227,11 +224,8 @@ where
                 }
 
                 // Send the response to the monitor
-                let mut monitor = self.monitor.clone();
                 let count = responses.1.len();
-                monitored.push(async move {
-                    monitor.collected(peer, msg, count).await;
-                });
+                self.monitor.collected(peer, msg, count);
             },
         }
     }

collector/src/p2p/mocks/handler.rs

@@ -58,7 +58,7 @@ impl crate::Handler for Handler {
     type Request = Request;
     type Response = Response;
 
-    async fn process(
+    fn process(
         &mut self,
         origin: Self::PublicKey,
         request: Self::Request,

collector/src/p2p/mocks/monitor.rs

@@ -34,7 +34,7 @@ impl crate::Monitor for Monitor {
     type PublicKey = PublicKey;
     type Response = Response;
 
-    async fn collected(
+    fn collected(
         &mut self,
         handler: Self::PublicKey,
         response: Self::Response,

consensus/src/marshal/coding/shards/engine.rs

@@ -439,7 +439,10 @@ where
             );
         // Keep the handle alive to prevent the background receiver from being aborted.
         let _receiver_handle = receiver_service.start();
-        let mut peer_set_subscription = self.peer_provider.subscribe().await;
+        let Ok(mut peer_set_subscription) = self.peer_provider.subscribe().await else {
+            debug!("peer set subscription failed");
+            return;
+        };
 
         select_loop! {
             self.context,

examples/reshare/src/dkg/actor.rs

@@ -654,7 +654,10 @@ mod tests {
     use commonware_math::algebra::Random;
     use commonware_p2p::{utils::mocks::inert_channel, PeerSetSubscription, Provider};
     use commonware_runtime::{deterministic, Runner, Supervisor as _};
-    use commonware_utils::{channel::mpsc, N3f1, NZUsize, TryCollect, NZU32};
+    use commonware_utils::{
+        channel::{mpsc, oneshot},
+        N3f1, NZUsize, TryCollect, NZU32,
+    };
     use core::marker::PhantomData;
     use std::collections::BTreeMap;
 
@@ -670,13 +673,17 @@ mod tests {
     impl<P: PublicKey> Provider for NoopManager<P> {
         type PublicKey = P;
 
-        async fn peer_set(&mut self, _: u64) -> Option<TrackedPeers<Self::PublicKey>> {
-            None
+        fn peer_set(&mut self, _: u64) -> oneshot::Receiver<Option<TrackedPeers<Self::PublicKey>>> {
+            let (sender, receiver) = oneshot::channel();
+            let _ = sender.send(None);
+            receiver
         }
 
-        async fn subscribe(&mut self) -> PeerSetSubscription<Self::PublicKey> {
+        fn subscribe(&mut self) -> oneshot::Receiver<PeerSetSubscription<Self::PublicKey>> {
             let (_, rx) = mpsc::unbounded_channel();
-            rx
+            let (sender, receiver) = oneshot::channel();
+            let _ = sender.send(rx);
+            receiver
         }
     }
 

p2p/src/authenticated/discovery/actors/dialer.rs

@@ -19,9 +19,9 @@ use commonware_runtime::{
     BufferPooler, Clock, ContextCell, Handle, Metrics, Network, Resolver, SinkOf, Spawner,
     StreamOf,
 };
-use commonware_utils::futures::Pool;
-use futures::future::{self, Either};
 use commonware_stream::encrypted::{dial, Config as StreamConfig};
+use commonware_utils::channel::oneshot;
+use futures::future::{self, Either};
 use rand::seq::SliceRandom;
 use rand_core::CryptoRngCore;
 use std::time::{Duration, SystemTime};
@@ -32,10 +32,24 @@ type SupervisorMailbox<E, C> =
     Mailbox<spawner::Message<SinkOf<E>, StreamOf<E>, <C as Signer>::PublicKey>>;
 
 enum TrackerQuery<C: PublicKey> {
+    Dialable(oneshot::Receiver<Dialable<C>>),
+    Dial(oneshot::Receiver<Option<Reservation<C>>>),
+}
+
+enum TrackerReply<C: PublicKey> {
     Dialable(Dialable<C>),
     Dial(Option<Reservation<C>>),
 }
 
+impl<C: PublicKey> TrackerQuery<C> {
+    async fn recv(&mut self) -> TrackerReply<C> {
+        match self {
+            Self::Dialable(receiver) => TrackerReply::Dialable(receiver.await.unwrap_or_default()),
+            Self::Dial(receiver) => TrackerReply::Dial(receiver.await.ok().flatten()),
+        }
+    }
+}
+
 /// Configuration for the dialer actor.
 pub struct Config<C: Signer> {
     /// Configuration for the stream.
@@ -173,42 +187,41 @@ impl<
         mut supervisor: SupervisorMailbox<E, C>,
     ) {
         let mut dial_deadline = self.context.current();
-        let mut queries = Pool::default();
-        let mut query_pending = false;
+        let mut query: Option<TrackerQuery<C::PublicKey>> = None;
         select_loop! {
             self.context,
             on_start => {
-                let wait_for_deadline = if query_pending {
+                let wait_for_deadline = if query.is_some() {
                     Either::Left(future::pending())
                 } else {
                     Either::Right(self.context.sleep_until(dial_deadline))
                 };
+                let wait_for_query = match &mut query {
+                    Some(query) => Either::Left(query.recv()),
+                    None => Either::Right(future::pending()),
+                };
             },
             on_stopped => {
                 debug!("context shutdown, stopping dialer");
             },
             _ = wait_for_deadline => {
                 let now = self.context.current();
                 if self.queue.is_empty() {
-                    let tracker = tracker.clone();
-                    queries.push(async move { TrackerQuery::Dialable(tracker.dialable().await) });
-                    query_pending = true;
+                    query = Some(TrackerQuery::Dialable(tracker.dialable()));
                     continue;
                 }
 
                 if let Some(peer) = self.queue.pop() {
-                    let tracker = tracker.clone();
-                    queries.push(async move { TrackerQuery::Dial(tracker.dial(peer).await) });
-                    query_pending = true;
+                    query = Some(TrackerQuery::Dial(tracker.dial(peer)));
                 } else {
                     dial_deadline = self.empty_queue_deadline(now, None);
                 }
             },
-            query = queries.next_completed() => {
-                query_pending = false;
+            reply = wait_for_query => {
+                query = None;
                 let now = self.context.current();
-                match query {
-                    TrackerQuery::Dialable(dialable) => {
+                match reply {
+                    TrackerReply::Dialable(dialable) => {
                         self.queue = dialable.peers;
                         self.queue.shuffle(self.context.as_mut());
                         dial_deadline = if self.queue.is_empty() {
@@ -217,7 +230,7 @@ impl<
                             now
                         };
                     }
-                    TrackerQuery::Dial(result) => {
+                    TrackerReply::Dial(result) => {
                         if let Some(reservation) = result {
                             self.dial_peer(reservation, &mut supervisor);
                             dial_deadline = now + self.dial_frequency;

p2p/src/authenticated/discovery/actors/listener.rs

@@ -97,7 +97,10 @@ impl<E: Spawner + BufferPooler + Clock + Network + CryptoRngCore + Metrics, C: S
     ) {
         let (peer, send, recv) = match listen(
             context,
-            |peer| tracker.acceptable(peer),
+            |peer| {
+                let accepted = tracker.acceptable(peer);
+                async move { accepted.await.unwrap_or(false) }
+            },
             stream_cfg,
             stream,
             sink,
@@ -113,7 +116,7 @@ impl<E: Spawner + BufferPooler + Clock + Network + CryptoRngCore + Metrics, C: S
         debug!(?peer, ?address, "completed handshake");
 
         // Attempt to claim the connection
-        let Some(reservation) = tracker.listen(peer.clone()).await else {
+        let Some(reservation) = tracker.listen(peer.clone()).await.ok().flatten() else {
             debug!(?peer, ?address, "unable to reserve connection to peer");
             return;
         };