From 94a52a41c2fb5695d5e184d219fc75e9881b8e73 Mon Sep 17 00:00:00 2001 From: Joel DSouza Date: Fri, 3 Jul 2026 20:03:20 +0200 Subject: [PATCH 1/9] feat(core): bombay-core crate + zero-box mailbox foundation (#112) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Start the part-by-part core rebuild (epic #122) in a fresh, god-level-bar crate rather than mutating the vendored kameo modules. The mailbox is the first spine part (#112), built standalone so it can be hard-tested before the run-loop is grafted onto it (#114/#116/#117). - `Mailboxed { type Msg }`: the scaffold seam the rebuilt `Actor` trait will later subsume. - `Signal` = `Message(A::Msg) | Stop`: a concrete, closed envelope with no `Box` — a `tell` moves the message into the queue slot (zero per-message alloc). - `Capacity(NonZeroUsize)`: makes both illegal capacities unrepresentable (0, and > tokio's MAX_PERMITS which panics), so `bounded` is infallible — validated at our own boundary rather than relying on tokio's panic. - bounded `send`/`try_send`/`recv`; `SendError`/`TrySendError::{Full,Closed}` carry the undelivered signal back to the caller. TDD: 5 tests (round-trip, backpressure hand-back, capacity boundaries, Stop-after-message FIFO). Clippy `--all-features` clean. --- Cargo.lock | 8 ++ Cargo.toml | 2 +- bombay-core/Cargo.toml | 19 +++ bombay-core/src/lib.rs | 10 ++ bombay-core/src/mailbox.rs | 245 +++++++++++++++++++++++++++++++++++++ 5 files changed, 283 insertions(+), 1 deletion(-) create mode 100644 bombay-core/Cargo.toml create mode 100644 bombay-core/src/lib.rs create mode 100644 bombay-core/src/mailbox.rs diff --git a/Cargo.lock b/Cargo.lock index 1da11c0..8879464 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -470,6 +470,14 @@ dependencies = [ "tracing-subscriber", ] +[[package]] +name = "bombay-core" +version = "0.1.0" +dependencies = [ + "proptest", + "tokio", +] + [[package]] name = "bombay_actors" version = "0.5.0" diff --git a/Cargo.toml b/Cargo.toml index 6881def..45fb962 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [workspace] resolver = "2" -members = [".", "actors", "console", "macros"] +members = [".", "actors", "bombay-core", "console", "macros"] [workspace.package] edition = "2024" diff --git a/bombay-core/Cargo.toml b/bombay-core/Cargo.toml new file mode 100644 index 0000000..8c1b8ed --- /dev/null +++ b/bombay-core/Cargo.toml @@ -0,0 +1,19 @@ +[package] +name = "bombay-core" +version = "0.1.0" +edition.workspace = true +license.workspace = true +authors.workspace = true +repository.workspace = true +description = "Bombay runtime core — the rebuilt local actor spine (mailbox, message, reply, actor loop). Transport- and domain-agnostic." +publish = false + +[dependencies] +tokio = { workspace = true, features = ["sync", "rt", "macros", "time"] } + +[dev-dependencies] +tokio = { workspace = true, features = ["rt", "rt-multi-thread", "macros", "sync", "time"] } +proptest = { workspace = true } + +[lints] +workspace = true diff --git a/bombay-core/src/lib.rs b/bombay-core/src/lib.rs new file mode 100644 index 0000000..b10f883 --- /dev/null +++ b/bombay-core/src/lib.rs @@ -0,0 +1,10 @@ +//! Bombay runtime core — the rebuilt local actor spine. +//! +//! Built card-by-card (M1 epic #122) with kameo as a reference oracle, held to +//! the god-level clippy bar from line one. Transport- and domain-agnostic: the +//! Zenoh remote tier and the nexus aggregate-runner sit on top of this. +//! +//! Nothing here is public API yet — the spine is assembled part-by-part and the +//! surface is settled once the whole core lands (#112–#121). + +pub mod mailbox; diff --git a/bombay-core/src/mailbox.rs b/bombay-core/src/mailbox.rs new file mode 100644 index 0000000..39caf9a --- /dev/null +++ b/bombay-core/src/mailbox.rs @@ -0,0 +1,245 @@ +//! The actor's in-memory mailbox: a bounded MPSC queue of [`Signal`]s. +//! +//! The local tier of the two-tier message model (#66): typed, in-memory, +//! **zero-serialize**. `tell` moves an `A::Msg` into a queue slot — no +//! per-message heap box. Built on `tokio::sync::mpsc`, bounded for backpressure. + +use std::{fmt, num::NonZeroUsize}; + +use tokio::sync::mpsc; + +/// A validated mailbox capacity: at least `1`, at most [`Capacity::MAX`]. +/// +/// Makes both illegal capacities unrepresentable, so [`bounded`] cannot fail: +/// zero is excluded by `NonZeroUsize`, and the upper bound is checked here +/// rather than trusting `tokio::sync::mpsc::channel` (which panics outside +/// `1..=MAX`). Validating at our own boundary is deliberate — we do not rely on +/// an upstream crate's panic as our error path (rule 4). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Capacity(NonZeroUsize); + +impl Capacity { + /// The largest capacity `tokio::sync::mpsc::channel` accepts — the tokio + /// `Semaphore` permit ceiling (`usize::MAX >> 3`). Mirrored here because + /// tokio does not expose it; the `capacity_at_the_upper_boundary_is_usable` + /// test guards this constant against tokio lowering the limit. + pub const MAX: usize = usize::MAX >> 3; + + /// Builds a `Capacity`, returning `None` if `value` exceeds [`Capacity::MAX`]. + #[must_use] + pub const fn new(value: NonZeroUsize) -> Option { + if value.get() > Self::MAX { + None + } else { + Some(Self(value)) + } + } + + /// The capacity as a `usize`, always in `1..=MAX`. + #[must_use] + pub const fn get(self) -> usize { + self.0.get() + } +} + +/// The seam between a mailbox and its actor. +/// +/// A mailbox is monomorphized per actor `A`, carrying that actor's single closed +/// message type `A::Msg` by value — no `Box`. This scaffold trait is what +/// the rebuilt `Actor` trait (#114/#116) will later subsume; keeping it separate +/// lets the mailbox be built and hard-tested on its own (#112). +/// +/// `Msg` is `Send + 'static` for now; the cfg-gated `MaybeSend` relaxation for +/// single-threaded client builds arrives with #9. +pub trait Mailboxed { + /// The actor's single closed message type, stored in the queue by value. + type Msg: Send + 'static; +} + +/// A signal in an actor's mailbox: a domain message or a system control signal. +/// +/// A **concrete, closed** envelope — no `Box` at either layer. `tell` moves +/// an `A::Msg` into a [`Signal::Message`] slot, so a send is zero-allocation. +#[expect( + clippy::exhaustive_enums, + reason = "the signal set is deliberately closed so the run-loop is a total match; \ + new arms (Stop, LinkDied, …) are added under their driving cards" +)] +pub enum Signal { + /// A domain message for the actor to handle. + Message(A::Msg), + /// Asks the actor to stop after draining messages queued before it. + Stop, +} + +/// Sends [`Signal`]s to an actor's mailbox. Cloneable; the channel stays open +/// while any sender is alive. +pub struct MailboxSender { + tx: mpsc::Sender>, +} + +/// The single consumer of an actor's mailbox. The run-loop pulls from it. +pub struct MailboxReceiver { + rx: mpsc::Receiver>, +} + +/// The receiver was dropped, so the signal could not be delivered. +/// +/// Carries the undelivered [`Signal`] back to the caller (rule 3: never silently +/// drop the payload). +pub struct SendError(pub Signal); + +impl fmt::Debug for SendError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str("SendError(receiver dropped)") + } +} + +/// Why a non-blocking [`MailboxSender::try_send`] could not enqueue a signal. +/// +/// Both variants carry the undelivered [`Signal`] back to the caller. `Full` is +/// retryable (drain, then retry); `Closed` is terminal (the actor is gone). +#[expect( + clippy::exhaustive_enums, + reason = "closed set — a try_send fails for exactly these two reasons" +)] +pub enum TrySendError { + /// The mailbox is at capacity; back off and retry. + Full(Signal), + /// The receiver has been dropped; the actor is no longer running. + Closed(Signal), +} + +impl fmt::Debug for TrySendError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Full(_) => f.write_str("TrySendError::Full(mailbox at capacity)"), + Self::Closed(_) => f.write_str("TrySendError::Closed(receiver dropped)"), + } + } +} + +/// Creates a bounded mailbox with room for `capacity` queued signals. +/// +/// Bounded is the only mode: a full mailbox exerts backpressure rather than +/// growing without limit (an unbounded queue is a memory footgun). Infallible +/// by construction — [`Capacity`] has already excluded the values that would +/// make the underlying channel panic. +#[must_use] +pub fn bounded(capacity: Capacity) -> (MailboxSender, MailboxReceiver) { + let (tx, rx) = mpsc::channel(capacity.get()); + (MailboxSender { tx }, MailboxReceiver { rx }) +} + +impl MailboxSender { + /// Sends `signal`, waiting for capacity if the mailbox is full. + /// + /// # Errors + /// + /// Returns [`SendError`] (carrying `signal` back) if the receiver has been + /// dropped, i.e. the actor is no longer running. + pub async fn send(&self, signal: Signal) -> Result<(), SendError> { + self.tx.send(signal).await.map_err(|err| SendError(err.0)) + } + + /// Tries to enqueue `signal` without waiting. + /// + /// # Errors + /// + /// Returns [`TrySendError::Full`] if the mailbox is at capacity, or + /// [`TrySendError::Closed`] if the receiver has been dropped. Both carry + /// `signal` back to the caller. + pub fn try_send(&self, signal: Signal) -> Result<(), TrySendError> { + self.tx.try_send(signal).map_err(|err| match err { + mpsc::error::TrySendError::Full(undelivered) => TrySendError::Full(undelivered), + mpsc::error::TrySendError::Closed(undelivered) => TrySendError::Closed(undelivered), + }) + } +} + +impl MailboxReceiver { + /// Receives the next signal, waiting until one is available. + /// + /// Returns `None` once the mailbox is closed and drained. + pub async fn recv(&mut self) -> Option> { + self.rx.recv().await + } +} + +#[cfg(test)] +mod tests { + use super::*; + + use std::num::NonZeroUsize; + + /// Scaffold actor for the mailbox tests. `Mailboxed` is the seam the + /// not-yet-rebuilt `Actor` trait (#114/#116) will later subsume. + struct Probe; + impl Mailboxed for Probe { + type Msg = u64; + } + + /// Builds a valid [`Capacity`] for tests; panics on out-of-range input, + /// which in a test is a programmer error in the test itself. + fn cap(n: usize) -> Capacity { + Capacity::new(NonZeroUsize::new(n).expect("test capacity must be nonzero")) + .expect("test capacity must be within Capacity::MAX") + } + + #[tokio::test] + async fn sent_message_is_received() { + let (tx, mut rx) = bounded::(cap(4)); + + tx.send(Signal::Message(42)).await.expect("send should succeed"); + + assert!(matches!(rx.recv().await, Some(Signal::Message(42)))); + } + + #[tokio::test] + async fn capacity_at_the_upper_boundary_is_usable() { + // A mailbox built at tokio's true maximum must not panic and must work. + // Guards Capacity::MAX against tokio ever lowering its semaphore limit. + let (tx, mut rx) = bounded::(cap(Capacity::MAX)); + + tx.try_send(Signal::Message(7)).expect("send into max-capacity mailbox"); + + assert!(matches!(rx.recv().await, Some(Signal::Message(7)))); + } + + #[test] + fn capacity_rejects_values_above_tokio_max() { + assert!(Capacity::new(NonZeroUsize::MIN).is_some()); + assert!(Capacity::new(NonZeroUsize::new(Capacity::MAX).expect("nonzero")).is_some()); + + let too_big = NonZeroUsize::new(Capacity::MAX.checked_add(1).expect("no overflow")).expect("nonzero"); + assert!(Capacity::new(too_big).is_none()); + // Capacity zero is unrepresentable: NonZeroUsize cannot hold it. + } + + #[tokio::test] + async fn stop_signal_is_delivered_in_order_after_a_message() { + let (tx, mut rx) = bounded::(cap(4)); + + tx.send(Signal::Message(1)).await.expect("message"); + tx.send(Signal::Stop).await.expect("stop"); + + // FIFO: the domain message precedes the control signal that followed it. + assert!(matches!(rx.recv().await, Some(Signal::Message(1)))); + assert!(matches!(rx.recv().await, Some(Signal::Stop))); + } + + #[tokio::test] + async fn full_mailbox_rejects_try_send_and_returns_the_message() { + let (tx, mut rx) = bounded::(cap(1)); + + tx.try_send(Signal::Message(1)).expect("first signal fits"); + + // Mailbox is now full: try_send must reject and hand the message back. + let rejected = tx.try_send(Signal::Message(2)); + assert!(matches!(rejected, Err(TrySendError::Full(Signal::Message(2))))); + + // Draining one slot frees capacity for the next try_send. + assert!(matches!(rx.recv().await, Some(Signal::Message(1)))); + tx.try_send(Signal::Message(3)).expect("fits after drain"); + } +} From 155a1e3bb1387046513d9a7ca2a40bed5b0d6978 Mon Sep 17 00:00:00 2001 From: Joel DSouza Date: Fri, 3 Jul 2026 20:19:11 +0200 Subject: [PATCH 2/9] feat(core): mailbox lifecycle, weak death-watch, LinkDied arm (#112) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Continue the standalone mailbox (#112) through the behaviour-test tier. Production additions: - `close()`: graceful stop — reject new sends, still drain what's queued. - `Signal::LinkDied(Box)`: the first control arm, deliberately boxed. A `size_of` test pins `Signal` at 16 bytes, proving the cold fat variant does not inflate the hot message slot (large-variant discipline). - `WeakMailboxSender` + `downgrade`/`upgrade` and `Clone` on the sender: the non-pinning liveness handle death-watch is built on — `upgrade()` is `None` once the last strong sender drops. (This keeps kameo's WeakSender approach, the reason the generational-arena was dropped in #122.) - scaffold `ActorId`/`StopReason`/`LinkDied` placeholders (owned by #121/#113/ #120), present only to give the LinkDied arm a concrete shape. Tests (TDD): +10 → 15 total. Lifecycle (close-drains, send-after-drop, recv-none-after-drain), LinkDied round-trip + slot-size guard, weak-sender tracks-last-strong + upgraded-can-send, a real multi-thread linearizability test (8 senders × 64 msgs, Barrier-synced, FIFO-per-sender + no loss/dup), and two proptests (capacity boundaries incl. MAX±1/usize::MAX; single-sender FIFO round-trip over arbitrary sequences × capacities). Clippy --all-features clean. --- bombay-core/src/mailbox.rs | 320 ++++++++++++++++++++++++++++++++++++- 1 file changed, 319 insertions(+), 1 deletion(-) diff --git a/bombay-core/src/mailbox.rs b/bombay-core/src/mailbox.rs index 39caf9a..64b04a0 100644 --- a/bombay-core/src/mailbox.rs +++ b/bombay-core/src/mailbox.rs @@ -56,6 +56,52 @@ pub trait Mailboxed { type Msg: Send + 'static; } +/// Scaffold actor identity. #121 replaces this with the identity-first AID / +/// key-expr `ActorId`; it exists here only so the mailbox's [`LinkDied`] arm has +/// a concrete shape to carry. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ActorId(u64); + +impl ActorId { + /// Wraps a raw identifier. + #[must_use] + pub const fn new(raw: u64) -> Self { + Self(raw) + } +} + +/// Scaffold reason a linked actor stopped. #113/#120 own the real +/// error/supervision model; the `String` in `Panicked` is deliberately the fat +/// field that makes [`LinkDied`] worth boxing. +#[derive(Debug, Clone)] +#[expect( + clippy::exhaustive_enums, + reason = "scaffold placeholder for the #113/#120 stop-reason model" +)] +pub enum StopReason { + /// The actor returned from its run-loop normally. + Normal, + /// The actor's handler panicked, with the panic message. + Panicked(String), +} + +/// The payload of a [`Signal::LinkDied`]: a linked actor has terminated. +/// +/// Boxed inside [`Signal`] because it is a **cold** control path — boxing it +/// keeps the hot [`Signal::Message`] slot small (see the large-variant +/// discipline). +#[derive(Debug, Clone)] +#[expect( + clippy::exhaustive_structs, + reason = "scaffold placeholder for the #120 links/supervision payload" +)] +pub struct LinkDied { + /// The dead actor's identity. + pub id: ActorId, + /// Why it stopped. + pub reason: StopReason, +} + /// A signal in an actor's mailbox: a domain message or a system control signal. /// /// A **concrete, closed** envelope — no `Box` at either layer. `tell` moves @@ -70,6 +116,9 @@ pub enum Signal { Message(A::Msg), /// Asks the actor to stop after draining messages queued before it. Stop, + /// A linked actor has terminated. Boxed: this is a cold path, and inlining + /// its fields would inflate every message slot (large-variant discipline). + LinkDied(Box), } /// Sends [`Signal`]s to an actor's mailbox. Cloneable; the channel stays open @@ -78,6 +127,37 @@ pub struct MailboxSender { tx: mpsc::Sender>, } +impl Clone for MailboxSender { + fn clone(&self) -> Self { + Self { tx: self.tx.clone() } + } +} + +/// A non-pinning handle to a mailbox: holding one does **not** keep it alive. +/// +/// [`upgrade`](Self::upgrade) yields a strong sender only while a real +/// [`MailboxSender`] still exists — the primitive death-watch is built on. +pub struct WeakMailboxSender { + weak: mpsc::WeakSender>, +} + +impl WeakMailboxSender { + /// Upgrades to a strong [`MailboxSender`], or `None` if every strong sender + /// has been dropped (the actor is gone). + #[must_use] + pub fn upgrade(&self) -> Option> { + self.weak.upgrade().map(|tx| MailboxSender { tx }) + } +} + +impl Clone for WeakMailboxSender { + fn clone(&self) -> Self { + Self { + weak: self.weak.clone(), + } + } +} + /// The single consumer of an actor's mailbox. The run-loop pulls from it. pub struct MailboxReceiver { rx: mpsc::Receiver>, @@ -155,6 +235,14 @@ impl MailboxSender { mpsc::error::TrySendError::Closed(undelivered) => TrySendError::Closed(undelivered), }) } + + /// Downgrades to a [`WeakMailboxSender`] that does not keep the mailbox open. + #[must_use] + pub fn downgrade(&self) -> WeakMailboxSender { + WeakMailboxSender { + weak: self.tx.downgrade(), + } + } } impl MailboxReceiver { @@ -164,13 +252,25 @@ impl MailboxReceiver { pub async fn recv(&mut self) -> Option> { self.rx.recv().await } + + /// Closes the mailbox: senders can no longer enqueue, but signals already + /// queued still drain through [`recv`](Self::recv) before it yields `None`. + /// + /// Used for a graceful stop — the run-loop finishes in-flight work rather + /// than dropping it. + pub fn close(&mut self) { + self.rx.close(); + } } #[cfg(test)] mod tests { use super::*; - use std::num::NonZeroUsize; + use std::{num::NonZeroUsize, sync::Arc}; + + use proptest::prelude::*; + use tokio::{runtime::Builder, sync::Barrier}; /// Scaffold actor for the mailbox tests. `Mailboxed` is the seam the /// not-yet-rebuilt `Actor` trait (#114/#116) will later subsume. @@ -179,6 +279,58 @@ mod tests { type Msg = u64; } + /// A message tagged with `(sender_id, seq)`; also proves `Msg` is any + /// concrete type, not just a primitive. + struct Tagged; + impl Mailboxed for Tagged { + type Msg = (u32, u32); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn n_senders_one_receiver_preserve_per_sender_order() { + const SENDERS: u32 = 8; + const PER_SENDER: u32 = 64; + + // Small capacity so senders genuinely contend and backpressure. + let (tx, mut rx) = bounded::(cap(4)); + let start = Arc::new(Barrier::new(SENDERS as usize)); + + let mut handles = Vec::with_capacity(SENDERS as usize); + for sender_id in 0..SENDERS { + let tx = tx.clone(); + let start = Arc::clone(&start); + handles.push(tokio::spawn(async move { + start.wait().await; // all senders race from the same instant + for seq in 0..PER_SENDER { + tx.send(Signal::Message((sender_id, seq))) + .await + .expect("send"); + } + })); + } + drop(tx); // recv ends only once every sender has dropped its clone + + let mut next_expected = vec![0u32; SENDERS as usize]; + let mut total = 0u32; + while let Some(signal) = rx.recv().await { + let Signal::Message((sender_id, seq)) = signal else { + panic!("unexpected non-message signal"); + }; + let slot = &mut next_expected[sender_id as usize]; + assert_eq!(seq, *slot, "FIFO-per-sender violated for sender {sender_id}"); + *slot += 1; + total += 1; + } + + assert_eq!(total, SENDERS * PER_SENDER, "lost or duplicated messages"); + for (sender_id, &count) in next_expected.iter().enumerate() { + assert_eq!(count, PER_SENDER, "sender {sender_id} did not fully arrive"); + } + for handle in handles { + handle.await.expect("sender task panicked"); + } + } + /// Builds a valid [`Capacity`] for tests; panics on out-of-range input, /// which in a test is a programmer error in the test itself. fn cap(n: usize) -> Capacity { @@ -216,6 +368,70 @@ mod tests { // Capacity zero is unrepresentable: NonZeroUsize cannot hold it. } + #[test] + fn link_died_variant_is_boxed_so_message_slots_stay_small() { + use std::mem::size_of; + + // The cold LinkDied variant is boxed, so a small-message actor's queue + // slot is bounded by the hot Message(u64) path — not inflated by + // LinkDied's fields (id + a String-bearing reason). Guards the + // "every slot = largest variant" trap; clippy::large_enum_variant is the + // compile-time backstop via the workspace bar. + assert!( + size_of::>() <= 2 * size_of::(), + "Signal slot is {} bytes; LinkDied is not boxed", + size_of::>() + ); + } + + #[tokio::test] + async fn link_died_signal_round_trips() { + let (tx, mut rx) = bounded::(cap(2)); + + tx.send(Signal::LinkDied(Box::new(LinkDied { + id: ActorId::new(7), + reason: StopReason::Normal, + }))) + .await + .expect("send link-died"); + + let Some(Signal::LinkDied(link_died)) = rx.recv().await else { + panic!("expected a LinkDied signal"); + }; + assert_eq!(link_died.id, ActorId::new(7)); + assert!(matches!(link_died.reason, StopReason::Normal)); + } + + #[tokio::test] + async fn weak_sender_tracks_the_last_strong_sender() { + let (tx, _rx) = bounded::(cap(2)); + let tx2 = tx.clone(); + let weak = tx.downgrade(); + + drop(tx); + assert!( + weak.upgrade().is_some(), + "one strong sender remains -> still alive" + ); + + drop(tx2); + assert!( + weak.upgrade().is_none(), + "all strong senders gone -> non-pinning weak handle reports dead" + ); + } + + #[tokio::test] + async fn upgraded_weak_sender_can_send() { + let (tx, mut rx) = bounded::(cap(2)); + let weak = tx.downgrade(); + + let strong = weak.upgrade().expect("channel still alive"); + strong.send(Signal::Message(5)).await.expect("send via upgraded"); + + assert!(matches!(rx.recv().await, Some(Signal::Message(5)))); + } + #[tokio::test] async fn stop_signal_is_delivered_in_order_after_a_message() { let (tx, mut rx) = bounded::(cap(4)); @@ -228,6 +444,49 @@ mod tests { assert!(matches!(rx.recv().await, Some(Signal::Stop))); } + #[tokio::test] + async fn closing_the_receiver_stops_new_sends_but_drains_queued() { + let (tx, mut rx) = bounded::(cap(4)); + tx.send(Signal::Message(1)).await.expect("queued before close"); + + rx.close(); + + // New sends are rejected (the message comes back)... + assert!(matches!( + tx.send(Signal::Message(2)).await, + Err(SendError(Signal::Message(2))) + )); + // ...but messages queued before the close still drain, then None. + assert!(matches!(rx.recv().await, Some(Signal::Message(1)))); + assert!(rx.recv().await.is_none()); + } + + #[tokio::test] + async fn send_after_receiver_dropped_returns_the_message() { + let (tx, rx) = bounded::(cap(4)); + drop(rx); + + assert!(matches!( + tx.send(Signal::Message(9)).await, + Err(SendError(Signal::Message(9))) + )); + assert!(matches!( + tx.try_send(Signal::Message(9)), + Err(TrySendError::Closed(Signal::Message(9))) + )); + } + + #[tokio::test] + async fn recv_returns_none_after_all_senders_dropped_and_drained() { + let (tx, mut rx) = bounded::(cap(4)); + tx.send(Signal::Message(1)).await.expect("queued"); + drop(tx); + + // Queued message drains first, then the closed-and-empty channel ends. + assert!(matches!(rx.recv().await, Some(Signal::Message(1)))); + assert!(rx.recv().await.is_none()); + } + #[tokio::test] async fn full_mailbox_rejects_try_send_and_returns_the_message() { let (tx, mut rx) = bounded::(cap(1)); @@ -242,4 +501,63 @@ mod tests { assert!(matches!(rx.recv().await, Some(Signal::Message(1)))); tx.try_send(Signal::Message(3)).expect("fits after drain"); } + + proptest! { + /// `Capacity::new` accepts a value iff it is within `MAX`, and preserves + /// it. The strategy pins the interesting boundaries: `1`, `MAX-1`, `MAX`, + /// `MAX+1`, `usize::MAX`. + #[test] + fn prop_capacity_accepts_iff_within_max( + n in prop_oneof![ + 1usize..=4096, + Just(Capacity::MAX - 1), + Just(Capacity::MAX), + Just(Capacity::MAX + 1), + Just(usize::MAX), + ], + ) { + let value = NonZeroUsize::new(n).expect("strategy yields n >= 1"); + let capacity = Capacity::new(value); + + prop_assert_eq!(capacity.is_some(), n <= Capacity::MAX); + if let Some(capacity) = capacity { + prop_assert_eq!(capacity.get(), n); + } + } + + /// A single sender's messages come out in the exact order they went in, + /// for any message sequence and any capacity — the queue neither drops, + /// duplicates, nor reorders (FIFO). + #[test] + fn prop_fifo_roundtrip_single_sender( + messages in prop::collection::vec(any::(), 0..200), + capacity in 1usize..=64, + ) { + let sent = messages.clone(); + let received = Builder::new_current_thread() + .build() + .expect("current-thread runtime") + .block_on(async move { + let (tx, mut rx) = bounded::(cap(capacity)); + let expected = messages.len(); + let producer = tokio::spawn(async move { + for message in messages { + tx.send(Signal::Message(message)).await.expect("send"); + } + }); + + let mut got = Vec::with_capacity(expected); + while got.len() < expected { + let Some(Signal::Message(message)) = rx.recv().await else { + break; + }; + got.push(message); + } + producer.await.expect("producer task"); + got + }); + + prop_assert_eq!(received, sent); + } + } } From b23df7ea74c7527b9096cd1063bd9f4be464ad61 Mon Sep 17 00:00:00 2001 From: Joel DSouza Date: Fri, 3 Jul 2026 20:36:09 +0200 Subject: [PATCH 3/9] feat(core): mailbox criterion bench + reproducible mutation gate (#112) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Measure, don't assume, and wire the tooling into the pinned flake so every check is reproducible (no floating `nix run nixpkgs#…`). - criterion bench (`benches/mailbox.rs`): zero-box `tell` enqueue ≈ 5.6 ns; send+recv round-trip ≈ 21.4 ns. Baseline for the #114/#118 two-tier wiring. - cargo-mutants: added to the devShell (pinned) and exposed as an on-demand `packages.mutants` (`nix build .#mutants`, mirrors the coverage package) that fails if any mutant survives. Result on mailbox.rs: 0 missed — 6 caught, 30 unviable, 4 detected-via-timeout. - Killed the 2 initial survivors (the hand-written SendError/TrySendError Debug impls) with an explicit Debug-format assertion. Tests: +1 → 16. Clippy --all-features clean; flake nixfmt-clean. --- Cargo.lock | 1 + bombay-core/Cargo.toml | 5 +++ bombay-core/benches/mailbox.rs | 73 ++++++++++++++++++++++++++++++++++ bombay-core/src/mailbox.rs | 14 +++++++ flake.nix | 21 ++++++++++ 5 files changed, 114 insertions(+) create mode 100644 bombay-core/benches/mailbox.rs diff --git a/Cargo.lock b/Cargo.lock index 8879464..c8ed3f0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -474,6 +474,7 @@ dependencies = [ name = "bombay-core" version = "0.1.0" dependencies = [ + "criterion", "proptest", "tokio", ] diff --git a/bombay-core/Cargo.toml b/bombay-core/Cargo.toml index 8c1b8ed..8d5c3ba 100644 --- a/bombay-core/Cargo.toml +++ b/bombay-core/Cargo.toml @@ -14,6 +14,11 @@ tokio = { workspace = true, features = ["sync", "rt", "macros", "time"] } [dev-dependencies] tokio = { workspace = true, features = ["rt", "rt-multi-thread", "macros", "sync", "time"] } proptest = { workspace = true } +criterion = { version = "0.8", features = ["async_tokio"] } + +[[bench]] +name = "mailbox" +harness = false [lints] workspace = true diff --git a/bombay-core/benches/mailbox.rs b/bombay-core/benches/mailbox.rs new file mode 100644 index 0000000..0eb5ad4 --- /dev/null +++ b/bombay-core/benches/mailbox.rs @@ -0,0 +1,73 @@ +//! Criterion benches for the mailbox hot path. +//! +//! The card's premise — a zero-box `tell` is cheaper than kameo's +//! `Box` enqueue — is un-templated: no framework ships this +//! shape, so we *measure* rather than assume (#112). These numbers are the +//! baseline the later two-tier wiring (#114/#118) is compared against. + +use std::num::NonZeroUsize; + +use bombay_core::mailbox::{Capacity, Mailboxed, Signal, bounded}; +use criterion::{BatchSize, Criterion, black_box, criterion_group, criterion_main}; + +struct Bench; +impl Mailboxed for Bench { + type Msg = u64; +} + +fn cap(n: usize) -> Capacity { + Capacity::new(NonZeroUsize::new(n).expect("nonzero")).expect("within max") +} + +/// Pure enqueue cost: how long to `try_send` 1_000 messages into a mailbox with +/// spare capacity. `iter_batched_ref` keeps the `bounded()` setup out of the +/// measured region, and the fresh mailbox per batch never fills, so this isolates +/// the move-into-slot cost of a `tell`. +fn enqueue(c: &mut Criterion) { + c.bench_function("tell_try_send_1k_u64", |b| { + b.iter_batched_ref( + || bounded::(cap(1024)), + |(tx, _rx)| { + for i in 0..1000u64 { + tx.try_send(Signal::Message(black_box(i))) + .expect("capacity available"); + } + }, + BatchSize::SmallInput, + ); + }); +} + +/// End-to-end throughput: 1_000 `send`s and 1_000 `recv`s across a producer task +/// and the consumer, on a current-thread runtime. +fn roundtrip(c: &mut Criterion) { + let rt = tokio::runtime::Builder::new_current_thread() + .build() + .expect("current-thread runtime"); + + c.bench_function("send_recv_roundtrip_1k_u64", |b| { + b.iter(|| { + rt.block_on(async { + let (tx, mut rx) = bounded::(cap(1024)); + let producer = tokio::spawn(async move { + for i in 0..1000u64 { + tx.send(Signal::Message(black_box(i))).await.expect("send"); + } + }); + + let mut received = 0u32; + while received < 1000 { + let Some(Signal::Message(_)) = rx.recv().await else { + break; + }; + received += 1; + } + producer.await.expect("producer"); + black_box(received) + }); + }); + }); +} + +criterion_group!(benches, enqueue, roundtrip); +criterion_main!(benches); diff --git a/bombay-core/src/mailbox.rs b/bombay-core/src/mailbox.rs index 64b04a0..01106c1 100644 --- a/bombay-core/src/mailbox.rs +++ b/bombay-core/src/mailbox.rs @@ -402,6 +402,20 @@ mod tests { assert!(matches!(link_died.reason, StopReason::Normal)); } + #[test] + fn error_debug_formats_are_stable() { + // The Debug impls are hand-written (so the error types don't inherit an + // `A::Msg: Debug` bound); pin their output so a regression is caught. + let send_err: SendError = SendError(Signal::Stop); + assert_eq!(format!("{send_err:?}"), "SendError(receiver dropped)"); + + let full: TrySendError = TrySendError::Full(Signal::Stop); + assert_eq!(format!("{full:?}"), "TrySendError::Full(mailbox at capacity)"); + + let closed: TrySendError = TrySendError::Closed(Signal::Stop); + assert_eq!(format!("{closed:?}"), "TrySendError::Closed(receiver dropped)"); + } + #[tokio::test] async fn weak_sender_tracks_the_last_strong_sender() { let (tx, _rx) = bounded::(cap(2)); diff --git a/flake.nix b/flake.nix index 7d2c5e1..0c3785b 100644 --- a/flake.nix +++ b/flake.nix @@ -224,6 +224,7 @@ cloc cargo-edit cargo-expand + cargo-mutants gh ]; }; @@ -251,6 +252,25 @@ # feature auto-enables via the self dev-dep, so the cucumber runners build. packages = let + # Mutation testing for the rebuilt core (card #112+). On-demand, NOT a + # gating check — like coverage, cargo-mutants rebuilds+tests once per + # mutant, far too slow for the per-push gate. Pinned via the flake's + # nixpkgs input (never `nix run nixpkgs#…`) so the run is reproducible. + # `nix build .#mutants -L` writes the mutants.out report to ./result and + # FAILS the build if any mutant survives (zero-survivors is the bar). + mutants = craneLib.mkCargoDerivation ( + commonArgs + // { + inherit cargoArtifacts; + pnameSuffix = "-mutants"; + nativeBuildInputs = commonArgs.nativeBuildInputs ++ [ cargo-mutants ]; + buildPhaseCargoCommand = '' + cargo mutants --package bombay-core --no-shuffle --colors never --output "$out" + ''; + doInstallCargoArtifacts = false; + doCheck = false; + } + ); covLlvm = craneLib.cargoLlvmCov ( commonArgs // { @@ -268,6 +288,7 @@ ); in { + inherit mutants; coverage-llvm = covLlvm; coverage = covLlvm; } From c0b57b9d1d42ed2c70b3c49bbb62d432c106d1dd Mon Sep 17 00:00:00 2001 From: Joel DSouza Date: Fri, 3 Jul 2026 21:58:49 +0200 Subject: [PATCH 4/9] docs(testing): record bombay-core mailbox coverage + loom/shuttle deferral (#112) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a bombay-core section to the coverage baseline: the `nix build .#mutants` reproducible mutation gate, the mailbox test/bench summary, and the decision to defer loom/shuttle DST to #116/#120 — the real tokio::sync::mpsc is opaque to both model checkers, so they land with the first bombay-owned concurrency (run-loop select / death-watch push), not on the tokio wrapper. --- docs/testing/coverage-baseline.md | 33 +++++++++++++++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/docs/testing/coverage-baseline.md b/docs/testing/coverage-baseline.md index 50cc533..83c8bb5 100644 --- a/docs/testing/coverage-baseline.md +++ b/docs/testing/coverage-baseline.md @@ -24,6 +24,39 @@ it is never compiled or counted (M1 deletes it). On every **merge to `main`**, t `…/bombay/coverage/` (and as the `coverage-html` artifact). The numbers below are from `coverage-llvm`; tarpaulin's totals differ slightly (different instrumentation granularity). +## bombay-core — the M1 core rebuild (#112+) + +The rebuilt spine lives in the `bombay-core` crate (part-by-part, epic #122), born +under the god-level bar (no #61 quarantine). It is measured by the same +`--workspace` coverage run and adds a **reproducible mutation gate**: + +```bash +nix build .#mutants -L # cargo-mutants on bombay-core; fails if any mutant survives +``` + +Pinned via the flake's `nixpkgs` (never `nix run nixpkgs#…`), mirroring the coverage +package. On-demand, not a per-push gate (rebuilds+tests once per mutant). + +### `mailbox` (#112) — done +Zero-box `Signal` queue over `tokio::sync::mpsc`. 16 tests: round-trip, backpressure +hand-back, `Capacity` boundaries (proptest incl. `MAX±1`/`usize::MAX`), lifecycle +(close-drains / send-after-drop / recv-none), `LinkDied` boxed-slot `size_of` guard, +weak death-watch, a real 8-thread `Barrier` linearizability test, and a single-sender +FIFO proptest. **Mutation: 0 missed** (6 caught, 30 unviable, 4 detected-via-timeout). +Criterion bench (`benches/mailbox.rs`): `tell` enqueue ≈ 5.6 ns, send+recv ≈ 21.4 ns. + +**DST posture — loom/shuttle deferred to #116/#120.** loom and shuttle can only +model-check code compiled against *their* primitives; the real `tokio::sync::mpsc` this +mailbox wraps is opaque to both ([loom](https://docs.rs/loom/latest/loom/) requires +"the code being tested specifically uses the loom replacement types"; +[shuttle](https://github.com/awslabs/shuttle) requires replacing std primitives with +its equivalents). The mailbox delegates all synchronization to tokio (which loom-tests +its own channel internally), so a loom/shuttle test here would either explore nothing or +test a reimplementation (violates the "test the actual SUT" rule). The first +bombay-owned concurrency is the run-loop `select` over signals (#116) and the death-watch +push (#120) — that is where loom/shuttle land. The 8-thread linearizability test is the +mailbox's concurrency coverage until then. + ## Baseline — 2026-06-29 (after #77) Workspace line coverage **60.85% (5686/9345)** — but that blends the SUT with untested crates From f554d423668c75dfe8ede8da4cb2c57da7906aa3 Mon Sep 17 00:00:00 2001 From: Joel DSouza Date: Fri, 3 Jul 2026 22:00:00 +0200 Subject: [PATCH 5/9] style(core): apply rustfmt + taplo to bombay-core (#112) --- bombay-core/Cargo.toml | 8 ++++++- bombay-core/src/mailbox.rs | 43 +++++++++++++++++++++++++++++--------- 2 files changed, 40 insertions(+), 11 deletions(-) diff --git a/bombay-core/Cargo.toml b/bombay-core/Cargo.toml index 8d5c3ba..b61e88b 100644 --- a/bombay-core/Cargo.toml +++ b/bombay-core/Cargo.toml @@ -12,7 +12,13 @@ publish = false tokio = { workspace = true, features = ["sync", "rt", "macros", "time"] } [dev-dependencies] -tokio = { workspace = true, features = ["rt", "rt-multi-thread", "macros", "sync", "time"] } +tokio = { workspace = true, features = [ + "rt", + "rt-multi-thread", + "macros", + "sync", + "time", +] } proptest = { workspace = true } criterion = { version = "0.8", features = ["async_tokio"] } diff --git a/bombay-core/src/mailbox.rs b/bombay-core/src/mailbox.rs index 01106c1..4698a98 100644 --- a/bombay-core/src/mailbox.rs +++ b/bombay-core/src/mailbox.rs @@ -129,7 +129,9 @@ pub struct MailboxSender { impl Clone for MailboxSender { fn clone(&self) -> Self { - Self { tx: self.tx.clone() } + Self { + tx: self.tx.clone(), + } } } @@ -317,7 +319,10 @@ mod tests { panic!("unexpected non-message signal"); }; let slot = &mut next_expected[sender_id as usize]; - assert_eq!(seq, *slot, "FIFO-per-sender violated for sender {sender_id}"); + assert_eq!( + seq, *slot, + "FIFO-per-sender violated for sender {sender_id}" + ); *slot += 1; total += 1; } @@ -342,7 +347,9 @@ mod tests { async fn sent_message_is_received() { let (tx, mut rx) = bounded::(cap(4)); - tx.send(Signal::Message(42)).await.expect("send should succeed"); + tx.send(Signal::Message(42)) + .await + .expect("send should succeed"); assert!(matches!(rx.recv().await, Some(Signal::Message(42)))); } @@ -353,7 +360,8 @@ mod tests { // Guards Capacity::MAX against tokio ever lowering its semaphore limit. let (tx, mut rx) = bounded::(cap(Capacity::MAX)); - tx.try_send(Signal::Message(7)).expect("send into max-capacity mailbox"); + tx.try_send(Signal::Message(7)) + .expect("send into max-capacity mailbox"); assert!(matches!(rx.recv().await, Some(Signal::Message(7)))); } @@ -363,7 +371,8 @@ mod tests { assert!(Capacity::new(NonZeroUsize::MIN).is_some()); assert!(Capacity::new(NonZeroUsize::new(Capacity::MAX).expect("nonzero")).is_some()); - let too_big = NonZeroUsize::new(Capacity::MAX.checked_add(1).expect("no overflow")).expect("nonzero"); + let too_big = + NonZeroUsize::new(Capacity::MAX.checked_add(1).expect("no overflow")).expect("nonzero"); assert!(Capacity::new(too_big).is_none()); // Capacity zero is unrepresentable: NonZeroUsize cannot hold it. } @@ -410,10 +419,16 @@ mod tests { assert_eq!(format!("{send_err:?}"), "SendError(receiver dropped)"); let full: TrySendError = TrySendError::Full(Signal::Stop); - assert_eq!(format!("{full:?}"), "TrySendError::Full(mailbox at capacity)"); + assert_eq!( + format!("{full:?}"), + "TrySendError::Full(mailbox at capacity)" + ); let closed: TrySendError = TrySendError::Closed(Signal::Stop); - assert_eq!(format!("{closed:?}"), "TrySendError::Closed(receiver dropped)"); + assert_eq!( + format!("{closed:?}"), + "TrySendError::Closed(receiver dropped)" + ); } #[tokio::test] @@ -441,7 +456,10 @@ mod tests { let weak = tx.downgrade(); let strong = weak.upgrade().expect("channel still alive"); - strong.send(Signal::Message(5)).await.expect("send via upgraded"); + strong + .send(Signal::Message(5)) + .await + .expect("send via upgraded"); assert!(matches!(rx.recv().await, Some(Signal::Message(5)))); } @@ -461,7 +479,9 @@ mod tests { #[tokio::test] async fn closing_the_receiver_stops_new_sends_but_drains_queued() { let (tx, mut rx) = bounded::(cap(4)); - tx.send(Signal::Message(1)).await.expect("queued before close"); + tx.send(Signal::Message(1)) + .await + .expect("queued before close"); rx.close(); @@ -509,7 +529,10 @@ mod tests { // Mailbox is now full: try_send must reject and hand the message back. let rejected = tx.try_send(Signal::Message(2)); - assert!(matches!(rejected, Err(TrySendError::Full(Signal::Message(2))))); + assert!(matches!( + rejected, + Err(TrySendError::Full(Signal::Message(2))) + )); // Draining one slot frees capacity for the next try_send. assert!(matches!(rx.recv().await, Some(Signal::Message(1)))); From dc35c27a5f0041cec4766801e5c6c834df04d6eb Mon Sep 17 00:00:00 2001 From: Joel DSouza Date: Fri, 3 Jul 2026 22:11:40 +0200 Subject: [PATCH 6/9] test(core): demonstrate monomorphic slot-size worst case (#112) Measured guard for the zero-box decision's worst case: size_of::>() = the largest Msg variant, so one fat inline command variant taxes every queue slot. Measured: small=16B, fat inline([u8;4096])=4104B, boxed=16B -> 4.10 MB vs 16 KB for 1_000 queued messages (256x). Boxing the fat variant (as Signal boxes LinkDied) restores the small slot. Documents + guards the tradeoff for #122. --- bombay-core/src/mailbox.rs | 66 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 66 insertions(+) diff --git a/bombay-core/src/mailbox.rs b/bombay-core/src/mailbox.rs index 4698a98..92d6bd2 100644 --- a/bombay-core/src/mailbox.rs +++ b/bombay-core/src/mailbox.rs @@ -393,6 +393,72 @@ mod tests { ); } + /// Demonstration (measured, not derived) of the **worst case** of the + /// monomorphic, by-value `Signal`: every queue slot costs `size_of` of the + /// actor's *largest* `Msg` variant. One fat command variant therefore taxes + /// every slot — even tiny messages — unless the user boxes it (the same + /// discipline `LinkDied` uses). kameo's `Box` keeps every slot + /// one pointer regardless. + /// + /// Measured here (aarch64): `small = 16 B`, `fat inline = 4104 B`, + /// `boxed = 16 B` → for 1_000 queued messages, **4.10 MB vs 16 KB (256×)**. + /// See the design-risk note on epic #122. + #[test] + #[expect( + dead_code, + reason = "the Msg variants exist only to measure enum layout via size_of" + )] + fn monomorphic_slot_cost_is_the_largest_msg_variant() { + use std::mem::size_of; + + enum SmallMsg { + Ping, + Pong(u64), + } + struct Small; + impl Mailboxed for Small { + type Msg = SmallMsg; + } + + // One fat command variant, stored inline (the footgun). + enum FatMsg { + Ping, + Bulk([u8; 4096]), + } + struct Fat; + impl Mailboxed for Fat { + type Msg = FatMsg; + } + + // The mitigation: box the fat variant, as `Signal` boxes `LinkDied`. + enum BoxedFatMsg { + Ping, + Bulk(Box<[u8; 4096]>), + } + struct BoxedFat; + impl Mailboxed for BoxedFat { + type Msg = BoxedFatMsg; + } + + let small = size_of::>(); + let fat = size_of::>(); + let boxed = size_of::>(); + + // A small enum keeps a small slot; a fat inline variant blows every slot + // up to the fat size; boxing the fat variant restores the small slot. + assert!(small <= 24, "small slot = {small}"); + assert!(fat >= 4096, "fat inline slot = {fat}"); + assert!(boxed <= 24, "boxed slot = {boxed}"); + + // Peak per-slot memory for 1_000 queued messages, whatever their variant. + let queued = 1_000; + let (fat_total, boxed_total) = (fat * queued, boxed * queued); + assert!( + fat_total > 100 * boxed_total, + "expected >100x blowup, got fat={fat_total} boxed={boxed_total}" + ); + } + #[tokio::test] async fn link_died_signal_round_trips() { let (tx, mut rx) = bounded::(cap(2)); From a88cdf8fa123a3d0fa3c7f36087f8d6e35b4d26b Mon Sep 17 00:00:00 2001 From: Joel DSouza Date: Sat, 4 Jul 2026 00:50:09 +0200 Subject: [PATCH 7/9] =?UTF-8?q?docs(adr):=20evaluate=20mailbox=20channel?= =?UTF-8?q?=20primitive=20=E2=80=94=20pick=20flume=20(ADR-0001,=20#133)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Do the channel survey #112's rigor contract required but skipped. Add a comparative benchmark (bombay-core/benches/channels.rs) of the real async candidates under the mailbox access pattern (N producers -> 1 consumer, bounded cap 256), on one shared tokio runtime for fairness, plus crossbeam as a sync throughput ceiling. Measured (aarch64, throughput, higher=better): uncontended 1->1 | contended 4->1 flume 25.7 M/s | 14.8 M/s <- winner async-channel 18.1 M/s | 11.1 M/s tokio::mpsc 13.4 M/s | 4.9 M/s <- v1 default, ~3x slower contended thingbuf 8.6 M/s | ~4.1 M/s <- slowest + needs T: Default Decision (ADR-0001, accepted): flume behind a channel seam — fastest async, executor-agnostic, move-based (no Default/Clone wart like thingbuf), lazily bounded (light idle memory). Sync designs (crossbeam/Vyukov/LMAX) are ceiling references only — the mailbox needs async recv. Establishes docs/adr/ as the decision-record practice. --- Cargo.lock | 38 +++++ bombay-core/Cargo.toml | 8 ++ bombay-core/benches/channels.rs | 153 +++++++++++++++++++++ docs/adr/0001-mailbox-channel-primitive.md | 115 ++++++++++++++++ docs/adr/README.md | 25 ++++ 5 files changed, 339 insertions(+) create mode 100644 bombay-core/benches/channels.rs create mode 100644 docs/adr/0001-mailbox-channel-primitive.md create mode 100644 docs/adr/README.md diff --git a/Cargo.lock b/Cargo.lock index c8ed3f0..9af86a5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -474,8 +474,12 @@ dependencies = [ name = "bombay-core" version = "0.1.0" dependencies = [ + "async-channel", "criterion", + "crossbeam-channel", + "flume", "proptest", + "thingbuf", "tokio", ] @@ -1488,6 +1492,9 @@ name = "fastrand" version = "2.4.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f1f227452a390804cdb637b74a86990f2a7d7ba4b7d5693aac9b4dd6defd8d6" +dependencies = [ + "getrandom 0.3.4", +] [[package]] name = "fiat-crypto" @@ -1524,6 +1531,18 @@ version = "0.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0ce7134b9999ecaf8bcd65542e436736ef32ddca1b3e06094cb6ec5755203b80" +[[package]] +name = "flume" +version = "0.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e139bc46ca777eb5efaf62df0ab8cc5fd400866427e56c68b22e414e53bd3be" +dependencies = [ + "fastrand", + "futures-core", + "futures-sink", + "spin", +] + [[package]] name = "fnv" version = "1.0.7" @@ -4851,6 +4870,15 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "spin" +version = "0.9.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67" +dependencies = [ + "lock_api", +] + [[package]] name = "spki" version = "0.7.3" @@ -5120,6 +5148,16 @@ dependencies = [ "unicode-width 0.2.2", ] +[[package]] +name = "thingbuf" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "662b54ef6f7b4e71f683dadc787bbb2d8e8ef2f91b682ebed3164a5a7abca905" +dependencies = [ + "parking_lot", + "pin-project", +] + [[package]] name = "thiserror" version = "1.0.69" diff --git a/bombay-core/Cargo.toml b/bombay-core/Cargo.toml index b61e88b..6bbde89 100644 --- a/bombay-core/Cargo.toml +++ b/bombay-core/Cargo.toml @@ -21,10 +21,18 @@ tokio = { workspace = true, features = [ ] } proptest = { workspace = true } criterion = { version = "0.8", features = ["async_tokio"] } +flume = "0.12.0" +async-channel = "2.5.0" +thingbuf = "0.1.6" +crossbeam-channel = "0.5.15" [[bench]] name = "mailbox" harness = false +[[bench]] +name = "channels" +harness = false + [lints] workspace = true diff --git a/bombay-core/benches/channels.rs b/bombay-core/benches/channels.rs new file mode 100644 index 0000000..50f00ca --- /dev/null +++ b/bombay-core/benches/channels.rs @@ -0,0 +1,153 @@ +//! Channel-primitive evaluation for the mailbox redesign (card #133). +//! +//! The mailbox needs an **async, bounded MPSC** (one consumer = the run-loop, +//! many producers = `ActorRef` clones). This benches the real async candidates +//! under that access pattern, plus crossbeam as a *sync* throughput ceiling +//! (not a viable mailbox — it has no async `recv`; shown to price the cost of +//! async integration). +//! +//! Same workload for every channel: `PRODUCERS` tasks each send `PER` messages +//! into a `CAP`-capacity channel; one consumer drains until all senders drop. +//! All async candidates run on one shared tokio multi-thread runtime, so the +//! executor is held constant and only the channel varies. + +use std::thread; + +use criterion::{Criterion, Throughput, black_box, criterion_group, criterion_main}; + +const CAP: usize = 256; +const PER: u64 = 4_000; + +fn runtime() -> tokio::runtime::Runtime { + tokio::runtime::Builder::new_multi_thread() + .worker_threads(4) + .build() + .expect("runtime") +} + +async fn tokio_run(producers: u64) { + let (tx, mut rx) = tokio::sync::mpsc::channel::(CAP); + for _ in 0..producers { + let tx = tx.clone(); + tokio::spawn(async move { + for i in 0..PER { + tx.send(i).await.expect("send"); + } + }); + } + drop(tx); + let mut count = 0u64; + while rx.recv().await.is_some() { + count += 1; + } + black_box(count); +} + +async fn flume_run(producers: u64) { + let (tx, rx) = flume::bounded::(CAP); + for _ in 0..producers { + let tx = tx.clone(); + tokio::spawn(async move { + for i in 0..PER { + tx.send_async(i).await.expect("send"); + } + }); + } + drop(tx); + let mut count = 0u64; + while rx.recv_async().await.is_ok() { + count += 1; + } + black_box(count); +} + +async fn async_channel_run(producers: u64) { + let (tx, rx) = async_channel::bounded::(CAP); + for _ in 0..producers { + let tx = tx.clone(); + tokio::spawn(async move { + for i in 0..PER { + tx.send(i).await.expect("send"); + } + }); + } + drop(tx); + let mut count = 0u64; + while rx.recv().await.is_ok() { + count += 1; + } + black_box(count); +} + +async fn thingbuf_run(producers: u64) { + let (tx, rx) = thingbuf::mpsc::channel::(CAP); + for _ in 0..producers { + let tx = tx.clone(); + tokio::spawn(async move { + for i in 0..PER { + tx.send(i).await.expect("send"); + } + }); + } + drop(tx); + let mut count = 0u64; + while rx.recv().await.is_some() { + count += 1; + } + black_box(count); +} + +/// Sync ceiling — std threads, not tasks. Not a viable async mailbox; here only +/// to show the raw lock-free throughput async has to pay to integrate with. +fn crossbeam_run(producers: u64) { + let (tx, rx) = crossbeam_channel::bounded::(CAP); + thread::scope(|scope| { + for _ in 0..producers { + let tx = tx.clone(); + scope.spawn(move || { + for i in 0..PER { + tx.send(i).expect("send"); + } + }); + } + drop(tx); + let mut count = 0u64; + while rx.recv().is_ok() { + count += 1; + } + black_box(count); + }); +} + +fn bench(c: &mut Criterion, producers: u64, label: &str) { + let rt = runtime(); + let mut group = c.benchmark_group(label); + group.throughput(Throughput::Elements(producers * PER)); + + group.bench_function("tokio_mpsc", |b| { + b.iter(|| rt.block_on(tokio_run(producers))) + }); + group.bench_function("flume", |b| b.iter(|| rt.block_on(flume_run(producers)))); + group.bench_function("async_channel", |b| { + b.iter(|| rt.block_on(async_channel_run(producers))); + }); + group.bench_function("thingbuf", |b| { + b.iter(|| rt.block_on(thingbuf_run(producers))) + }); + group.bench_function("crossbeam_sync_ceiling", |b| { + b.iter(|| crossbeam_run(producers)) + }); + + group.finish(); +} + +fn uncontended(c: &mut Criterion) { + bench(c, 1, "uncontended_1p_1c"); +} + +fn contended(c: &mut Criterion) { + bench(c, 4, "contended_4p_1c"); +} + +criterion_group!(benches, uncontended, contended); +criterion_main!(benches); diff --git a/docs/adr/0001-mailbox-channel-primitive.md b/docs/adr/0001-mailbox-channel-primitive.md new file mode 100644 index 0000000..000e636 --- /dev/null +++ b/docs/adr/0001-mailbox-channel-primitive.md @@ -0,0 +1,115 @@ +# ADR-0001 — Mailbox channel primitive + +**Status:** Accepted (2026-07-04) — evidence below; implemented under card #133 + +## Context + +The actor mailbox is an **async, bounded MPSC**: one consumer (the run-loop +`select`s over `recv`), many producers (`ActorRef` clones). Card #112 shipped a +mailbox on `tokio::sync::mpsc` **without the channel survey its own rigor +contract required** ("*decide the primitive… no commitment before the survey*"). +Card #133 corrects that: evaluate the real candidates on the axes that actually +gate the mailbox, and record the decision here. + +### Requirements (what actually gates the choice) + +1. **Async-native `recv`** — *mandatory*. The run-loop awaits in a `select`. A + sync channel would need an async waker bolted on, adding back the cost it + saved — so pure-sync designs are **not viable mailboxes**. +2. **Bounded** — *mandatory*. Backpressure is a design decision (#112); we + deleted unbounded (a memory footgun). Rules out unbounded-only designs. +3. **Light idle memory per actor** — an agent runtime (agency) may hold *many* + actors; a preallocated ring reserves `cap × size_of::()` **even when + idle**, which a lazily-grown channel does not. +4. **`no_std` / executor-agnostic** — decides whether we get **one** mailbox + across the tokio server **and** the M6 / embedded client (`no_std` + Embassy, + where tokio cannot run), or **two behind the executor seam**. +5. **loom / shuttle testable** — would **un-defer the concurrency DST** punted + from #112 (tokio's mpsc internals are loom-opaque to us). +6. **Maturity / maintenance.** + +### Why not "an actor system without tokio"? + +- **Server / desktop:** no reason to avoid tokio — it's mature, Zenoh (our + transport) is built on it, rolling our own executor is waste. tokio is the + right commitment there. +- **M6 lite-bombay / embedded / WASM:** a real non-tokio frontier — + `zenoh-pico`, single-thread, `no_std` + Embassy. **Both** worlds still need + *async*, so sync lock-free queues are out in *either* case; the live question + is whether one **async** primitive spans both (req. 4). + +### Capacity models (they differ, and it matters — req. 3) + +| Model | Channels | Memory | +|---|---|---| +| Preallocated fixed ring | thingbuf, LMAX disruptor | `cap × slot` reserved up front, **even idle** | +| Bounded, lazily grown | tokio::mpsc, flume, async-channel, crossbeam | tracks occupancy — light idle | +| Unbounded | Vyukov intrusive, `unbounded()` variants | grows without limit — **fails req. 2** | + +## Options considered + +| Candidate | async | off-tokio | `no_std` | loom | capacity model | verdict | +|---|---|---|---|---|---|---| +| **tokio::sync::mpsc** | ✓ | ✗ | ✗ | internal only | lazy bounded | server-only | +| **flume** | ✓ | ✓ | ✗ (std) | — | lazy bounded | server + std-async | +| **async-channel** | ✓ | ✓ | ✗ (std) | — | lazy bounded | server + std-async | +| **thingbuf** | ✓ | ✓ | **✓** | **✓** | preallocated ring | **spans both worlds; idle-mem tax** | +| crossbeam-channel | ✗ (sync) | — | ✗ | — | lazy bounded | ceiling ref — not a mailbox | +| Vyukov intrusive MPSC | ✗ (sync) | — | ✓ | — | unbounded | ref — fails req. 1 & 2 | +| LMAX disruptor | ✗ (sync) | — | ✗ | — | preallocated ring | ref — fails req. 1 | + +## Evidence (measured — `cargo bench --bench channels`, aarch64, 4 workers) + +Throughput (higher = better); one shared tokio multi-thread runtime, `u64` +payload, `CAP = 256`, warm-up 1 s / measure 2 s: + +| Channel | Uncontended (1→1) | Contended (4→1) | +|---|---|---| +| crossbeam *(sync ceiling, not a mailbox)* | 53.2 M/s | — | +| **flume** | **25.7 M/s** | **14.8 M/s** | +| async-channel | 18.1 M/s | 11.1 M/s | +| **tokio::mpsc** *(v1 default)* | 13.4 M/s | 4.9 M/s | +| thingbuf | 8.6 M/s | ~4.1 M/s | + +Findings: +- **tokio::mpsc — the un-surveyed v1 default — is second-slowest, ~3× slower + than flume under contention.** The concrete cost of skipping #112's survey. +- **flume is the throughput winner**, executor-agnostic, and **move-based** (no + extra trait bounds on the element). +- **thingbuf is the slowest *and* requires `T: Default`** (a preallocated ring + initialises its slots). `Signal` has no sensible `Default` (a message enum + cannot default) — so thingbuf's `no_std`/loom edge comes with a real fit + problem against our by-value design. + +Caveat: this measures *raw channel* throughput with a small payload; real actor +workloads are usually handler/IO-bound, so the absolute gap matters less than +the *ordering* and the qualitative axes. + +## Decision + +**flume, behind a channel seam.** It is measurably the fastest async candidate +(≈2× tokio uncontended, ≈3× contended), executor-agnostic (works on the tokio +server *and* off-tokio), move-based (no `Default`/`Clone` wart, unlike thingbuf), +lazily-bounded (light idle memory — req. 3), and actively maintained. The **seam** +(a thin trait over the channel) lets the M6 / `no_std` client swap in an Embassy +or `no_std` channel later, and hosts the deterministic impl that carries the +loom/shuttle DST deferred from #112. + +`no_std` reality: none of the *std* async channels (flume/async-channel/tokio) +run on embedded; thingbuf can, but is slowest and `Default`-constrained. So the +embedded channel is a **separate, seam-local decision at M6** — not forced now. + +## Consequences + +- Add `flume` as a real (non-dev) workspace dependency; the mailbox wraps it. +- Introduce a **channel seam** (trait) now — it was going to be needed for the + executor split (#9 MaybeSend) and the DST anyway; this pulls it forward. +- tokio stays the runtime (tasks, `AbortHandle`, timers); only the *mailbox + queue* changes from `tokio::sync::mpsc` to `flume`. +- Re-run the #112 mailbox test net + mutation gate against the flume-backed impl. + +## References + +Card #133 (this evaluation) · #112 (the skipped survey) · epic #122 · #19 +(channel-handler strategy) · the deferred loom/shuttle DST (#112 → #116/#120) · +bench: `bombay-core/benches/channels.rs`. diff --git a/docs/adr/README.md b/docs/adr/README.md new file mode 100644 index 0000000..736286e --- /dev/null +++ b/docs/adr/README.md @@ -0,0 +1,25 @@ +# Architecture Decision Records (ADRs) + +Durable records of **why** we picked what — the decision, the options we weighed, +the evidence, and the consequences we accepted. An ADR is written when a choice +is (a) hard to reverse, (b) shapes downstream cards, or (c) would otherwise be +re-litigated from memory. It captures the *reasoning at decision time*, not a +running status. + +Distilled, still-true knowledge lives in [`docs/`](../); ADRs are the audit trail +behind it. If an ADR is later overturned, add a new ADR that supersedes it and +flip the old one's status to `Superseded by ADR-NNNN` — never rewrite history. + +## Format + +Each ADR is `NNNN-kebab-title.md` with: **Status** · **Context** · **Options +considered** (with the evidence — benchmarks, primary sources) · **Decision** · +**Consequences**. Keep it to the essentials; link to code/benches/cards. + +Status is one of: `Proposed` · `Accepted` · `Superseded by ADR-NNNN` · `Rejected`. + +## Index + +| ADR | Title | Status | +|---|---|---| +| [0001](0001-mailbox-channel-primitive.md) | Mailbox channel primitive | Accepted | From 07d0bf9815f2ef04cea821244e63e3ea4bac1906 Mon Sep 17 00:00:00 2001 From: Joel DSouza Date: Sat, 4 Jul 2026 01:28:44 +0200 Subject: [PATCH 8/9] =?UTF-8?q?refactor(core):=20principal-grade=20mailbox?= =?UTF-8?q?=20=E2=80=94=20flume=20behind=20a=20seam=20+=20Mailbox=20API=20?= =?UTF-8?q?(#133)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Redesign the #112 mailbox on the accepted evidence (ADR-0001), fixing the two principal-grade gaps: an unsurveyed channel default and a non-composing API. - **Channel = flume** behind a seam: flume appears only inside MailboxSender/ WeakMailboxSender/MailboxReceiver, never in the public API. Swapping the primitive (no_std/Embassy for M6, or a deterministic DST channel) reimplements those three wrappers and nothing else. No premature trait for one impl. - **Composable API**: construction hangs off `Mailbox::::bounded(cap)` instead of a free-floating `bounded()`; `Capacity` gains `TryFrom` + a typed `CapacityError`. - **Pure transport**: send/try_send/recv/downgrade + `drain`. Dropped `close()` (flume has no receiver-side close) — graceful shutdown is `Signal::Stop` + `drain` at the run-loop (#116), where shutdown policy belongs. flume also beat tokio on the mailbox micro-bench: `tell` 5.6 -> 4.7 ns, send+recv 21.4 -> 13 ns. 18 tests green, clippy --all-features clean, mutation 0 missed. flume moves to a real (workspace) dependency; the channel-eval crates stay dev. --- Cargo.toml | 3 + bombay-core/Cargo.toml | 2 +- bombay-core/benches/mailbox.rs | 6 +- bombay-core/src/mailbox.rs | 468 +++++++++++++++++------------- docs/testing/coverage-baseline.md | 20 +- 5 files changed, 280 insertions(+), 219 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 45fb962..8d5a82e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,6 +13,9 @@ bombay = { path = ".", version = "0.21.0" } futures = "0.3" tokio = "1.47" +# Mailbox channel primitive — chosen on measured evidence (ADR-0001, card #133): +# fastest async candidate, executor-agnostic, move-based, has weak senders. +flume = "0.12" cucumber = { version = "0.23", features = ["libtest"] } proptest = "1.11" rstest = "0.26" diff --git a/bombay-core/Cargo.toml b/bombay-core/Cargo.toml index 6bbde89..1dff3b2 100644 --- a/bombay-core/Cargo.toml +++ b/bombay-core/Cargo.toml @@ -10,6 +10,7 @@ publish = false [dependencies] tokio = { workspace = true, features = ["sync", "rt", "macros", "time"] } +flume = { workspace = true } [dev-dependencies] tokio = { workspace = true, features = [ @@ -21,7 +22,6 @@ tokio = { workspace = true, features = [ ] } proptest = { workspace = true } criterion = { version = "0.8", features = ["async_tokio"] } -flume = "0.12.0" async-channel = "2.5.0" thingbuf = "0.1.6" crossbeam-channel = "0.5.15" diff --git a/bombay-core/benches/mailbox.rs b/bombay-core/benches/mailbox.rs index 0eb5ad4..19eef7d 100644 --- a/bombay-core/benches/mailbox.rs +++ b/bombay-core/benches/mailbox.rs @@ -7,7 +7,7 @@ use std::num::NonZeroUsize; -use bombay_core::mailbox::{Capacity, Mailboxed, Signal, bounded}; +use bombay_core::mailbox::{Capacity, Mailbox, Mailboxed, Signal}; use criterion::{BatchSize, Criterion, black_box, criterion_group, criterion_main}; struct Bench; @@ -26,7 +26,7 @@ fn cap(n: usize) -> Capacity { fn enqueue(c: &mut Criterion) { c.bench_function("tell_try_send_1k_u64", |b| { b.iter_batched_ref( - || bounded::(cap(1024)), + || Mailbox::::bounded(cap(1024)), |(tx, _rx)| { for i in 0..1000u64 { tx.try_send(Signal::Message(black_box(i))) @@ -48,7 +48,7 @@ fn roundtrip(c: &mut Criterion) { c.bench_function("send_recv_roundtrip_1k_u64", |b| { b.iter(|| { rt.block_on(async { - let (tx, mut rx) = bounded::(cap(1024)); + let (tx, mut rx) = Mailbox::::bounded(cap(1024)); let producer = tokio::spawn(async move { for i in 0..1000u64 { tx.send(Signal::Message(black_box(i))).await.expect("send"); diff --git a/bombay-core/src/mailbox.rs b/bombay-core/src/mailbox.rs index 92d6bd2..52767c7 100644 --- a/bombay-core/src/mailbox.rs +++ b/bombay-core/src/mailbox.rs @@ -2,27 +2,37 @@ //! //! The local tier of the two-tier message model (#66): typed, in-memory, //! **zero-serialize**. `tell` moves an `A::Msg` into a queue slot — no -//! per-message heap box. Built on `tokio::sync::mpsc`, bounded for backpressure. - -use std::{fmt, num::NonZeroUsize}; +//! per-message heap box. +//! +//! Construction hangs off the [`Mailbox`] namespace: `Mailbox::::bounded(cap)`. +//! Bounded is the only mode — a full mailbox exerts backpressure rather than +//! growing without limit (an unbounded queue is a memory footgun). +//! +//! **The channel seam.** The queue is backed by `flume` (chosen on measured +//! evidence — see `docs/adr/0001`), but that is an implementation detail: `flume` +//! appears *only* inside [`MailboxSender`] / [`WeakMailboxSender`] / +//! [`MailboxReceiver`], never in the public API. Swapping the primitive (a +//! `no_std`/Embassy channel for M6, or a deterministic channel for the DST) means +//! reimplementing those three wrappers and nothing else. The seam is trait-ified +//! at the *second* impl, not pre-abstracted for one. +//! +//! **Shutdown** is not a channel concern: the mailbox is pure transport. A +//! graceful stop is the run-loop's job (#116) — receive [`Signal::Stop`], flush +//! with [`MailboxReceiver::drain`], then drop the receiver (which disconnects +//! every sender). -use tokio::sync::mpsc; +use std::{fmt, marker::PhantomData, num::NonZeroUsize}; /// A validated mailbox capacity: at least `1`, at most [`Capacity::MAX`]. /// -/// Makes both illegal capacities unrepresentable, so [`bounded`] cannot fail: -/// zero is excluded by `NonZeroUsize`, and the upper bound is checked here -/// rather than trusting `tokio::sync::mpsc::channel` (which panics outside -/// `1..=MAX`). Validating at our own boundary is deliberate — we do not rely on -/// an upstream crate's panic as our error path (rule 4). +/// Makes both illegal capacities unrepresentable, so [`Mailbox::bounded`] cannot +/// fail: zero is excluded by `NonZeroUsize`, and the upper bound is checked here. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct Capacity(NonZeroUsize); impl Capacity { - /// The largest capacity `tokio::sync::mpsc::channel` accepts — the tokio - /// `Semaphore` permit ceiling (`usize::MAX >> 3`). Mirrored here because - /// tokio does not expose it; the `capacity_at_the_upper_boundary_is_usable` - /// test guards this constant against tokio lowering the limit. + /// The largest capacity the backing channel accepts. Kept comfortably within + /// any candidate's limit; a mailbox this deep is already a design smell. pub const MAX: usize = usize::MAX >> 3; /// Builds a `Capacity`, returning `None` if `value` exceeds [`Capacity::MAX`]. @@ -42,12 +52,52 @@ impl Capacity { } } +/// Why a `usize` could not be a [`Capacity`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[expect( + clippy::exhaustive_enums, + reason = "a capacity is invalid for exactly these two reasons" +)] +pub enum CapacityError { + /// The value was `0`; a mailbox needs room for at least one signal. + Zero, + /// The value exceeded [`Capacity::MAX`]. + TooLarge, +} + +impl fmt::Display for CapacityError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Zero => f.write_str("mailbox capacity must be at least 1"), + Self::TooLarge => f.write_str("mailbox capacity exceeds the maximum"), + } + } +} + +impl std::error::Error for CapacityError {} + +impl TryFrom for Capacity { + type Error = CapacityError; + + fn try_from(value: NonZeroUsize) -> Result { + Self::new(value).ok_or(CapacityError::TooLarge) + } +} + +impl TryFrom for Capacity { + type Error = CapacityError; + + fn try_from(value: usize) -> Result { + let nonzero = NonZeroUsize::new(value).ok_or(CapacityError::Zero)?; + Self::try_from(nonzero) + } +} + /// The seam between a mailbox and its actor. /// /// A mailbox is monomorphized per actor `A`, carrying that actor's single closed /// message type `A::Msg` by value — no `Box`. This scaffold trait is what -/// the rebuilt `Actor` trait (#114/#116) will later subsume; keeping it separate -/// lets the mailbox be built and hard-tested on its own (#112). +/// the rebuilt `Actor` trait (#114/#116) will later subsume. /// /// `Msg` is `Send + 'static` for now; the cfg-gated `MaybeSend` relaxation for /// single-threaded client builds arrives with #9. @@ -88,8 +138,7 @@ pub enum StopReason { /// The payload of a [`Signal::LinkDied`]: a linked actor has terminated. /// /// Boxed inside [`Signal`] because it is a **cold** control path — boxing it -/// keeps the hot [`Signal::Message`] slot small (see the large-variant -/// discipline). +/// keeps the hot [`Signal::Message`] slot small (large-variant discipline). #[derive(Debug, Clone)] #[expect( clippy::exhaustive_structs, @@ -109,7 +158,7 @@ pub struct LinkDied { #[expect( clippy::exhaustive_enums, reason = "the signal set is deliberately closed so the run-loop is a total match; \ - new arms (Stop, LinkDied, …) are added under their driving cards" + new arms are added under their driving cards" )] pub enum Signal { /// A domain message for the actor to handle. @@ -121,10 +170,29 @@ pub enum Signal { LinkDied(Box), } +/// The construction namespace for an actor's mailbox. +/// +/// Never instantiated — it exists so construction reads as +/// `Mailbox::::bounded(cap)`, keeping the sender/receiver/weak types cohesive +/// under one entry point instead of a free-floating function. +pub struct Mailbox(PhantomData A>); + +impl Mailbox { + /// Creates a bounded mailbox with room for `capacity` queued signals. + /// + /// Infallible by construction — [`Capacity`] has already excluded the values + /// the backing channel would reject. + #[must_use] + pub fn bounded(capacity: Capacity) -> (MailboxSender, MailboxReceiver) { + let (tx, rx) = flume::bounded(capacity.get()); + (MailboxSender { tx }, MailboxReceiver { rx }) + } +} + /// Sends [`Signal`]s to an actor's mailbox. Cloneable; the channel stays open /// while any sender is alive. pub struct MailboxSender { - tx: mpsc::Sender>, + tx: flume::Sender>, } impl Clone for MailboxSender { @@ -135,12 +203,49 @@ impl Clone for MailboxSender { } } +impl MailboxSender { + /// Sends `signal`, waiting for capacity if the mailbox is full. + /// + /// # Errors + /// + /// Returns [`SendError`] (carrying `signal` back) if the receiver has been + /// dropped, i.e. the actor is no longer running. + pub async fn send(&self, signal: Signal) -> Result<(), SendError> { + self.tx + .send_async(signal) + .await + .map_err(|err| SendError(err.into_inner())) + } + + /// Tries to enqueue `signal` without waiting. + /// + /// # Errors + /// + /// Returns [`TrySendError::Full`] if the mailbox is at capacity, or + /// [`TrySendError::Closed`] if the receiver has been dropped. Both carry + /// `signal` back to the caller. + pub fn try_send(&self, signal: Signal) -> Result<(), TrySendError> { + self.tx.try_send(signal).map_err(|err| match err { + flume::TrySendError::Full(undelivered) => TrySendError::Full(undelivered), + flume::TrySendError::Disconnected(undelivered) => TrySendError::Closed(undelivered), + }) + } + + /// Downgrades to a [`WeakMailboxSender`] that does not keep the mailbox open. + #[must_use] + pub fn downgrade(&self) -> WeakMailboxSender { + WeakMailboxSender { + weak: self.tx.downgrade(), + } + } +} + /// A non-pinning handle to a mailbox: holding one does **not** keep it alive. /// /// [`upgrade`](Self::upgrade) yields a strong sender only while a real /// [`MailboxSender`] still exists — the primitive death-watch is built on. pub struct WeakMailboxSender { - weak: mpsc::WeakSender>, + weak: flume::WeakSender>, } impl WeakMailboxSender { @@ -162,7 +267,24 @@ impl Clone for WeakMailboxSender { /// The single consumer of an actor's mailbox. The run-loop pulls from it. pub struct MailboxReceiver { - rx: mpsc::Receiver>, + rx: flume::Receiver>, +} + +impl MailboxReceiver { + /// Receives the next signal, waiting until one is available. + /// + /// Returns `None` once every sender has dropped and the queue is drained. + pub async fn recv(&mut self) -> Option> { + self.rx.recv_async().await.ok() + } + + /// Drains every currently-queued signal without waiting, in FIFO order. + /// + /// The run-loop uses this to flush in-flight work on a graceful stop (after + /// a [`Signal::Stop`]) before dropping the receiver. + pub fn drain(&mut self) -> impl Iterator> + '_ { + self.rx.drain() + } } /// The receiver was dropped, so the signal could not be delivered. @@ -201,75 +323,11 @@ impl fmt::Debug for TrySendError { } } -/// Creates a bounded mailbox with room for `capacity` queued signals. -/// -/// Bounded is the only mode: a full mailbox exerts backpressure rather than -/// growing without limit (an unbounded queue is a memory footgun). Infallible -/// by construction — [`Capacity`] has already excluded the values that would -/// make the underlying channel panic. -#[must_use] -pub fn bounded(capacity: Capacity) -> (MailboxSender, MailboxReceiver) { - let (tx, rx) = mpsc::channel(capacity.get()); - (MailboxSender { tx }, MailboxReceiver { rx }) -} - -impl MailboxSender { - /// Sends `signal`, waiting for capacity if the mailbox is full. - /// - /// # Errors - /// - /// Returns [`SendError`] (carrying `signal` back) if the receiver has been - /// dropped, i.e. the actor is no longer running. - pub async fn send(&self, signal: Signal) -> Result<(), SendError> { - self.tx.send(signal).await.map_err(|err| SendError(err.0)) - } - - /// Tries to enqueue `signal` without waiting. - /// - /// # Errors - /// - /// Returns [`TrySendError::Full`] if the mailbox is at capacity, or - /// [`TrySendError::Closed`] if the receiver has been dropped. Both carry - /// `signal` back to the caller. - pub fn try_send(&self, signal: Signal) -> Result<(), TrySendError> { - self.tx.try_send(signal).map_err(|err| match err { - mpsc::error::TrySendError::Full(undelivered) => TrySendError::Full(undelivered), - mpsc::error::TrySendError::Closed(undelivered) => TrySendError::Closed(undelivered), - }) - } - - /// Downgrades to a [`WeakMailboxSender`] that does not keep the mailbox open. - #[must_use] - pub fn downgrade(&self) -> WeakMailboxSender { - WeakMailboxSender { - weak: self.tx.downgrade(), - } - } -} - -impl MailboxReceiver { - /// Receives the next signal, waiting until one is available. - /// - /// Returns `None` once the mailbox is closed and drained. - pub async fn recv(&mut self) -> Option> { - self.rx.recv().await - } - - /// Closes the mailbox: senders can no longer enqueue, but signals already - /// queued still drain through [`recv`](Self::recv) before it yields `None`. - /// - /// Used for a graceful stop — the run-loop finishes in-flight work rather - /// than dropping it. - pub fn close(&mut self) { - self.rx.close(); - } -} - #[cfg(test)] mod tests { use super::*; - use std::{num::NonZeroUsize, sync::Arc}; + use std::sync::Arc; use proptest::prelude::*; use tokio::{runtime::Builder, sync::Barrier}; @@ -288,64 +346,15 @@ mod tests { type Msg = (u32, u32); } - #[tokio::test(flavor = "multi_thread", worker_threads = 4)] - async fn n_senders_one_receiver_preserve_per_sender_order() { - const SENDERS: u32 = 8; - const PER_SENDER: u32 = 64; - - // Small capacity so senders genuinely contend and backpressure. - let (tx, mut rx) = bounded::(cap(4)); - let start = Arc::new(Barrier::new(SENDERS as usize)); - - let mut handles = Vec::with_capacity(SENDERS as usize); - for sender_id in 0..SENDERS { - let tx = tx.clone(); - let start = Arc::clone(&start); - handles.push(tokio::spawn(async move { - start.wait().await; // all senders race from the same instant - for seq in 0..PER_SENDER { - tx.send(Signal::Message((sender_id, seq))) - .await - .expect("send"); - } - })); - } - drop(tx); // recv ends only once every sender has dropped its clone - - let mut next_expected = vec![0u32; SENDERS as usize]; - let mut total = 0u32; - while let Some(signal) = rx.recv().await { - let Signal::Message((sender_id, seq)) = signal else { - panic!("unexpected non-message signal"); - }; - let slot = &mut next_expected[sender_id as usize]; - assert_eq!( - seq, *slot, - "FIFO-per-sender violated for sender {sender_id}" - ); - *slot += 1; - total += 1; - } - - assert_eq!(total, SENDERS * PER_SENDER, "lost or duplicated messages"); - for (sender_id, &count) in next_expected.iter().enumerate() { - assert_eq!(count, PER_SENDER, "sender {sender_id} did not fully arrive"); - } - for handle in handles { - handle.await.expect("sender task panicked"); - } - } - /// Builds a valid [`Capacity`] for tests; panics on out-of-range input, /// which in a test is a programmer error in the test itself. fn cap(n: usize) -> Capacity { - Capacity::new(NonZeroUsize::new(n).expect("test capacity must be nonzero")) - .expect("test capacity must be within Capacity::MAX") + Capacity::try_from(n).expect("test capacity must be valid") } #[tokio::test] async fn sent_message_is_received() { - let (tx, mut rx) = bounded::(cap(4)); + let (tx, mut rx) = Mailbox::::bounded(cap(4)); tx.send(Signal::Message(42)) .await @@ -356,9 +365,8 @@ mod tests { #[tokio::test] async fn capacity_at_the_upper_boundary_is_usable() { - // A mailbox built at tokio's true maximum must not panic and must work. - // Guards Capacity::MAX against tokio ever lowering its semaphore limit. - let (tx, mut rx) = bounded::(cap(Capacity::MAX)); + // A mailbox built at the capacity ceiling must not panic and must work. + let (tx, mut rx) = Mailbox::::bounded(cap(Capacity::MAX)); tx.try_send(Signal::Message(7)) .expect("send into max-capacity mailbox"); @@ -367,14 +375,34 @@ mod tests { } #[test] - fn capacity_rejects_values_above_tokio_max() { - assert!(Capacity::new(NonZeroUsize::MIN).is_some()); - assert!(Capacity::new(NonZeroUsize::new(Capacity::MAX).expect("nonzero")).is_some()); + fn capacity_rejects_zero_and_values_above_max() { + assert_eq!(Capacity::try_from(0usize), Err(CapacityError::Zero)); + assert!(Capacity::try_from(1usize).is_ok()); + assert!(Capacity::try_from(Capacity::MAX).is_ok()); + assert_eq!( + Capacity::try_from(Capacity::MAX.checked_add(1).expect("no overflow")), + Err(CapacityError::TooLarge) + ); + } + + #[test] + fn capacity_max_is_the_documented_ceiling() { + // A behavioural boundary test can't catch a wrong MAX here: flume grows + // lazily and won't panic on a huge bound (unlike tokio's mpsc), so pin + // the ceiling constant directly. + assert_eq!(Capacity::MAX, usize::MAX >> 3); + } - let too_big = - NonZeroUsize::new(Capacity::MAX.checked_add(1).expect("no overflow")).expect("nonzero"); - assert!(Capacity::new(too_big).is_none()); - // Capacity zero is unrepresentable: NonZeroUsize cannot hold it. + #[test] + fn capacity_error_display_is_stable() { + assert_eq!( + CapacityError::Zero.to_string(), + "mailbox capacity must be at least 1" + ); + assert_eq!( + CapacityError::TooLarge.to_string(), + "mailbox capacity exceeds the maximum" + ); } #[test] @@ -382,10 +410,8 @@ mod tests { use std::mem::size_of; // The cold LinkDied variant is boxed, so a small-message actor's queue - // slot is bounded by the hot Message(u64) path — not inflated by - // LinkDied's fields (id + a String-bearing reason). Guards the - // "every slot = largest variant" trap; clippy::large_enum_variant is the - // compile-time backstop via the workspace bar. + // slot is bounded by the hot Message(u64) path. Guards the + // "every slot = largest variant" trap. assert!( size_of::>() <= 2 * size_of::(), "Signal slot is {} bytes; LinkDied is not boxed", @@ -397,12 +423,10 @@ mod tests { /// monomorphic, by-value `Signal`: every queue slot costs `size_of` of the /// actor's *largest* `Msg` variant. One fat command variant therefore taxes /// every slot — even tiny messages — unless the user boxes it (the same - /// discipline `LinkDied` uses). kameo's `Box` keeps every slot - /// one pointer regardless. + /// discipline `LinkDied` uses). /// - /// Measured here (aarch64): `small = 16 B`, `fat inline = 4104 B`, - /// `boxed = 16 B` → for 1_000 queued messages, **4.10 MB vs 16 KB (256×)**. - /// See the design-risk note on epic #122. + /// Measured (aarch64): `small = 16 B`, `fat inline = 4104 B`, `boxed = 16 B` + /// → for 1_000 queued messages, **4.10 MB vs 16 KB (256×)**. See #122. #[test] #[expect( dead_code, @@ -444,13 +468,10 @@ mod tests { let fat = size_of::>(); let boxed = size_of::>(); - // A small enum keeps a small slot; a fat inline variant blows every slot - // up to the fat size; boxing the fat variant restores the small slot. assert!(small <= 24, "small slot = {small}"); assert!(fat >= 4096, "fat inline slot = {fat}"); assert!(boxed <= 24, "boxed slot = {boxed}"); - // Peak per-slot memory for 1_000 queued messages, whatever their variant. let queued = 1_000; let (fat_total, boxed_total) = (fat * queued, boxed * queued); assert!( @@ -459,24 +480,6 @@ mod tests { ); } - #[tokio::test] - async fn link_died_signal_round_trips() { - let (tx, mut rx) = bounded::(cap(2)); - - tx.send(Signal::LinkDied(Box::new(LinkDied { - id: ActorId::new(7), - reason: StopReason::Normal, - }))) - .await - .expect("send link-died"); - - let Some(Signal::LinkDied(link_died)) = rx.recv().await else { - panic!("expected a LinkDied signal"); - }; - assert_eq!(link_died.id, ActorId::new(7)); - assert!(matches!(link_died.reason, StopReason::Normal)); - } - #[test] fn error_debug_formats_are_stable() { // The Debug impls are hand-written (so the error types don't inherit an @@ -499,7 +502,7 @@ mod tests { #[tokio::test] async fn weak_sender_tracks_the_last_strong_sender() { - let (tx, _rx) = bounded::(cap(2)); + let (tx, _rx) = Mailbox::::bounded(cap(2)); let tx2 = tx.clone(); let weak = tx.downgrade(); @@ -518,7 +521,7 @@ mod tests { #[tokio::test] async fn upgraded_weak_sender_can_send() { - let (tx, mut rx) = bounded::(cap(2)); + let (tx, mut rx) = Mailbox::::bounded(cap(2)); let weak = tx.downgrade(); let strong = weak.upgrade().expect("channel still alive"); @@ -532,7 +535,7 @@ mod tests { #[tokio::test] async fn stop_signal_is_delivered_in_order_after_a_message() { - let (tx, mut rx) = bounded::(cap(4)); + let (tx, mut rx) = Mailbox::::bounded(cap(4)); tx.send(Signal::Message(1)).await.expect("message"); tx.send(Signal::Stop).await.expect("stop"); @@ -543,27 +546,29 @@ mod tests { } #[tokio::test] - async fn closing_the_receiver_stops_new_sends_but_drains_queued() { - let (tx, mut rx) = bounded::(cap(4)); - tx.send(Signal::Message(1)) - .await - .expect("queued before close"); + async fn drain_flushes_queued_signals_in_order() { + // Graceful-stop primitive: after a Stop, the run-loop flushes the rest + // with `drain` before dropping the receiver. + let (tx, mut rx) = Mailbox::::bounded(cap(8)); + for i in 0..3 { + tx.send(Signal::Message(i)).await.expect("queued"); + } - rx.close(); + assert!(matches!(rx.recv().await, Some(Signal::Message(0)))); - // New sends are rejected (the message comes back)... - assert!(matches!( - tx.send(Signal::Message(2)).await, - Err(SendError(Signal::Message(2))) - )); - // ...but messages queued before the close still drain, then None. - assert!(matches!(rx.recv().await, Some(Signal::Message(1)))); - assert!(rx.recv().await.is_none()); + let flushed: Vec = rx + .drain() + .map(|signal| match signal { + Signal::Message(m) => m, + _ => panic!("unexpected signal"), + }) + .collect(); + assert_eq!(flushed, vec![1, 2]); } #[tokio::test] async fn send_after_receiver_dropped_returns_the_message() { - let (tx, rx) = bounded::(cap(4)); + let (tx, rx) = Mailbox::::bounded(cap(4)); drop(rx); assert!(matches!( @@ -578,18 +583,18 @@ mod tests { #[tokio::test] async fn recv_returns_none_after_all_senders_dropped_and_drained() { - let (tx, mut rx) = bounded::(cap(4)); + let (tx, mut rx) = Mailbox::::bounded(cap(4)); tx.send(Signal::Message(1)).await.expect("queued"); drop(tx); - // Queued message drains first, then the closed-and-empty channel ends. + // Queued message drains first, then the disconnected channel ends. assert!(matches!(rx.recv().await, Some(Signal::Message(1)))); assert!(rx.recv().await.is_none()); } #[tokio::test] async fn full_mailbox_rejects_try_send_and_returns_the_message() { - let (tx, mut rx) = bounded::(cap(1)); + let (tx, mut rx) = Mailbox::::bounded(cap(1)); tx.try_send(Signal::Message(1)).expect("first signal fits"); @@ -605,13 +610,62 @@ mod tests { tx.try_send(Signal::Message(3)).expect("fits after drain"); } + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn n_senders_one_receiver_preserve_per_sender_order() { + const SENDERS: u32 = 8; + const PER_SENDER: u32 = 64; + + // Small capacity so senders genuinely contend and backpressure. + let (tx, mut rx) = Mailbox::::bounded(cap(4)); + let start = Arc::new(Barrier::new(SENDERS as usize)); + + let mut handles = Vec::with_capacity(SENDERS as usize); + for sender_id in 0..SENDERS { + let tx = tx.clone(); + let start = Arc::clone(&start); + handles.push(tokio::spawn(async move { + start.wait().await; // all senders race from the same instant + for seq in 0..PER_SENDER { + tx.send(Signal::Message((sender_id, seq))) + .await + .expect("send"); + } + })); + } + drop(tx); // recv ends only once every sender has dropped its clone + + let mut next_expected = vec![0u32; SENDERS as usize]; + let mut total = 0u32; + while let Some(signal) = rx.recv().await { + let Signal::Message((sender_id, seq)) = signal else { + panic!("unexpected non-message signal"); + }; + let slot = &mut next_expected[sender_id as usize]; + assert_eq!( + seq, *slot, + "FIFO-per-sender violated for sender {sender_id}" + ); + *slot += 1; + total += 1; + } + + assert_eq!(total, SENDERS * PER_SENDER, "lost or duplicated messages"); + for (sender_id, &count) in next_expected.iter().enumerate() { + assert_eq!(count, PER_SENDER, "sender {sender_id} did not fully arrive"); + } + for handle in handles { + handle.await.expect("sender task panicked"); + } + } + proptest! { - /// `Capacity::new` accepts a value iff it is within `MAX`, and preserves - /// it. The strategy pins the interesting boundaries: `1`, `MAX-1`, `MAX`, - /// `MAX+1`, `usize::MAX`. + /// `Capacity::try_from` accepts a value iff it is in `1..=MAX`, and + /// preserves it. The strategy pins the boundaries: `0`, `1`, `MAX-1`, + /// `MAX`, `MAX+1`, `usize::MAX`. #[test] - fn prop_capacity_accepts_iff_within_max( + fn prop_capacity_accepts_iff_in_range( n in prop_oneof![ + Just(0usize), 1usize..=4096, Just(Capacity::MAX - 1), Just(Capacity::MAX), @@ -619,11 +673,9 @@ mod tests { Just(usize::MAX), ], ) { - let value = NonZeroUsize::new(n).expect("strategy yields n >= 1"); - let capacity = Capacity::new(value); - - prop_assert_eq!(capacity.is_some(), n <= Capacity::MAX); - if let Some(capacity) = capacity { + let capacity = Capacity::try_from(n); + prop_assert_eq!(capacity.is_ok(), (1..=Capacity::MAX).contains(&n)); + if let Ok(capacity) = capacity { prop_assert_eq!(capacity.get(), n); } } @@ -641,7 +693,7 @@ mod tests { .build() .expect("current-thread runtime") .block_on(async move { - let (tx, mut rx) = bounded::(cap(capacity)); + let (tx, mut rx) = Mailbox::::bounded(cap(capacity)); let expected = messages.len(); let producer = tokio::spawn(async move { for message in messages { diff --git a/docs/testing/coverage-baseline.md b/docs/testing/coverage-baseline.md index 83c8bb5..2817b69 100644 --- a/docs/testing/coverage-baseline.md +++ b/docs/testing/coverage-baseline.md @@ -37,13 +37,19 @@ nix build .#mutants -L # cargo-mutants on bombay-core; fails if any mutant sur Pinned via the flake's `nixpkgs` (never `nix run nixpkgs#…`), mirroring the coverage package. On-demand, not a per-push gate (rebuilds+tests once per mutant). -### `mailbox` (#112) — done -Zero-box `Signal` queue over `tokio::sync::mpsc`. 16 tests: round-trip, backpressure -hand-back, `Capacity` boundaries (proptest incl. `MAX±1`/`usize::MAX`), lifecycle -(close-drains / send-after-drop / recv-none), `LinkDied` boxed-slot `size_of` guard, -weak death-watch, a real 8-thread `Barrier` linearizability test, and a single-sender -FIFO proptest. **Mutation: 0 missed** (6 caught, 30 unviable, 4 detected-via-timeout). -Criterion bench (`benches/mailbox.rs`): `tell` enqueue ≈ 5.6 ns, send+recv ≈ 21.4 ns. +### `mailbox` (#112, redesigned #133) — done +Zero-box `Signal` queue behind a **`flume`** channel (chosen on measured evidence — +ADR-0001; `flume` is isolated inside the sender/receiver wrappers = the seam). +Construction hangs off the `Mailbox::::bounded(cap)` namespace (composable, no +free-floating `bounded()`). Pure transport: `send`/`try_send`/`recv`/`downgrade`/`drain` +— **no `close()`**; graceful shutdown is `Signal::Stop` + `drain` at the run-loop (#116). +18 tests: round-trip, backpressure hand-back, `Capacity` boundaries (proptest incl. +`0`/`MAX±1`/`usize::MAX`) + `CapacityError`, `MAX`-constant + `Display` guards, lifecycle +(send-after-drop / recv-none / drain-flush), `LinkDied` boxed-slot `size_of` guard + +monomorphic worst-case demo, weak death-watch, an 8-thread `Barrier` linearizability +test, and a single-sender FIFO proptest. **Mutation: 0 missed** (`nix build .#mutants`). +Criterion (`benches/mailbox.rs`): `tell` ≈ **4.7 ns**, send+recv ≈ **13 ns** (flume beat +tokio here too). Channel eval bench: `benches/channels.rs`. **DST posture — loom/shuttle deferred to #116/#120.** loom and shuttle can only model-check code compiled against *their* primitives; the real `tokio::sync::mpsc` this From 250ed2e5da157236f9836ad78709b4a55d4ca552 Mon Sep 17 00:00:00 2001 From: Joel DSouza Date: Sat, 4 Jul 2026 01:48:50 +0200 Subject: [PATCH 9/9] bench(core): realistic ~40B command payload for mailbox + channel benches (#133) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit u64 is the zero-box best case, not a typical message. Switch both benches to a realistic ~40-byte command (a few fields) so the by-value copy cost a real Signal slot pays is measured. Decision holds: flume still wins (26.0 M/s uncontended, 16.0 M/s contended 4->1; ~2x tokio, which stays slowest async). Ordering is payload-robust — u64 gave the same ends, only the middle two reordered. Mailbox: tell ~5.7 ns, roundtrip ~18.4 ns (still ~40% faster than the tokio v1). ADR-0001 + coverage doc updated. --- bombay-core/benches/channels.rs | 51 ++++++++++++++++------ bombay-core/benches/mailbox.rs | 41 +++++++++++++---- docs/adr/0001-mailbox-channel-primitive.md | 42 ++++++++++-------- docs/testing/coverage-baseline.md | 5 ++- 4 files changed, 97 insertions(+), 42 deletions(-) diff --git a/bombay-core/benches/channels.rs b/bombay-core/benches/channels.rs index 50f00ca..d5e5eca 100644 --- a/bombay-core/benches/channels.rs +++ b/bombay-core/benches/channels.rs @@ -6,6 +6,10 @@ //! (not a viable mailbox — it has no async `recv`; shown to price the cost of //! async integration). //! +//! Payload is a realistically-sized command (~40 B, `Default + Clone` so the +//! preallocated-ring candidates can hold it), not a bare `u64` — so the by-value +//! copy cost a real `Signal` slot pays is included. +//! //! Same workload for every channel: `PRODUCERS` tasks each send `PER` messages //! into a `CAP`-capacity channel; one consumer drains until all senders drop. //! All async candidates run on one shared tokio multi-thread runtime, so the @@ -18,6 +22,27 @@ use criterion::{Criterion, Throughput, black_box, criterion_group, criterion_mai const CAP: usize = 256; const PER: u64 = 4_000; +/// A realistically-sized actor command (~40 bytes). `Default + Clone` for the +/// preallocated-ring channels (thingbuf). +#[derive(Clone, Copy, Default)] +struct Command { + id: u64, + correlation: u64, + kind: u32, + amount: i64, + flags: u64, +} + +fn command(i: u64) -> Command { + Command { + id: i, + correlation: i ^ 0x5555_5555, + kind: (i & 0xff) as u32, + amount: i as i64, + flags: i.rotate_left(7), + } +} + fn runtime() -> tokio::runtime::Runtime { tokio::runtime::Builder::new_multi_thread() .worker_threads(4) @@ -26,12 +51,12 @@ fn runtime() -> tokio::runtime::Runtime { } async fn tokio_run(producers: u64) { - let (tx, mut rx) = tokio::sync::mpsc::channel::(CAP); + let (tx, mut rx) = tokio::sync::mpsc::channel::(CAP); for _ in 0..producers { let tx = tx.clone(); tokio::spawn(async move { for i in 0..PER { - tx.send(i).await.expect("send"); + tx.send(command(i)).await.expect("send"); } }); } @@ -44,12 +69,12 @@ async fn tokio_run(producers: u64) { } async fn flume_run(producers: u64) { - let (tx, rx) = flume::bounded::(CAP); + let (tx, rx) = flume::bounded::(CAP); for _ in 0..producers { let tx = tx.clone(); tokio::spawn(async move { for i in 0..PER { - tx.send_async(i).await.expect("send"); + tx.send_async(command(i)).await.expect("send"); } }); } @@ -62,12 +87,12 @@ async fn flume_run(producers: u64) { } async fn async_channel_run(producers: u64) { - let (tx, rx) = async_channel::bounded::(CAP); + let (tx, rx) = async_channel::bounded::(CAP); for _ in 0..producers { let tx = tx.clone(); tokio::spawn(async move { for i in 0..PER { - tx.send(i).await.expect("send"); + tx.send(command(i)).await.expect("send"); } }); } @@ -80,12 +105,12 @@ async fn async_channel_run(producers: u64) { } async fn thingbuf_run(producers: u64) { - let (tx, rx) = thingbuf::mpsc::channel::(CAP); + let (tx, rx) = thingbuf::mpsc::channel::(CAP); for _ in 0..producers { let tx = tx.clone(); tokio::spawn(async move { for i in 0..PER { - tx.send(i).await.expect("send"); + tx.send(command(i)).await.expect("send"); } }); } @@ -100,13 +125,13 @@ async fn thingbuf_run(producers: u64) { /// Sync ceiling — std threads, not tasks. Not a viable async mailbox; here only /// to show the raw lock-free throughput async has to pay to integrate with. fn crossbeam_run(producers: u64) { - let (tx, rx) = crossbeam_channel::bounded::(CAP); + let (tx, rx) = crossbeam_channel::bounded::(CAP); thread::scope(|scope| { for _ in 0..producers { let tx = tx.clone(); scope.spawn(move || { for i in 0..PER { - tx.send(i).expect("send"); + tx.send(command(i)).expect("send"); } }); } @@ -125,17 +150,17 @@ fn bench(c: &mut Criterion, producers: u64, label: &str) { group.throughput(Throughput::Elements(producers * PER)); group.bench_function("tokio_mpsc", |b| { - b.iter(|| rt.block_on(tokio_run(producers))) + b.iter(|| rt.block_on(tokio_run(producers))); }); group.bench_function("flume", |b| b.iter(|| rt.block_on(flume_run(producers)))); group.bench_function("async_channel", |b| { b.iter(|| rt.block_on(async_channel_run(producers))); }); group.bench_function("thingbuf", |b| { - b.iter(|| rt.block_on(thingbuf_run(producers))) + b.iter(|| rt.block_on(thingbuf_run(producers))); }); group.bench_function("crossbeam_sync_ceiling", |b| { - b.iter(|| crossbeam_run(producers)) + b.iter(|| crossbeam_run(producers)); }); group.finish(); diff --git a/bombay-core/benches/mailbox.rs b/bombay-core/benches/mailbox.rs index 19eef7d..1a926ca 100644 --- a/bombay-core/benches/mailbox.rs +++ b/bombay-core/benches/mailbox.rs @@ -2,34 +2,57 @@ //! //! The card's premise — a zero-box `tell` is cheaper than kameo's //! `Box` enqueue — is un-templated: no framework ships this -//! shape, so we *measure* rather than assume (#112). These numbers are the -//! baseline the later two-tier wiring (#114/#118) is compared against. +//! shape, so we *measure* rather than assume (#112/#133). +//! +//! Payload is a realistically-sized command (~40 B), not a bare `u64`, so the +//! by-value copy cost that a real `Signal` slot pays is measured honestly. use std::num::NonZeroUsize; use bombay_core::mailbox::{Capacity, Mailbox, Mailboxed, Signal}; use criterion::{BatchSize, Criterion, black_box, criterion_group, criterion_main}; +/// A realistically-sized actor command (~40 bytes) — a handful of fields, closer +/// to a real closed-enum `Msg` variant than a bare `u64`. +#[derive(Clone, Copy, Default)] +struct Command { + id: u64, + correlation: u64, + kind: u32, + amount: i64, + flags: u64, +} + struct Bench; impl Mailboxed for Bench { - type Msg = u64; + type Msg = Command; +} + +fn command(i: u64) -> Command { + Command { + id: i, + correlation: i ^ 0x5555_5555, + kind: (i & 0xff) as u32, + amount: i as i64, + flags: i.rotate_left(7), + } } fn cap(n: usize) -> Capacity { Capacity::new(NonZeroUsize::new(n).expect("nonzero")).expect("within max") } -/// Pure enqueue cost: how long to `try_send` 1_000 messages into a mailbox with +/// Pure enqueue cost: how long to `try_send` 1_000 commands into a mailbox with /// spare capacity. `iter_batched_ref` keeps the `bounded()` setup out of the /// measured region, and the fresh mailbox per batch never fills, so this isolates /// the move-into-slot cost of a `tell`. fn enqueue(c: &mut Criterion) { - c.bench_function("tell_try_send_1k_u64", |b| { + c.bench_function("tell_try_send_1k_command", |b| { b.iter_batched_ref( || Mailbox::::bounded(cap(1024)), |(tx, _rx)| { for i in 0..1000u64 { - tx.try_send(Signal::Message(black_box(i))) + tx.try_send(Signal::Message(black_box(command(i)))) .expect("capacity available"); } }, @@ -45,13 +68,15 @@ fn roundtrip(c: &mut Criterion) { .build() .expect("current-thread runtime"); - c.bench_function("send_recv_roundtrip_1k_u64", |b| { + c.bench_function("send_recv_roundtrip_1k_command", |b| { b.iter(|| { rt.block_on(async { let (tx, mut rx) = Mailbox::::bounded(cap(1024)); let producer = tokio::spawn(async move { for i in 0..1000u64 { - tx.send(Signal::Message(black_box(i))).await.expect("send"); + tx.send(Signal::Message(black_box(command(i)))) + .await + .expect("send"); } }); diff --git a/docs/adr/0001-mailbox-channel-primitive.md b/docs/adr/0001-mailbox-channel-primitive.md index 000e636..6afd3ae 100644 --- a/docs/adr/0001-mailbox-channel-primitive.md +++ b/docs/adr/0001-mailbox-channel-primitive.md @@ -60,30 +60,34 @@ gate the mailbox, and record the decision here. ## Evidence (measured — `cargo bench --bench channels`, aarch64, 4 workers) -Throughput (higher = better); one shared tokio multi-thread runtime, `u64` -payload, `CAP = 256`, warm-up 1 s / measure 2 s: +Throughput (higher = better); one shared tokio multi-thread runtime, +**realistic ~40 B `Command` payload** (a few fields — closer to a real `Signal` +slot than a bare `u64`), `CAP = 256`, full-length runs: | Channel | Uncontended (1→1) | Contended (4→1) | |---|---|---| -| crossbeam *(sync ceiling, not a mailbox)* | 53.2 M/s | — | -| **flume** | **25.7 M/s** | **14.8 M/s** | -| async-channel | 18.1 M/s | 11.1 M/s | -| **tokio::mpsc** *(v1 default)* | 13.4 M/s | 4.9 M/s | -| thingbuf | 8.6 M/s | ~4.1 M/s | +| crossbeam *(sync ceiling, not a mailbox)* | 36.7 M/s | 20.3 M/s | +| **flume** | **26.0 M/s** | **16.0 M/s** | +| async-channel | 22.2 M/s | 13.0 M/s | +| thingbuf | 14.1 M/s | 8.3 M/s | +| **tokio::mpsc** *(v1 default)* | 13.1 M/s | 5.8 M/s | Findings: -- **tokio::mpsc — the un-surveyed v1 default — is second-slowest, ~3× slower - than flume under contention.** The concrete cost of skipping #112's survey. -- **flume is the throughput winner**, executor-agnostic, and **move-based** (no - extra trait bounds on the element). -- **thingbuf is the slowest *and* requires `T: Default`** (a preallocated ring - initialises its slots). `Signal` has no sensible `Default` (a message enum - cannot default) — so thingbuf's `no_std`/loom edge comes with a real fit - problem against our by-value design. - -Caveat: this measures *raw channel* throughput with a small payload; real actor -workloads are usually handler/IO-bound, so the absolute gap matters less than -the *ordering* and the qualitative axes. +- **flume is the throughput winner** — ~2× tokio uncontended, ~2.7× contended. + Executor-agnostic and **move-based** (no extra trait bounds on the element). +- **tokio::mpsc — the un-surveyed v1 default — is the slowest async candidate.** + The concrete cost of skipping #112's survey. +- **The pick is payload-robust:** a `u64` (16 B) run gave the *same* ordering at + the ends (flume fastest, tokio slowest async); only the middle two reordered + (thingbuf/async-channel improved as the larger copy amortised differently). So + the decision is not an artifact of a toy payload. +- **thingbuf requires `T: Default`** (a preallocated ring initialises its slots). + `Signal` has no sensible `Default` (a message enum cannot default) — so its + `no_std`/loom edge comes with a real fit problem against our by-value design. + +Caveat: this is *raw channel* throughput; real actor workloads are usually +handler/IO-bound, so the absolute gap matters less than the *ordering* and the +qualitative axes. ## Decision diff --git a/docs/testing/coverage-baseline.md b/docs/testing/coverage-baseline.md index 2817b69..5b4758f 100644 --- a/docs/testing/coverage-baseline.md +++ b/docs/testing/coverage-baseline.md @@ -48,8 +48,9 @@ free-floating `bounded()`). Pure transport: `send`/`try_send`/`recv`/`downgrade` (send-after-drop / recv-none / drain-flush), `LinkDied` boxed-slot `size_of` guard + monomorphic worst-case demo, weak death-watch, an 8-thread `Barrier` linearizability test, and a single-sender FIFO proptest. **Mutation: 0 missed** (`nix build .#mutants`). -Criterion (`benches/mailbox.rs`): `tell` ≈ **4.7 ns**, send+recv ≈ **13 ns** (flume beat -tokio here too). Channel eval bench: `benches/channels.rs`. +Criterion (`benches/mailbox.rs`, realistic ~40 B command): `tell` ≈ **5.7 ns**, send+recv +≈ **18.4 ns** (~40 % faster than the tokio v1 on the same bench). Channel eval: +`benches/channels.rs` (ADR-0001) — flume wins at both `u64` and `~40 B` payloads. **DST posture — loom/shuttle deferred to #116/#120.** loom and shuttle can only model-check code compiled against *their* primitives; the real `tokio::sync::mpsc` this