diff --git a/Cargo.lock b/Cargo.lock index 1da11c0..c8ed3f0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -470,6 +470,15 @@ dependencies = [ "tracing-subscriber", ] +[[package]] +name = "bombay-core" +version = "0.1.0" +dependencies = [ + "criterion", + "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..b61e88b --- /dev/null +++ b/bombay-core/Cargo.toml @@ -0,0 +1,30 @@ +[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 } +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/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..92d6bd2 --- /dev/null +++ b/bombay-core/src/mailbox.rs @@ -0,0 +1,666 @@ +//! 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; +} + +/// 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 +/// 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, + /// 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 +/// while any sender is alive. +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>, +} + +/// 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), + }) + } + + /// 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 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. + struct Probe; + impl Mailboxed for Probe { + 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 { + 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. + } + + #[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::>() + ); + } + + /// 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)); + + 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 + // `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)); + 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)); + + 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 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)); + + 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"); + } + + 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); + } + } +} 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 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; }