diff --git a/Cargo.lock b/Cargo.lock index 1da11c0..9af86a5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -470,6 +470,19 @@ dependencies = [ "tracing-subscriber", ] +[[package]] +name = "bombay-core" +version = "0.1.0" +dependencies = [ + "async-channel", + "criterion", + "crossbeam-channel", + "flume", + "proptest", + "thingbuf", + "tokio", +] + [[package]] name = "bombay_actors" version = "0.5.0" @@ -1479,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" @@ -1515,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" @@ -4842,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" @@ -5111,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/Cargo.toml b/Cargo.toml index 6881def..8d5a82e 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" @@ -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 new file mode 100644 index 0000000..1dff3b2 --- /dev/null +++ b/bombay-core/Cargo.toml @@ -0,0 +1,38 @@ +[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"] } +flume = { workspace = true } + +[dev-dependencies] +tokio = { workspace = true, features = [ + "rt", + "rt-multi-thread", + "macros", + "sync", + "time", +] } +proptest = { workspace = true } +criterion = { version = "0.8", features = ["async_tokio"] } +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..d5e5eca --- /dev/null +++ b/bombay-core/benches/channels.rs @@ -0,0 +1,178 @@ +//! 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). +//! +//! 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 +//! 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; + +/// 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) + .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(command(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(command(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(command(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(command(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(command(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/bombay-core/benches/mailbox.rs b/bombay-core/benches/mailbox.rs new file mode 100644 index 0000000..1a926ca --- /dev/null +++ b/bombay-core/benches/mailbox.rs @@ -0,0 +1,98 @@ +//! 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/#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 = 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 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_command", |b| { + b.iter_batched_ref( + || Mailbox::::bounded(cap(1024)), + |(tx, _rx)| { + for i in 0..1000u64 { + tx.try_send(Signal::Message(black_box(command(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_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(command(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..52767c7 --- /dev/null +++ b/bombay-core/src/mailbox.rs @@ -0,0 +1,718 @@ +//! 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. +//! +//! 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 std::{fmt, marker::PhantomData, num::NonZeroUsize}; + +/// A validated mailbox capacity: at least `1`, at most [`Capacity::MAX`]. +/// +/// 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 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`]. + #[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() + } +} + +/// 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. +/// +/// `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 (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 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), +} + +/// 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: flume::Sender>, +} + +impl Clone for MailboxSender { + fn clone(&self) -> Self { + Self { + tx: self.tx.clone(), + } + } +} + +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: flume::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: 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. +/// +/// 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)"), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + use std::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); + } + + /// 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::try_from(n).expect("test capacity must be valid") + } + + #[tokio::test] + async fn sent_message_is_received() { + let (tx, mut rx) = Mailbox::::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 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"); + + assert!(matches!(rx.recv().await, Some(Signal::Message(7)))); + } + + #[test] + 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); + } + + #[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] + 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. Guards the + // "every slot = largest variant" trap. + 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). + /// + /// 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, + 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::>(); + + assert!(small <= 24, "small slot = {small}"); + assert!(fat >= 4096, "fat inline slot = {fat}"); + assert!(boxed <= 24, "boxed slot = {boxed}"); + + 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}" + ); + } + + #[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) = Mailbox::::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) = Mailbox::::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) = Mailbox::::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 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"); + } + + assert!(matches!(rx.recv().await, Some(Signal::Message(0)))); + + 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) = Mailbox::::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) = Mailbox::::bounded(cap(4)); + tx.send(Signal::Message(1)).await.expect("queued"); + drop(tx); + + // 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) = Mailbox::::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"); + } + + #[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::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_in_range( + n in prop_oneof![ + Just(0usize), + 1usize..=4096, + Just(Capacity::MAX - 1), + Just(Capacity::MAX), + Just(Capacity::MAX + 1), + Just(usize::MAX), + ], + ) { + 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); + } + } + + /// 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) = Mailbox::::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/adr/0001-mailbox-channel-primitive.md b/docs/adr/0001-mailbox-channel-primitive.md new file mode 100644 index 0000000..6afd3ae --- /dev/null +++ b/docs/adr/0001-mailbox-channel-primitive.md @@ -0,0 +1,119 @@ +# 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, +**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)* | 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: +- **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 + +**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 | diff --git a/docs/testing/coverage-baseline.md b/docs/testing/coverage-baseline.md index 50cc533..5b4758f 100644 --- a/docs/testing/coverage-baseline.md +++ b/docs/testing/coverage-baseline.md @@ -24,6 +24,46 @@ 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, 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`, 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 +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; }