[actor] Add Backoff Metrics (#3802)

Author: patrick-ogrady

.github/workflows/publish.yml

@@ -60,11 +60,6 @@ jobs:
       continue-on-error: true
       env:
         CARGO_REGISTRY_TOKEN: ${{ secrets.CARGO_REGISTRY_TOKEN }}
-    - name: Publish actor
-      run: cargo publish --manifest-path actor/Cargo.toml
-      continue-on-error: true
-      env:
-        CARGO_REGISTRY_TOKEN: ${{ secrets.CARGO_REGISTRY_TOKEN }}
     - name: Publish math
       run: cargo publish --manifest-path math/Cargo.toml
       continue-on-error: true
@@ -85,6 +80,11 @@ jobs:
       continue-on-error: true
       env:
         CARGO_REGISTRY_TOKEN: ${{ secrets.CARGO_REGISTRY_TOKEN }}
+    - name: Publish actor
+      run: cargo publish --manifest-path actor/Cargo.toml
+      continue-on-error: true
+      env:
+        CARGO_REGISTRY_TOKEN: ${{ secrets.CARGO_REGISTRY_TOKEN }}
     - name: Publish storage
       run: cargo publish --manifest-path storage/Cargo.toml
       continue-on-error: true

Cargo.lock

@@ -16,6 +16,7 @@ workspace = true
 [dependencies]
 cfg-if.workspace = true
 commonware-macros.workspace = true
+commonware-runtime.workspace = true
 crossbeam-queue.workspace = true
 futures-util.workspace = true
 loom = { workspace = true, features = ["futures"], optional = true }

actor/Cargo.toml

@@ -1,4 +1,8 @@
 use commonware_actor::{mailbox, Feedback};
+use commonware_runtime::{
+    telemetry::metrics::{Metric, Registered, Registration},
+    Metrics, Name, Supervisor,
+};
 use commonware_utils::NZUsize;
 use criterion::{criterion_group, BatchSize, Criterion, Throughput};
 use futures::pin_mut;
@@ -15,6 +19,44 @@ const PRODUCERS: usize = 4;
 const PRODUCER_MESSAGES: usize = 16 * 1024;
 const REPLACE_CAPACITY: usize = 1024;
 
+#[derive(Clone, Copy, Debug, Default)]
+struct NoopMetrics;
+
+impl Supervisor for NoopMetrics {
+    fn name(&self) -> Name {
+        Name::default()
+    }
+
+    fn child(&self, _label: &'static str) -> Self {
+        Self
+    }
+
+    fn with_attribute(self, _key: &'static str, _value: impl std::fmt::Display) -> Self {
+        self
+    }
+}
+
+impl Metrics for NoopMetrics {
+    fn register<N: Into<String>, H: Into<String>, M: Metric>(
+        &self,
+        _name: N,
+        _help: H,
+        metric: M,
+    ) -> Registered<M> {
+        Registered::with_registration(metric, Registration::from(()))
+    }
+
+    fn encode(&self) -> String {
+        String::new()
+    }
+}
+
+fn new<T: mailbox::Policy>(
+    capacity: std::num::NonZeroUsize,
+) -> (mailbox::Sender<T>, mailbox::Receiver<T>) {
+    mailbox::new(NoopMetrics, capacity)
+}
+
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 enum Policy {
     Drop,
@@ -77,7 +119,7 @@ fn bench_enqueue_ready(c: &mut Criterion) {
 
     group.bench_function(format!("capacity={CAPACITY}"), |b| {
         b.iter_batched(
-            || mailbox::new::<Message>(NZUsize!(CAPACITY)),
+            || new::<Message>(NZUsize!(CAPACITY)),
             |(sender, _receiver)| {
                 for _ in 0..MESSAGES {
                     let result = sender.enqueue(black_box(Message::drop()));
@@ -99,7 +141,7 @@ fn bench_try_recv_ready(c: &mut Criterion) {
     group.bench_function(format!("capacity={CAPACITY}"), |b| {
         b.iter_batched(
             || {
-                let (sender, receiver) = mailbox::new::<Message>(NZUsize!(CAPACITY));
+                let (sender, receiver) = new::<Message>(NZUsize!(CAPACITY));
                 for _ in 0..MESSAGES {
                     assert_eq!(sender.enqueue(Message::drop()), Feedback::Ok);
                 }
@@ -124,7 +166,7 @@ fn bench_try_recv_overflow(c: &mut Criterion) {
     group.bench_function(format!("capacity={CAPACITY} overflow={MESSAGES}"), |b| {
         b.iter_batched(
             || {
-                let (sender, receiver) = mailbox::new::<Message>(NZUsize!(CAPACITY));
+                let (sender, receiver) = new::<Message>(NZUsize!(CAPACITY));
                 for _ in 0..CAPACITY {
                     assert_eq!(sender.enqueue(Message::drop()), Feedback::Ok);
                 }
@@ -151,7 +193,7 @@ fn bench_round_trip_ready(c: &mut Criterion) {
 
     group.bench_function(format!("capacity={CAPACITY}"), |b| {
         b.iter_batched(
-            || mailbox::new::<Message>(NZUsize!(CAPACITY)),
+            || new::<Message>(NZUsize!(CAPACITY)),
             |(sender, mut receiver)| {
                 for _ in 0..MESSAGES {
                     let result = sender.enqueue(black_box(Message::drop()));
@@ -173,7 +215,7 @@ fn bench_recv_waiting(c: &mut Criterion) {
 
     group.bench_function("capacity=1", |b| {
         b.iter_batched(
-            || mailbox::new::<Message>(NZUsize!(1)),
+            || new::<Message>(NZUsize!(1)),
             |(sender, mut receiver)| {
                 futures::executor::block_on(async {
                     for _ in 0..MESSAGES {
@@ -207,7 +249,7 @@ fn bench_overflow_drop(c: &mut Criterion) {
     group.bench_function("capacity=1", |b| {
         b.iter_batched(
             || {
-                let (sender, receiver) = mailbox::new::<Message>(NZUsize!(1));
+                let (sender, receiver) = new::<Message>(NZUsize!(1));
                 assert_eq!(sender.enqueue(Message::drop()), Feedback::Ok);
                 (sender, receiver)
             },
@@ -232,7 +274,7 @@ fn bench_overflow_spill(c: &mut Criterion) {
     group.bench_function("capacity=1", |b| {
         b.iter_batched(
             || {
-                let (sender, receiver) = mailbox::new::<Message>(NZUsize!(1));
+                let (sender, receiver) = new::<Message>(NZUsize!(1));
                 assert_eq!(sender.enqueue(Message::drop()), Feedback::Ok);
                 (sender, receiver)
             },
@@ -251,7 +293,7 @@ fn bench_overflow_spill(c: &mut Criterion) {
 }
 
 fn replace_queue(newest: bool) -> (mailbox::Sender<Message>, mailbox::Receiver<Message>) {
-    let (sender, receiver) = mailbox::new::<Message>(NZUsize!(REPLACE_CAPACITY));
+    let (sender, receiver) = new::<Message>(NZUsize!(REPLACE_CAPACITY));
 
     for _ in 0..REPLACE_CAPACITY {
         assert_eq!(sender.enqueue(Message::drop()), Feedback::Ok);
@@ -300,7 +342,7 @@ fn bench_concurrent_enqueue(c: &mut Criterion) {
 
     group.bench_function(format!("producers={PRODUCERS} capacity={total}"), |b| {
         b.iter(|| {
-            let (sender, _receiver) = mailbox::new::<Message>(NZUsize!(total));
+            let (sender, _receiver) = new::<Message>(NZUsize!(total));
 
             std::thread::scope(|scope| {
                 for _ in 0..PRODUCERS {

actor/src/benches/mailbox.rs

@@ -40,6 +40,10 @@
 //! however, are not globally ordered and may be observed in any interleaving.
 
 use crate::Feedback;
+use commonware_runtime::{
+    telemetry::metrics::{Counter, MetricsExt as _},
+    Metrics,
+};
 use std::{
     collections::VecDeque,
     fmt,
@@ -368,6 +372,7 @@ impl Drop for Mutation<'_> {
 struct State<T: Policy> {
     ready: Ready<T>,
     overflow: OverflowState<T>,
+    backoff: Counter,
     closed: AtomicBool,
     senders: AtomicUsize,
     waker: AtomicWaker,
@@ -435,6 +440,11 @@ impl<T: Policy> Sender<T> {
             self.state.closed.load(Ordering::Acquire)
         });
 
+        // Record any backoff.
+        if feedback == Feedback::Backoff {
+            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
@@ -530,10 +540,11 @@ impl<T: Policy> Drop for Receiver<T> {
 }
 
 /// Create a new bounded mailbox.
-pub fn new<T: Policy>(capacity: NonZeroUsize) -> (Sender<T>, Receiver<T>) {
+pub fn new<T: Policy>(metrics: impl Metrics, capacity: NonZeroUsize) -> (Sender<T>, Receiver<T>) {
     let state = Arc::new(State {
         ready: Ready::new(capacity.get()),
         overflow: OverflowState::new(),
+        backoff: metrics.counter("backoff", "number of enqueue calls that requested backoff"),
         closed: AtomicBool::new(false),
         senders: AtomicUsize::new(1),
         waker: AtomicWaker::new(),
@@ -546,10 +557,52 @@ pub fn new<T: Policy>(capacity: NonZeroUsize) -> (Sender<T>, Receiver<T>) {
     )
 }
 
+#[cfg(test)]
+mod mocks {
+    use commonware_runtime::{
+        telemetry::metrics::{Metric, Registered, Registration},
+        Metrics as RuntimeMetrics, Name, Supervisor,
+    };
+    use std::fmt;
+
+    #[derive(Clone, Copy, Debug, Default)]
+    pub(super) struct Metrics;
+
+    impl Supervisor for Metrics {
+        fn name(&self) -> Name {
+            Name::default()
+        }
+
+        fn child(&self, _label: &'static str) -> Self {
+            Self
+        }
+
+        fn with_attribute(self, _key: &'static str, _value: impl fmt::Display) -> Self {
+            self
+        }
+    }
+
+    impl RuntimeMetrics for Metrics {
+        fn register<N: Into<String>, H: Into<String>, M: Metric>(
+            &self,
+            _name: N,
+            _help: H,
+            metric: M,
+        ) -> Registered<M> {
+            Registered::with_registration(metric, Registration::from(()))
+        }
+
+        fn encode(&self) -> String {
+            String::new()
+        }
+    }
+}
+
 #[cfg(all(test, not(feature = "loom")))]
 mod tests {
-    use super::*;
+    use super::{mocks, *};
     use commonware_macros::test_async;
+    use commonware_runtime::{deterministic, Runner as _, Supervisor};
     use commonware_utils::{channel::oneshot, NZUsize};
     use futures::{
         pin_mut,
@@ -562,6 +615,10 @@ mod tests {
         Arc,
     };
 
+    fn new<T: Policy>(capacity: NonZeroUsize) -> (Sender<T>, Receiver<T>) {
+        super::new(mocks::Metrics, capacity)
+    }
+
     #[derive(Debug, PartialEq, Eq)]
     enum Message {
         Update(u64),
@@ -706,6 +763,23 @@ mod tests {
         assert_eq!(receiver.try_recv(), Ok(Message::Buffered(2)));
     }
 
+    #[test]
+    fn backoff_metric_counts_backoff_feedback() {
+        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);
+            assert_eq!(sender.enqueue(Message::Buffered(2)), Feedback::Backoff);
+            assert_eq!(sender.enqueue(Message::Buffered(3)), Feedback::Backoff);
+
+            let buffer = context.encode();
+            assert!(
+                buffer.contains("mailbox_backoff_total 2"),
+                "missing backoff count in metrics: {buffer}"
+            );
+        });
+    }
+
     #[test]
     fn try_recv_drains_buffered_messages_after_senders_drop() {
         let (sender, mut receiver) = new(NZUsize!(1));
@@ -990,7 +1064,7 @@ mod tests {
 
 #[cfg(all(test, feature = "loom"))]
 mod loom_tests {
-    use super::*;
+    use super::{mocks, *};
     use commonware_utils::NZUsize;
     use futures::pin_mut;
     use loom::{
@@ -1005,6 +1079,10 @@ mod loom_tests {
         task::{RawWaker, RawWakerVTable, Waker},
     };
 
+    fn new<T: Policy>(capacity: NonZeroUsize) -> (Sender<T>, Receiver<T>) {
+        super::new(mocks::Metrics, capacity)
+    }
+
     #[derive(Clone, Copy, Debug, PartialEq, Eq)]
     enum Message {
         Drop(u8),

actor/src/mailbox.rs

@@ -117,7 +117,8 @@ where
     /// Creates a new engine with the given context and configuration.
     /// Returns the engine and a mailbox for sending messages to the engine.
     pub fn new(context: E, cfg: Config<P, M::Cfg, D>) -> (Self, Mailbox<P, M>) {
-        let (mailbox_sender, mailbox_receiver) = mailbox::new(cfg.mailbox_size);
+        let (mailbox_sender, mailbox_receiver) =
+            mailbox::new(context.child("mailbox"), cfg.mailbox_size);
         let mailbox = Mailbox::<P, M>::new(mailbox_sender);
 
         let metrics = metrics::Metrics::init(&context);

broadcast/src/buffered/engine.rs

@@ -12,7 +12,7 @@ use commonware_actor::{
 use commonware_codec::{Decode, DecodeExt, Encode};
 use commonware_cryptography::{certificate::Scheme, Digest};
 use commonware_parallel::Sequential;
-use commonware_runtime::{spawn_cell, ContextCell, Handle, Spawner};
+use commonware_runtime::{spawn_cell, ContextCell, Handle, Metrics, Spawner};
 use commonware_utils::{channel::oneshot, NZUsize};
 use rand_core::CryptoRngCore;
 use std::collections::{btree_map::Entry, BTreeMap, HashSet, VecDeque};
@@ -63,12 +63,12 @@ pub struct Reporter<R: CryptoRngCore, S: Scheme, D: Digest> {
 
 impl<R, S, D> Reporter<R, S, D>
 where
-    R: CryptoRngCore,
+    R: CryptoRngCore + Metrics,
     S: scheme::Scheme<D>,
     D: Digest,
 {
     pub fn new(rng: R, scheme: S) -> (Self, Mailbox<S, D>) {
-        let (sender, receiver) = mailbox::new(NZUsize!(1024));
+        let (sender, receiver) = mailbox::new(rng.child("mailbox"), NZUsize!(1024));
         (
             Self {
                 context: ContextCell::new(rng),

consensus/src/aggregation/mocks/reporter.rs

@@ -369,7 +369,7 @@ where
     /// Create a new [`Engine`] with the given configuration.
     pub fn new(context: E, config: Config<P, S, X, D, C, H, B, T>) -> (Self, Mailbox<B, C, H, P>) {
         let metrics = ShardMetrics::new(&context);
-        let (sender, mailbox) = mailbox::new(config.mailbox_size);
+        let (sender, mailbox) = mailbox::new(context.child("mailbox"), config.mailbox_size);
         (
             Self {
                 context: ContextCell::new(context),

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

@@ -335,7 +335,7 @@ where
         let _ = processed_height.try_set(last_processed_height.get());
 
         // Initialize mailbox
-        let (sender, mailbox) = mailbox::new(config.mailbox_size);
+        let (sender, mailbox) = mailbox::new(context.child("mailbox"), config.mailbox_size);
         (
             Self {
                 context: ContextCell::new(context),

consensus/src/marshal/core/actor.rs

@@ -61,10 +61,10 @@ where
     R: Receiver<PublicKey = P>,
     P: PublicKey,
 {
-    let (sender, receiver) = mailbox::new(config.mailbox_size);
+    let (sender, receiver) = mailbox::new(context.child("handler"), config.mailbox_size);
     let handler = handler::Handler::new(sender);
     let (resolver_engine, resolver) = p2p::Engine::new(
-        context,
+        context.child("resolver"),
         p2p::Config {
             peer_provider: config.peer_provider,
             blocker: config.blocker,