[examples] Migrate to Actor Mailbox (#3806)

Author: patrick-ogrady

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

@@ -166,7 +166,7 @@ use commonware_runtime::{
 };
 use commonware_utils::{
     bitmap::BitMap,
-    channel::{fallible::OneshotExt, mpsc, oneshot},
+    channel::{fallible::OneshotExt, oneshot},
     ordered::{Quorum, Set},
 };
 use rand::Rng;
@@ -251,9 +251,9 @@ where
     /// Capacity of the channel between the background receiver and the engine.
     ///
     /// The background receiver decodes incoming network messages in a separate
-    /// task and forwards them to the engine over an `mpsc` channel with this
+    /// task and forwards them to the engine over a mailbox with this
     /// capacity.
-    pub background_channel_capacity: usize,
+    pub background_channel_capacity: NonZeroUsize,
 
     /// Provider for peer set information. Pre-leader shards are buffered per
     /// peer only while that peer appears in the
@@ -324,7 +324,7 @@ where
     latest_primary_peers: Set<P>,
 
     /// Capacity of the background receiver channel.
-    background_channel_capacity: usize,
+    background_channel_capacity: NonZeroUsize,
 
     /// An ephemeral cache of reconstructed blocks, keyed by commitment.
     ///
@@ -412,8 +412,8 @@ where
             self.context.network_buffer_pool().clone(),
             sender,
         );
-        let (receiver_service, mut receiver): (_, mpsc::Receiver<(P, Shard<C, H>)>) =
-            WrappedBackgroundReceiver::new(
+        let (receiver_service, mut receiver) =
+            WrappedBackgroundReceiver::<_, P, X, _, Shard<C, H>>::new(
                 self.context.child("shard_ingress"),
                 receiver,
                 self.shard_codec_cfg.clone(),
@@ -1801,7 +1801,7 @@ mod tests {
                         strategy: STRATEGY,
                         mailbox_size: NZUsize!(1024),
                         peer_buffer_size: NZUsize!(64),
-                        background_channel_capacity: 1024,
+                        background_channel_capacity: NZUsize!(1024),
                         peer_provider: oracle.manager(),
                     };
 
@@ -1840,7 +1840,7 @@ mod tests {
                         strategy: STRATEGY,
                         mailbox_size: NZUsize!(1024),
                         peer_buffer_size: NZUsize!(64),
-                        background_channel_capacity: 1024,
+                        background_channel_capacity: NZUsize!(1024),
                         peer_provider: oracle.manager(),
                     };
 
@@ -3624,7 +3624,7 @@ mod tests {
                 strategy: STRATEGY,
                 mailbox_size: NZUsize!(1024),
                 peer_buffer_size: NZUsize!(64),
-                background_channel_capacity: 1024,
+                background_channel_capacity: NZUsize!(1024),
                 peer_provider: oracle.manager(),
             };
 
@@ -3753,7 +3753,7 @@ mod tests {
                 strategy: STRATEGY,
                 mailbox_size: NZUsize!(1024),
                 peer_buffer_size: NZUsize!(64),
-                background_channel_capacity: 1024,
+                background_channel_capacity: NZUsize!(1024),
                 peer_provider: oracle.manager(),
             };
             let (broadcaster_engine, broadcaster_mailbox) =
@@ -3776,7 +3776,7 @@ mod tests {
                 strategy: STRATEGY,
                 mailbox_size: NZUsize!(1024),
                 peer_buffer_size: NZUsize!(64),
-                background_channel_capacity: 1024,
+                background_channel_capacity: NZUsize!(1024),
                 peer_provider: oracle.manager(),
             };
             let (receiver_engine, receiver_mailbox) =
@@ -4555,7 +4555,7 @@ mod tests {
                 strategy: STRATEGY,
                 mailbox_size: NZUsize!(1024),
                 peer_buffer_size: NZUsize!(64),
-                background_channel_capacity: 1024,
+                background_channel_capacity: NZUsize!(1024),
                 peer_provider: oracle.manager(),
             };
 
@@ -4658,7 +4658,7 @@ mod tests {
                 strategy: STRATEGY,
                 mailbox_size: NZUsize!(16),
                 peer_buffer_size: NZUsize!(4),
-                background_channel_capacity: 16,
+                background_channel_capacity: NZUsize!(16),
                 peer_provider: oracle.manager(),
             };
 
@@ -4791,7 +4791,7 @@ mod tests {
                 strategy: STRATEGY,
                 mailbox_size: NZUsize!(1024),
                 peer_buffer_size: NZUsize!(64),
-                background_channel_capacity: 1024,
+                background_channel_capacity: NZUsize!(1024),
                 peer_provider: oracle.manager(),
             };
 
@@ -4946,7 +4946,7 @@ mod tests {
                 strategy: STRATEGY,
                 mailbox_size: NZUsize!(1024),
                 peer_buffer_size: NZUsize!(64),
-                background_channel_capacity: 1024,
+                background_channel_capacity: NZUsize!(1024),
                 peer_provider: oracle.manager(),
             };
 

consensus/src/marshal/mocks/harness.rs

@@ -2507,7 +2507,7 @@ impl TestHarness for CodingHarness {
             strategy: Sequential,
             mailbox_size: NZUsize!(10),
             peer_buffer_size: NZUsize!(64),
-            background_channel_capacity: 1024,
+            background_channel_capacity: NZUsize!(1024),
             peer_provider: oracle.manager(),
         };
         let (shard_engine, shard_mailbox) =
@@ -2694,7 +2694,7 @@ impl TestHarness for CodingHarness {
             strategy: Sequential,
             mailbox_size: NZUsize!(10),
             peer_buffer_size: NZUsize!(64),
-            background_channel_capacity: 1024,
+            background_channel_capacity: NZUsize!(1024),
             peer_provider: oracle.manager(),
         };
         let (shard_engine, shard_mailbox) =

examples/reshare/src/dkg/actor.rs

@@ -304,7 +304,7 @@ where
                 share: epoch_state.share.clone(),
                 dealers: dealers.clone(),
             };
-            orchestrator.enter(transition).await;
+            orchestrator.enter(transition);
 
             // Register a channel for this round
             let (mut round_sender, mut round_receiver) = dkg_mux
@@ -576,7 +576,7 @@ where
                         };
 
                         // Exit the engine for this epoch now that the boundary is finalized
-                        orchestrator.exit(epoch).await;
+                        orchestrator.exit(epoch);
 
                         // If the update is stop, wait forever.
                         if let PostUpdate::Stop = callback.on_update(update).await {
@@ -773,7 +773,8 @@ mod tests {
                 },
             );
             let (sender, receiver) = inert_channel(&peer_config.participants);
-            let (orchestrator_sender, mut orchestrator_receiver) = mpsc::channel(4);
+            let (orchestrator_sender, mut orchestrator_receiver) =
+                mailbox::new(context.child("orchestrator_mailbox"), NZUsize!(4));
             actor.start(
                 None,
                 None,

examples/reshare/src/engine.rs

@@ -303,7 +303,7 @@ where
                 marshal: marshal_mailbox,
                 strategy: config.strategy.clone(),
                 muxer_size: MAILBOX_SIZE.get(),
-                mailbox_size: MAILBOX_SIZE.get(),
+                mailbox_size: MAILBOX_SIZE,
                 partition_prefix: format!("{}_consensus", config.partition_prefix),
                 _phantom: PhantomData,
             },

examples/reshare/src/orchestrator/actor.rs

@@ -5,6 +5,7 @@ use crate::{
     orchestrator::{ingress::Message, Mailbox},
     BLOCKS_PER_EPOCH,
 };
+use commonware_actor::mailbox;
 use commonware_consensus::{
     marshal::{core::Mailbox as MarshalMailbox, standard::Standard},
     simplex::{self, elector::Config as Elector, scheme, types::Context, Plan},
@@ -17,7 +18,7 @@ use commonware_cryptography::{
 use commonware_macros::select_loop;
 use commonware_p2p::{
     utils::mux::{Builder, MuxHandle, Muxer},
-    Blocker, Receiver, Sender,
+    Blocker, Sender,
 };
 use commonware_parallel::Strategy;
 use commonware_runtime::{
@@ -26,9 +27,9 @@ use commonware_runtime::{
     telemetry::metrics::{Gauge, GaugeExt, MetricsExt as _},
     BufferPooler, Clock, ContextCell, Handle, Metrics, Network, Spawner, Storage,
 };
-use commonware_utils::{channel::mpsc, vec::NonEmptyVec, NZUsize, NZU16};
+use commonware_utils::{vec::NonEmptyVec, NZUsize, NZU16};
 use rand_core::CryptoRngCore;
-use std::{collections::BTreeMap, marker::PhantomData, time::Duration};
+use std::{collections::BTreeMap, marker::PhantomData, num::NonZeroUsize, time::Duration};
 use tracing::{debug, info, warn};
 
 /// Configuration for the orchestrator.
@@ -51,7 +52,7 @@ where
     pub strategy: T,
 
     pub muxer_size: usize,
-    pub mailbox_size: usize,
+    pub mailbox_size: NonZeroUsize,
 
     // Partition prefix used for orchestrator metadata persistence
     pub partition_prefix: String,
@@ -74,7 +75,7 @@ where
     Provider<S, C>: EpochProvider<Variant = V, PublicKey = C::PublicKey, Scheme = S>,
 {
     context: ContextCell<E>,
-    mailbox: mpsc::Receiver<Message<V, C::PublicKey>>,
+    mailbox: mailbox::Receiver<Message<V, C::PublicKey>>,
     application: A,
 
     oracle: B,
@@ -109,7 +110,7 @@ where
         context: E,
         config: Config<B, V, C, H, A, S, L, T>,
     ) -> (Self, Mailbox<V, C::PublicKey>) {
-        let (sender, mailbox) = mpsc::channel(config.mailbox_size);
+        let (sender, mailbox) = mailbox::new(context.child("mailbox"), config.mailbox_size);
         let page_cache_ref = CacheRef::from_pooler(&context, NZU16!(16_384), NZUsize!(10_000));
 
         // Register latest_epoch gauge for Grafana integration
@@ -138,15 +139,15 @@ where
         mut self,
         votes: (
             impl Sender<PublicKey = C::PublicKey>,
-            impl Receiver<PublicKey = C::PublicKey>,
+            impl commonware_p2p::Receiver<PublicKey = C::PublicKey>,
         ),
         certificates: (
             impl Sender<PublicKey = C::PublicKey>,
-            impl Receiver<PublicKey = C::PublicKey>,
+            impl commonware_p2p::Receiver<PublicKey = C::PublicKey>,
         ),
         resolver: (
             impl Sender<PublicKey = C::PublicKey>,
-            impl Receiver<PublicKey = C::PublicKey>,
+            impl commonware_p2p::Receiver<PublicKey = C::PublicKey>,
         ),
     ) -> Handle<()> {
         spawn_cell!(self.context, self.run(votes, certificates, resolver,))
@@ -156,15 +157,15 @@ where
         mut self,
         (vote_sender, vote_receiver): (
             impl Sender<PublicKey = C::PublicKey>,
-            impl Receiver<PublicKey = C::PublicKey>,
+            impl commonware_p2p::Receiver<PublicKey = C::PublicKey>,
         ),
         (certificate_sender, certificate_receiver): (
             impl Sender<PublicKey = C::PublicKey>,
-            impl Receiver<PublicKey = C::PublicKey>,
+            impl commonware_p2p::Receiver<PublicKey = C::PublicKey>,
         ),
         (resolver_sender, resolver_receiver): (
             impl Sender<PublicKey = C::PublicKey>,
-            impl Receiver<PublicKey = C::PublicKey>,
+            impl commonware_p2p::Receiver<PublicKey = C::PublicKey>,
         ),
     ) {
         // Start muxers for each physical channel used by consensus
@@ -287,15 +288,15 @@ where
         scheme: S,
         vote_mux: &mut MuxHandle<
             impl Sender<PublicKey = C::PublicKey>,
-            impl Receiver<PublicKey = C::PublicKey>,
+            impl commonware_p2p::Receiver<PublicKey = C::PublicKey>,
         >,
         certificate_mux: &mut MuxHandle<
             impl Sender<PublicKey = C::PublicKey>,
-            impl Receiver<PublicKey = C::PublicKey>,
+            impl commonware_p2p::Receiver<PublicKey = C::PublicKey>,
         >,
         resolver_mux: &mut MuxHandle<
             impl Sender<PublicKey = C::PublicKey>,
-            impl Receiver<PublicKey = C::PublicKey>,
+            impl commonware_p2p::Receiver<PublicKey = C::PublicKey>,
         >,
     ) -> Handle<()> {
         // Start the new engine

examples/reshare/src/orchestrator/ingress.rs

@@ -1,11 +1,13 @@
 //! Inbound communication channel for epoch transitions.
 
+use commonware_actor::mailbox::{Policy, Sender};
 use commonware_consensus::types::Epoch;
 use commonware_cryptography::{
     bls12381::primitives::{group, sharing::Sharing, variant::Variant},
     PublicKey,
 };
-use commonware_utils::{channel::mpsc, ordered::Set};
+use commonware_utils::ordered::Set;
+use std::collections::VecDeque;
 use tracing::error;
 
 /// Messages that can be sent to the orchestrator.
@@ -14,6 +16,36 @@ pub enum Message<V: Variant, P: PublicKey> {
     Exit(Epoch),
 }
 
+impl<V: Variant, P: PublicKey> Policy for Message<V, P> {
+    type Overflow = VecDeque<Self>;
+
+    fn handle(overflow: &mut VecDeque<Self>, message: Self) -> bool {
+        match message {
+            Self::Enter(transition) => {
+                let epoch = transition.epoch;
+                if let Some(index) = overflow
+                    .iter()
+                    .position(|pending| matches!(pending, Self::Exit(pending) if *pending == epoch))
+                {
+                    overflow.remove(index);
+                } else {
+                    overflow.push_back(Self::Enter(transition));
+                }
+            }
+            Self::Exit(epoch) => {
+                if let Some(index) = overflow.iter().position(
+                    |pending| matches!(pending, Self::Enter(pending) if pending.epoch == epoch),
+                ) {
+                    overflow.remove(index);
+                } else {
+                    overflow.push_back(Self::Exit(epoch));
+                }
+            }
+        }
+        true
+    }
+}
+
 /// A notification of an epoch transition.
 pub struct EpochTransition<V: Variant, P: PublicKey> {
     /// The epoch to transition to.
@@ -29,24 +61,24 @@ pub struct EpochTransition<V: Variant, P: PublicKey> {
 /// Inbound communication channel for epoch transitions.
 #[derive(Debug, Clone)]
 pub struct Mailbox<V: Variant, P: PublicKey> {
-    sender: mpsc::Sender<Message<V, P>>,
+    sender: Sender<Message<V, P>>,
 }
 
 impl<V: Variant, P: PublicKey> Mailbox<V, P> {
     /// Create a new [Mailbox].
-    pub const fn new(sender: mpsc::Sender<Message<V, P>>) -> Self {
+    pub const fn new(sender: Sender<Message<V, P>>) -> Self {
         Self { sender }
     }
 
-    pub async fn enter(&mut self, transition: EpochTransition<V, P>) {
-        if let Err(err) = self.sender.send(Message::Enter(transition)).await {
-            error!(?err, "failed to send epoch transition");
+    pub fn enter(&mut self, transition: EpochTransition<V, P>) {
+        if !self.sender.enqueue(Message::Enter(transition)).accepted() {
+            error!("failed to send epoch transition");
         }
     }
 
-    pub async fn exit(&mut self, epoch: Epoch) {
-        if let Err(err) = self.sender.send(Message::Exit(epoch)).await {
-            error!(?err, "failed to send epoch exit");
+    pub fn exit(&mut self, epoch: Epoch) {
+        if !self.sender.enqueue(Message::Exit(epoch)).accepted() {
+            error!("failed to send epoch exit");
         }
     }
 }