[p2p] Migrate to Mailbox (#3795)

Author: patrick-ogrady

Cargo.lock

@@ -92,11 +92,12 @@ impl Message {
 impl mailbox::Policy for Message {
     type Overflow = VecDeque<Self>;
 
-    fn handle(overflow: &mut VecDeque<Self>, message: Self) {
+    fn handle(overflow: &mut VecDeque<Self>, message: Self) -> bool {
         match message.policy {
-            Policy::Drop => {}
+            Policy::Drop => false,
             Policy::Spill => {
                 overflow.push_back(message);
+                true
             }
             Policy::Replace => {
                 if let Some(pending) = overflow
@@ -108,6 +109,7 @@ impl mailbox::Policy for Message {
                 } else {
                     overflow.push_back(message);
                 }
+                true
             }
         }
     }
@@ -256,7 +258,7 @@ fn bench_overflow_drop(c: &mut Criterion) {
             |(sender, _receiver)| {
                 for _ in 0..MESSAGES {
                     let result = sender.enqueue(black_box(Message::drop()));
-                    assert_eq!(result, Feedback::Backoff);
+                    assert_eq!(result, Feedback::Rejected);
                     black_box(result);
                 }
             },

actor/src/benches/mailbox.rs

@@ -15,8 +15,10 @@ commonware_macros::stability_scope!(BETA {
     pub enum Feedback {
         /// The work was accepted within the configured capacity.
         Ok,
-        /// The submission exceeded the configured capacity and requests sender backoff.
+        /// The submission exceeded configured capacity but was handled by the overflow policy.
         Backoff,
+        /// The submission exceeded configured capacity and was rejected by the overflow policy.
+        Rejected,
         /// The endpoint is closed.
         Closed,
     }

actor/src/lib.rs

@@ -93,9 +93,15 @@ pub trait Policy: Sized {
 
     /// Handle `message` when it cannot enter the bounded ready queue immediately.
     ///
-    /// Messages already in the ready queue are not provided here. Policy changes only apply to
-    /// overflow retained beyond ready capacity. Policies may append, remove, replace, reorder, or
-    /// clear overflow, and are responsible for bounding it when a hard memory limit is required.
+    /// Returns `true` when the policy considered the message's effects. This includes
+    /// retaining the message, coalescing it with retained work, replacing older retained work,
+    /// or deliberately doing no work because the message is already satisfied, superseded, or no
+    /// longer needed (for example, a request whose response channel is already closed). These
+    /// no-op cases are still handled: there is no remaining work, so there is no loss to report.
+    ///
+    /// Returns `false` only when the policy rejects the message under backpressure. This is the
+    /// lossy case: the submitted work was not semantically handled, and callers that care should
+    /// retry or treat the submission as failed.
     ///
     /// # Warning
     ///
@@ -106,7 +112,7 @@ pub trait Policy: Sized {
     /// This method should not unwind after mutating `overflow`. A panic, including one from a
     /// destructor triggered while editing `overflow`, can leave retained overflow data stranded in
     /// the mailbox.
-    fn handle(overflow: &mut Self::Overflow, message: Self);
+    fn handle(overflow: &mut Self::Overflow, message: Self) -> bool;
 }
 
 // `activity` packs the published overflow state and in-flight overflow
@@ -299,9 +305,13 @@ impl<T: Policy> OverflowState<T> {
         };
 
         // Preserve overflow order, or handle a still-full ready queue.
-        T::handle(&mut queue, message);
+        let handled = T::handle(&mut queue, message);
         mutation.publish(queue.is_empty());
-        Feedback::Backoff
+        if handled {
+            Feedback::Backoff
+        } else {
+            Feedback::Rejected
+        }
     }
 
     fn refill(&self, ready: &Ready<T>) {
@@ -445,10 +455,10 @@ impl<T: Policy> Sender<T> {
             self.state.backoff.inc();
         }
 
-        // Wake on any handled enqueue because a receiver may have skipped
-        // refill while this overflow mutation was active. By the time we wake,
-        // the mutation has published its overflow state. Spurious wakes are
-        // acceptable.
+        // Wake after any non-closed slow-path enqueue because a receiver may
+        // have skipped refill while this overflow mutation was active. By the
+        // time we wake, the mutation has published its overflow state. Spurious
+        // wakes are acceptable.
         if feedback != Feedback::Closed {
             self.state.waker.wake();
         }
@@ -631,7 +641,7 @@ mod tests {
     impl Policy for Message {
         type Overflow = VecDeque<Self>;
 
-        fn handle(overflow: &mut VecDeque<Self>, message: Self) {
+        fn handle(overflow: &mut VecDeque<Self>, message: Self) -> bool {
             match message {
                 Self::Update(value) => {
                     if let Some(index) = overflow
@@ -641,21 +651,24 @@ mod tests {
                         overflow.remove(index);
                     }
                     overflow.push_back(Self::Update(value));
+                    true
                 }
                 Self::Required(_) | Self::Buffered(_) => {
                     overflow.push_back(message);
+                    true
                 }
                 Self::Hint(value) => {
                     let Some(index) = overflow
                         .iter()
                         .rposition(|pending| matches!(pending, Self::Update(_)))
                     else {
-                        return;
+                        return true;
                     };
                     overflow.remove(index);
                     overflow.push_back(Self::Hint(value));
+                    true
                 }
-                Self::Vote(_) => {}
+                Self::Vote(_) => false,
             }
         }
     }
@@ -667,8 +680,9 @@ mod tests {
     impl Policy for Ack {
         type Overflow = VecDeque<Self>;
 
-        fn handle(overflow: &mut VecDeque<Self>, message: Self) {
+        fn handle(overflow: &mut VecDeque<Self>, message: Self) -> bool {
             overflow.push_back(message);
+            true
         }
     }
 
@@ -738,7 +752,7 @@ mod tests {
     async fn full_inbox_rejects_non_replaceable_message() {
         let (sender, mut receiver) = new(NZUsize!(1));
         assert_eq!(sender.enqueue(Message::Vote(1)), Feedback::Ok);
-        assert_eq!(sender.enqueue(Message::Vote(2)), Feedback::Backoff);
+        assert_eq!(sender.enqueue(Message::Vote(2)), Feedback::Rejected);
 
         assert_eq!(receiver.recv().await, Some(Message::Vote(1)));
     }
@@ -780,6 +794,25 @@ mod tests {
         });
     }
 
+    #[test]
+    fn rejected_feedback_is_not_accepted_or_counted_as_backoff() {
+        let executor = deterministic::Runner::default();
+        executor.start(|context| async move {
+            let (sender, _receiver) = super::new(context.child("mailbox"), NZUsize!(1));
+            assert_eq!(sender.enqueue(Message::Vote(1)), Feedback::Ok);
+            let feedback = sender.enqueue(Message::Vote(2));
+
+            assert_eq!(feedback, Feedback::Rejected);
+            assert!(!feedback.accepted());
+
+            let buffer = context.encode();
+            assert!(
+                buffer.contains("mailbox_backoff_total 0"),
+                "unexpected backoff count in metrics: {buffer}"
+            );
+        });
+    }
+
     #[test]
     fn try_recv_drains_buffered_messages_after_senders_drop() {
         let (sender, mut receiver) = new(NZUsize!(1));
@@ -1007,9 +1040,10 @@ mod tests {
     impl Policy for ClearingMessage {
         type Overflow = VecDeque<Self>;
 
-        fn handle(overflow: &mut VecDeque<Self>, message: Self) {
+        fn handle(overflow: &mut VecDeque<Self>, message: Self) -> bool {
             overflow.push_back(message);
             overflow.clear();
+            true
         }
     }
 
@@ -1038,8 +1072,9 @@ mod tests {
     impl Policy for SpillMessage {
         type Overflow = VecDeque<Self>;
 
-        fn handle(overflow: &mut VecDeque<Self>, message: Self) {
+        fn handle(overflow: &mut VecDeque<Self>, message: Self) -> bool {
             overflow.push_back(message);
+            true
         }
     }
 
@@ -1131,11 +1166,12 @@ mod loom_tests {
     impl Policy for Message {
         type Overflow = VecDeque<Self>;
 
-        fn handle(overflow: &mut VecDeque<Self>, message: Self) {
+        fn handle(overflow: &mut VecDeque<Self>, message: Self) -> bool {
             match message {
-                Self::Drop(_) => {}
+                Self::Drop(_) => false,
                 Self::Spill(_) => {
                     overflow.push_back(message);
+                    true
                 }
             }
         }
@@ -1144,7 +1180,7 @@ mod loom_tests {
     impl Policy for OrderedMessage {
         type Overflow = VecDeque<Self>;
 
-        fn handle(overflow: &mut VecDeque<Self>, message: Self) {
+        fn handle(overflow: &mut VecDeque<Self>, message: Self) -> bool {
             let gate = match &message {
                 Self::Item(_) => None,
                 Self::Coordinated(_, gate) => Some(gate.clone()),
@@ -1156,15 +1192,16 @@ mod loom_tests {
                     thread::yield_now();
                 }
             }
+            true
         }
     }
 
     impl Policy for ReplacingMessage {
         type Overflow = VecDeque<Self>;
 
-        fn handle(overflow: &mut VecDeque<Self>, message: Self) {
+        fn handle(overflow: &mut VecDeque<Self>, message: Self) -> bool {
             match message {
-                Self::FillReady => {}
+                Self::FillReady => false,
                 Self::Replace(_) => {
                     if let Some(pending) = overflow
                         .iter_mut()
@@ -1175,6 +1212,7 @@ mod loom_tests {
                     } else {
                         overflow.push_back(message);
                     }
+                    true
                 }
             }
         }
@@ -1183,16 +1221,18 @@ mod loom_tests {
     impl Policy for TrackedMessage {
         type Overflow = VecDeque<Self>;
 
-        fn handle(overflow: &mut VecDeque<Self>, message: Self) {
+        fn handle(overflow: &mut VecDeque<Self>, message: Self) -> bool {
             overflow.push_back(message);
+            true
         }
     }
 
     impl Policy for CyclicMessage {
         type Overflow = VecDeque<Self>;
 
-        fn handle(overflow: &mut VecDeque<Self>, message: Self) {
+        fn handle(overflow: &mut VecDeque<Self>, message: Self) -> bool {
             overflow.push_back(message);
+            true
         }
     }
 

actor/src/mailbox.rs

@@ -188,7 +188,7 @@ where
                     message,
                 } => {
                     trace!("mailbox: broadcast");
-                    self.handle_broadcast(&mut sender, recipients, message).await;
+                    self.handle_broadcast(&mut sender, recipients, message);
                 }
                 Message::Subscribe { digest, responder } => {
                     trace!("mailbox: subscribe");
@@ -232,7 +232,7 @@ where
     ////////////////////////////////////////
 
     /// Handles a `broadcast` request from the application.
-    async fn handle_broadcast<Sr: Sender<PublicKey = P>>(
+    fn handle_broadcast<Sr: Sender<PublicKey = P>>(
         &mut self,
         sender: &mut WrappedSender<Sr, M>,
         recipients: Recipients<P>,
@@ -243,9 +243,7 @@ where
         let _ = self.insert_message(self.public_key.clone(), digest, msg.clone());
 
         // Broadcast the message to the network
-        if let Err(err) = sender.send(recipients, msg, self.priority).await {
-            error!(?err, "failed to send message");
-        }
+        sender.send(recipients, msg, self.priority);
     }
 
     /// Handles a `subscribe` request from the application.

broadcast/src/buffered/engine.rs

@@ -77,12 +77,13 @@ impl<P: PublicKey, M: Digestible> Overflow<Message<P, M>> for Pending<P, M> {
 impl<P: PublicKey, M: Digestible> Policy for Message<P, M> {
     type Overflow = Pending<P, M>;
 
-    fn handle(overflow: &mut Self::Overflow, message: Self) {
+    fn handle(overflow: &mut Self::Overflow, message: Self) -> bool {
         if message.response_closed() {
-            return;
+            return true;
         }
 
         overflow.0.push_back(message);
+        true
     }
 }