Stage 12: round-trip allocation gate and Embassy outage semantics (design 054 §5, §8)

Makefile

@@ -249,6 +249,8 @@ test:
 	cargo test --package aimdb-mqtt-connector --no-default-features --features "_test-tokio-broker" --test tokio_broker
 	@printf "$(YELLOW)  → Testing MQTT connector (both backends, one broker, one process)$(NC)\n"
 	cargo test --package aimdb-mqtt-connector --no-default-features --features "_test-backend-parity" --test backend_parity
+	@printf "$(YELLOW)  → Testing MQTT connector (allocations per round trip, both backends)$(NC)\n"
+	cargo test --package aimdb-mqtt-connector --no-default-features --features "_test-backend-parity" --test alloc_round_trip
 	@printf "$(YELLOW)  → Testing MQTT connector (mqtts:// against a pinned self-signed root)$(NC)\n"
 	cargo test --package aimdb-mqtt-connector --no-default-features --features "_test-tls-broker" --test tls_broker
 	@printf "$(YELLOW)  → Testing MQTT connector (event-driven session: wake cadence, partial packets, QoS 1)$(NC)\n"
@@ -400,6 +402,8 @@ clippy:
 	cargo clippy --package aimdb-mqtt-connector --no-default-features --features "_test-tokio-broker" --test tokio_broker -- -D warnings
 	@printf "$(YELLOW)  → Clippy on MQTT connector (backend parity)$(NC)\n"
 	cargo clippy --package aimdb-mqtt-connector --no-default-features --features "_test-backend-parity" --test backend_parity -- -D warnings
+	@printf "$(YELLOW)  → Clippy on MQTT connector (allocations per round trip)$(NC)\n"
+	cargo clippy --package aimdb-mqtt-connector --no-default-features --features "_test-backend-parity" --test alloc_round_trip -- -D warnings
 	@printf "$(YELLOW)  → Clippy on MQTT connector (mqtts:// host smoke)$(NC)\n"
 	cargo clippy --package aimdb-mqtt-connector --no-default-features --features "_test-tls-broker" --test tls_broker -- -D warnings
 	@printf "$(YELLOW)  → Clippy on MQTT connector (event-driven session criteria)$(NC)\n"

aimdb-embassy-adapter/tests/outbound_routes.rs

@@ -0,0 +1,104 @@
+//! What a connector pulling from `OutboundRoutes` sees after an outage, on the
+//! Embassy buffers: the same cases as the Tokio adapter's
+//! `outage_semantics_per_buffer_type`, driven on the host with a no-op waker.
+#![cfg(all(feature = "embassy-sync", feature = "embassy-time"))]
+
+use std::future::Future;
+use std::pin::Pin;
+use std::sync::Arc;
+use std::task::{Context, Poll, Waker};
+
+use aimdb_core::buffer::DynBuffer;
+use aimdb_core::connector::ConnectorBuilder;
+use aimdb_core::executor::test_support::NoopRuntimeOps;
+use aimdb_core::{AimDb, AimDbBuilder, DbResult, OutboundRoutes};
+use aimdb_embassy_adapter::EmbassyBuffer;
+use futures::executor::block_on;
+
+// No-op defmt logger + host time driver, so the binary links.
+aimdb_embassy_adapter::host_test_stubs!();
+
+type Futures = Vec<Pin<Box<dyn Future<Output = ()> + Send + 'static>>>;
+
+/// Lets `link_to("test://…")` register; drives nothing.
+struct TestConnector;
+
+impl ConnectorBuilder for TestConnector {
+    fn build<'a>(
+        &'a self,
+        _db: &'a AimDb,
+    ) -> Pin<Box<dyn Future<Output = DbResult<Futures>> + Send + 'a>> {
+        Box::pin(async { Ok(Vec::new()) })
+    }
+    fn scheme(&self) -> &str {
+        "test"
+    }
+}
+
+const KEYS: [&str; 3] = ["r0", "r1", "r2"];
+
+/// One record per buffer, keyed `r0`, `r1`, …, each linked to `test://r{i}`.
+fn db(buffers: Vec<Box<dyn DynBuffer<u32>>>) -> AimDb {
+    let mut builder = AimDbBuilder::new()
+        .runtime(Arc::new(NoopRuntimeOps))
+        .with_connector(TestConnector);
+    for (i, buffer) in buffers.into_iter().enumerate() {
+        builder.configure::<u32>(KEYS[i], move |reg| {
+            reg.buffer_raw(buffer)
+                .link_to(&format!("test://r{i}"))
+                .with_serializer(|_ctx, v: &u32| Ok(v.to_le_bytes().to_vec()))
+                .finish();
+        });
+    }
+    block_on(builder.build()).expect("build").0
+}
+
+/// Every message available now, as (route, value).
+fn drain(o: &mut OutboundRoutes) -> Vec<(usize, u32)> {
+    let mut cx = Context::from_waker(Waker::noop());
+    let mut out = Vec::new();
+    while let Poll::Ready(Some(m)) = o.poll_next(&mut cx) {
+        out.push((
+            m.route.id,
+            u32::from_le_bytes(m.payload.as_slice().try_into().unwrap()),
+        ));
+    }
+    out
+}
+
+fn values(got: &[(usize, u32)], route: usize) -> Vec<u32> {
+    got.iter()
+        .filter(|(id, _)| *id == route)
+        .map(|(_, v)| *v)
+        .collect()
+}
+
+#[test]
+fn outage_semantics_per_buffer_type() {
+    let db = db(vec![
+        Box::new(EmbassyBuffer::<u32, 16, 4, 4, 4>::new_watch()),
+        Box::new(EmbassyBuffer::<u32, 16, 4, 4, 4>::new_mailbox()),
+        Box::new(EmbassyBuffer::<u32, 16, 4, 4, 4>::new_spmc()),
+    ]);
+    let mut o = OutboundRoutes::new(&db, "test").unwrap();
+    for v in 0..5 {
+        for key in KEYS {
+            db.produce::<u32>(key, v).unwrap();
+        }
+    }
+    let got = drain(&mut o);
+    assert_eq!(values(&got, 0), [4], "single-latest");
+    assert_eq!(values(&got, 1), [4], "mailbox");
+    assert_eq!(values(&got, 2), [0, 1, 2, 3, 4], "spmc ring");
+}
+
+#[test]
+fn an_spmc_ring_that_overflows_reports_lag_then_recovers() {
+    let db = db(vec![Box::new(EmbassyBuffer::<u32, 4, 4, 4, 4>::new_spmc())]);
+    let mut o = OutboundRoutes::new(&db, "test").unwrap();
+    for v in 0..10 {
+        db.produce::<u32>("r0", v).unwrap();
+    }
+    assert_eq!(values(&drain(&mut o), 0), [6, 7, 8, 9]);
+    assert_eq!(o.stats(0).unwrap().lagged, 6);
+}

aimdb-mqtt-connector/tests/alloc_round_trip.rs

@@ -0,0 +1,264 @@
+//! Allocations per MQTT round trip, per backend (`_test-backend-parity`).
+//!
+//! One round trip: produce → PUBLISH QoS 1 → the broker's PUBACK and echo →
+//! inbound dispatch → the client's PUBACK → reader `recv`. The database, its
+//! connector and the produce/recv loop run on one thread with a current-thread
+//! runtime; a counting allocator counts only on that thread, so the broker on
+//! another thread is not measured.
+//!
+//! The embedded backend's one remaining copy per message is topic and payload
+//! from `OutboundRoutes`' scratch into the encoded frame; it allocates nothing.
+//! The native backend's count is `rumqttc`'s: `AsyncClient::publish` takes an
+//! owned topic and payload, and builds its own request.
+#![cfg(feature = "_test-backend-parity")]
+
+use std::alloc::{GlobalAlloc, Layout, System};
+use std::cell::Cell;
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+use tokio::net::TcpListener;
+
+use aimdb_core::buffer::BufferCfg;
+use aimdb_core::connector::{ConnectorBuilder, SerializeError};
+use aimdb_core::AimDbBuilder;
+use aimdb_mqtt_connector::MqttConnector;
+use aimdb_tokio_adapter::net::TokioNet;
+use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt};
+
+mod common;
+use common::echo_broker;
+
+// Each test binary defines these exactly once.
+#[defmt::global_logger]
+struct HostTestLogger;
+unsafe impl defmt::Logger for HostTestLogger {
+    fn acquire() {}
+    unsafe fn flush() {}
+    unsafe fn release() {}
+    unsafe fn write(_bytes: &[u8]) {}
+}
+#[defmt::panic_handler]
+fn defmt_panic() -> ! {
+    core::panic!("defmt panic in host test")
+}
+defmt::timestamp!("{=u64:us}", 0);
+
+struct HostClock;
+impl embassy_time_driver::Driver for HostClock {
+    fn now(&self) -> u64 {
+        use std::sync::OnceLock;
+        static START: OnceLock<Instant> = OnceLock::new();
+        let start = START.get_or_init(Instant::now);
+        (start.elapsed().as_micros() * u128::from(embassy_time_driver::TICK_HZ) / 1_000_000) as u64
+    }
+    fn schedule_wake(&self, _at: u64, waker: &core::task::Waker) {
+        waker.wake_by_ref();
+    }
+}
+embassy_time_driver::time_driver_impl!(static HOST_CLOCK: HostClock = HostClock);
+
+// ---------------------------------------------------------------------------
+// Counting allocator: per-thread counters, so tests may run in parallel.
+// ---------------------------------------------------------------------------
+
+struct Counting;
+
+thread_local! {
+    /// Set on the database thread; nothing else is counted.
+    static COUNT_HERE: Cell<bool> = const { Cell::new(false) };
+    /// Set only during the measured round trips.
+    static WINDOW: Cell<bool> = const { Cell::new(false) };
+    static ALLOCS: Cell<usize> = const { Cell::new(0) };
+    static BYTES: Cell<usize> = const { Cell::new(0) };
+    /// Bytes this thread allocated and has not freed.
+    static LIVE: Cell<usize> = const { Cell::new(0) };
+}
+
+fn add(cell: &'static std::thread::LocalKey<Cell<usize>>, n: usize) {
+    cell.with(|c| c.set(c.get() + n));
+}
+
+unsafe impl GlobalAlloc for Counting {
+    unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
+        if COUNT_HERE.with(Cell::get) {
+            add(&LIVE, layout.size());
+            if WINDOW.with(Cell::get) {
+                add(&ALLOCS, 1);
+                add(&BYTES, layout.size());
+            }
+        }
+        unsafe { System.alloc(layout) }
+    }
+    unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
+        if COUNT_HERE.with(Cell::get) {
+            LIVE.with(|c| c.set(c.get().saturating_sub(layout.size())));
+        }
+        unsafe { System.dealloc(ptr, layout) }
+    }
+}
+
+#[global_allocator]
+static GLOBAL: Counting = Counting;
+
+// ---------------------------------------------------------------------------
+// The round trip.
+// ---------------------------------------------------------------------------
+
+const WARMUP: u64 = 100;
+const MEASURED: u64 = 300;
+const TOPIC: &str = "mqtt://rt/ping";
+
+struct Report {
+    allocs: usize,
+    bytes: usize,
+    live_before: usize,
+    median: Duration,
+    min: Duration,
+    max: Duration,
+}
+
+/// Round trips through `connector` on a fresh thread; `COUNT_HERE` is on for
+/// that thread only.
+fn measure(connector: impl ConnectorBuilder + 'static) -> Report {
+    std::thread::spawn(move || {
+        COUNT_HERE.with(|c| c.set(true));
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("runtime");
+        runtime.block_on(round_trips(connector))
+    })
+    .join()
+    .expect("database thread")
+}
+
+async fn round_trips(connector: impl ConnectorBuilder + 'static) -> Report {
+    let mut builder = AimDbBuilder::new()
+        .runtime(Arc::new(TokioAdapter))
+        .with_connector(connector);
+    builder.configure::<u64>("ping", |reg| {
+        reg.buffer(BufferCfg::SingleLatest)
+            .link_to(TOPIC)
+            .with_serializer(|_ctx, v: &u64| Ok(v.to_le_bytes().to_vec()))
+            .with_serializer_into(8, |_ctx, v: &u64, out| {
+                out.get_mut(..8)
+                    .ok_or(SerializeError::BufferTooSmall)?
+                    .copy_from_slice(&v.to_le_bytes());
+                Ok(8)
+            })
+            .finish();
+    });
+    builder.configure::<u64>("pong", |reg| {
+        reg.buffer(BufferCfg::SingleLatest)
+            .link_from(TOPIC)
+            .with_deserializer(|_ctx, data: &[u8]| {
+                data.try_into()
+                    .map(u64::from_le_bytes)
+                    .map_err(|_| String::from("not 8 bytes"))
+            })
+            .finish();
+    });
+    let (db, runner) = builder.build().await.expect("build db");
+    tokio::spawn(runner.run());
+    let producer = db.producer::<u64>("ping").expect("producer");
+    let mut pong = db.subscribe::<u64>("pong").expect("subscribe");
+
+    // Warm-up also waits out the connect and subscribe.
+    tokio::time::timeout(Duration::from_secs(60), async {
+        for n in 0..WARMUP {
+            round_trip(&producer, &mut pong, n).await;
+        }
+    })
+    .await
+    .expect("warm-up round trips");
+
+    let mut latencies = Vec::with_capacity(MEASURED as usize);
+    let live_before = LIVE.with(Cell::get);
+    ALLOCS.with(|c| c.set(0));
+    BYTES.with(|c| c.set(0));
+    WINDOW.with(|w| w.set(true));
+    for n in WARMUP..WARMUP + MEASURED {
+        let start = Instant::now();
+        round_trip(&producer, &mut pong, n).await;
+        latencies.push(start.elapsed());
+    }
+    WINDOW.with(|w| w.set(false));
+
+    latencies.sort();
+    Report {
+        allocs: ALLOCS.with(Cell::get),
+        bytes: BYTES.with(Cell::get),
+        live_before,
+        median: latencies[latencies.len() / 2],
+        min: latencies[0],
+        max: latencies[latencies.len() - 1],
+    }
+}
+
+/// Produce `n` and wait until it comes back through the broker.
+async fn round_trip(
+    producer: &aimdb_core::Producer<u64>,
+    pong: &mut aimdb_core::buffer::Reader<u64>,
+    n: u64,
+) {
+    producer.produce(n);
+    while pong.recv().await.expect("pong open") != n {}
+}
+
+/// An echo broker on its own thread; returns its port.
+fn broker() -> u16 {
+    let (tx, rx) = std::sync::mpsc::channel();
+    std::thread::spawn(move || {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("runtime");
+        runtime.block_on(async move {
+            let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+            tx.send(listener.local_addr().unwrap().port()).unwrap();
+            echo_broker(listener).await;
+        });
+    });
+    rx.recv().expect("broker port")
+}
+
+fn print(backend: &str, r: &Report) {
+    println!(
+        "{backend:<8} {:>6.2} allocs/round trip {:>8.1} bytes/round trip  live heap before: {} B  latency median {:?} ({:?}–{:?})",
+        r.allocs as f64 / MEASURED as f64,
+        r.bytes as f64 / MEASURED as f64,
+        r.live_before,
+        r.median,
+        r.min,
+        r.max
+    );
+}
+
+/// The embedded backend allocates nothing per round trip.
+#[test]
+fn the_embedded_backend_allocates_nothing_per_round_trip() {
+    let port = broker();
+    let report =
+        measure(MqttConnector::new(format!("mqtt://127.0.0.1:{port}")).transport(TokioNet::tcp()));
+    print("embedded", &report);
+    assert_eq!(
+        report.allocs, 0,
+        "{} allocations in {MEASURED} round trips",
+        report.allocs
+    );
+}
+
+/// The native backend's allocations are `rumqttc`'s; the bound is what the
+/// prototype measured.
+#[test]
+fn the_native_backend_stays_within_rumqttcs_allocations() {
+    let port = broker();
+    let report = measure(MqttConnector::new(format!("mqtt://127.0.0.1:{port}")));
+    print("native", &report);
+    let per_round_trip = report.allocs as f64 / MEASURED as f64;
+    assert!(
+        per_round_trip <= 11.0,
+        "{per_round_trip:.2} allocations per round trip"
+    );
+}