core(mailbox): principal-grade redesign — flume (ADR-0001) + composable Mailbox API (#133)

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"

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

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::<Command>(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::<Command>(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::<Command>(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::<Command>(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::<Command>(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);