From 69664b807764498aadf86e84cc731fdd9fec4499 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alexander=20Schn=C3=B6rch?= Date: Sun, 4 Oct 2026 17:40:14 +0000 Subject: [PATCH 1/2] =?UTF-8?q?feat(mqtt):=20encode=20embedded=20packets?= =?UTF-8?q?=20into=20one=20bbqueue=20write=20ring=20(design=20054=20=C2=A7?= =?UTF-8?q?4.7)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 16 + aimdb-mqtt-connector/Cargo.toml | 5 + aimdb-mqtt-connector/src/embedded/mod.rs | 1 + aimdb-mqtt-connector/src/embedded/session.rs | 5 + .../src/embedded/session_loop.rs | 214 ++++++----- aimdb-mqtt-connector/src/embedded/tls.rs | 5 + .../src/embedded/write_ring.rs | 358 ++++++++++++++++++ 7 files changed, 496 insertions(+), 108 deletions(-) create mode 100644 aimdb-mqtt-connector/src/embedded/write_ring.rs diff --git a/Cargo.lock b/Cargo.lock index 265f74df..46f2d412 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -264,6 +264,7 @@ dependencies = [ "aimdb-embassy-adapter", "aimdb-mountain-mqtt", "aimdb-tokio-adapter", + "bbqueue", "critical-section", "defmt 1.1.1", "embassy-futures 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)", @@ -684,6 +685,15 @@ version = "1.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06" +[[package]] +name = "bbqueue" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68917624e17aad88607cb5a5936f6da9b607c48c711e4e9ed101e7189aed28c2" +dependencies = [ + "const-init", +] + [[package]] name = "bit_field" version = "0.10.3" @@ -939,6 +949,12 @@ version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b396d1f76d455557e1218ec8066ae14bba60b4b36ecd55577ba979f5db7ecaa" +[[package]] +name = "const-init" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4bd422bfb4f24a97243f60b6a4443e63d810c925d8da4bb2d8fde26a7c1d57ec" + [[package]] name = "const-oid" version = "0.9.6" diff --git a/aimdb-mqtt-connector/Cargo.toml b/aimdb-mqtt-connector/Cargo.toml index 187c709d..8680be7f 100644 --- a/aimdb-mqtt-connector/Cargo.toml +++ b/aimdb-mqtt-connector/Cargo.toml @@ -49,6 +49,8 @@ embedded = [ # `mountain-mqtt`'s packet types take `heapless::Vec`, so the session names # it to build them. Same 0.8 the dependency itself resolves. "dep:heapless", + # The session's write ring. + "dep:bbqueue", ] # Convenience bundle: `embedded` plus the Embassy transport and clock. The @@ -157,6 +159,9 @@ embassy-time = { version = "0.5.1", optional = true } embassy-sync = { version = "0.8.0", path = "../_external/embassy/embassy-sync", optional = true } embassy-futures = { workspace = true, optional = true } heapless = { workspace = true, optional = true } +# Byte ring the embedded session encodes outbound packets into. `AtomicCoord` +# needs atomic CAS, which every supported embedded target has. +bbqueue = { version = "0.7", default-features = false, features = ["alloc"], optional = true } embassy-net = { version = "0.9.0", optional = true, features = [ "tcp", "dhcpv4", diff --git a/aimdb-mqtt-connector/src/embedded/mod.rs b/aimdb-mqtt-connector/src/embedded/mod.rs index 1201b07d..7fe77910 100644 --- a/aimdb-mqtt-connector/src/embedded/mod.rs +++ b/aimdb-mqtt-connector/src/embedded/mod.rs @@ -13,6 +13,7 @@ pub(crate) mod session; // replace the polled loop. pub(crate) mod packet_reader; pub(crate) mod session_loop; +pub(crate) mod write_ring; // TLS transport + SNTP time source. #[cfg(feature = "embassy-tls")] diff --git a/aimdb-mqtt-connector/src/embedded/session.rs b/aimdb-mqtt-connector/src/embedded/session.rs index 2adc9f63..1d2b1c8d 100644 --- a/aimdb-mqtt-connector/src/embedded/session.rs +++ b/aimdb-mqtt-connector/src/embedded/session.rs @@ -64,6 +64,7 @@ where use mountain_mqtt::data::quality_of_service::QualityOfService; use crate::embedded::session_loop::run_session; + use crate::embedded::write_ring::{WriteRing, DEFAULT_WRITE_BUFFER}; // Built once and borrowed for the loop; re-sent on every connection. let subscribe_topics: alloc::vec::Vec<(&str, QualityOfService)> = topics @@ -71,6 +72,9 @@ where .map(|topic| (topic.as_str(), QualityOfService::Qos1)) .collect(); + // Allocated once and reused by every session. + let ring = WriteRing::new(DEFAULT_WRITE_BUFFER); + loop { let mut stream = match dialer.connect(&host, port).await { Ok(stream) => stream, @@ -92,6 +96,7 @@ where &subscribe_topics, &events, &actions, + &ring, &settings, &dialer, runtime.as_ref(), diff --git a/aimdb-mqtt-connector/src/embedded/session_loop.rs b/aimdb-mqtt-connector/src/embedded/session_loop.rs index ec8975aa..542d227b 100644 --- a/aimdb-mqtt-connector/src/embedded/session_loop.rs +++ b/aimdb-mqtt-connector/src/embedded/session_loop.rs @@ -5,25 +5,25 @@ //! partially-read packet is ever discarded — which is what lets the TLS path //! share this loop. //! -//! [`read_into`] and [`write_out`] know no MQTT; [`client_loop`] owns all -//! client state and wakes only on data, an action or a deadline. Raw chunks -//! cross the inbound channel rather than whole packets, so framing needs no -//! second packet-sized buffer. +//! [`read_into`] and [`WriteRing::write_out`] know no MQTT; [`client_loop`] +//! owns all client state and wakes only on data, an action or a deadline. Raw +//! chunks cross the inbound channel rather than whole packets, so framing +//! needs no second packet-sized buffer. Outbound packets are encoded straight +//! into the connector's [`WriteRing`]. use core::convert::Infallible; +use core::future::poll_fn; +use core::task::Poll; use core::time::Duration; use aimdb_core::session::{ByteRead, ByteWrite, Delay}; use aimdb_core::RuntimeOps; -use alloc::vec::Vec; use embassy_futures::select::{select3, Either3}; use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex; use embassy_sync::channel::Channel; use mountain_mqtt::client::{ClientError, ConnectionSettings}; use mountain_mqtt::client_state::{ClientState, ClientStateNoQueue, ClientStateReceiveEvent}; -use mountain_mqtt::codec::mqtt_writer::{MqttBufWriter, MqttLenWriter, MqttWriter}; -use mountain_mqtt::codec::write::Write; use mountain_mqtt::data::property::ConnectProperty; use mountain_mqtt::data::quality_of_service::QualityOfService; use mountain_mqtt::error::{PacketReadError, PacketWriteError}; @@ -32,6 +32,7 @@ use mountain_mqtt::packets::packet_generic::PacketGeneric; use crate::embedded::manager::{now_ms, Error, FromApplicationMessage, Settings}; use crate::embedded::packet_reader::PacketReader; +use crate::embedded::write_ring::{encoded_len, WriteRing, CONTROL_RESERVE}; use crate::embedded::{ ActionChannel, AimdbMqttAction, AimdbMqttEvent, EventChannel, BUFFER_SIZE, MAX_PROPERTIES, }; @@ -41,11 +42,15 @@ const RX_CHUNK: usize = 256; /// The largest MQTT packet the session can receive. /// -/// Reassembly, the inbound slots and the encode buffer all come out of one -/// `BUFFER_SIZE`; outbound packets are encoded to exactly-sized `Vec`s -/// rather than a fixed buffer. +/// Reassembly and the inbound slots come out of one `BUFFER_SIZE`; outbound +/// packets go into the connector's write ring instead. const PACKET_BUFFER_SIZE: usize = BUFFER_SIZE - 2 * RX_CHUNK; +/// Ring space waited for before each received packet is parsed: enough for +/// the PUBACK a QoS 1 publish needs (6 bytes; mountain-mqtt adds no +/// properties). +const PUBACK_ROOM: usize = 16; + /// One chunk of freshly read bytes, in flight from the read half to the loop. type Chunk = heapless::Vec; @@ -54,6 +59,10 @@ type Chunk = heapless::Vec; /// Connects, subscribes `subscribe_topics`, then dispatches actions and /// forwards events. Returns only on failure — the caller reconnects. /// +/// `ring` is the connector's, reused across sessions; whatever an old session +/// left in it is discarded first, so nothing reaches the new socket ahead of +/// its CONNECT. +/// /// **At most once**: an action is taken off `actions` before it is performed, /// so the one in flight when a session ends is lost. Everything still queued /// survives. @@ -65,6 +74,7 @@ pub(crate) async fn run_session( subscribe_topics: &[(&str, QualityOfService)], events: &EventChannel, actions: &ActionChannel, + ring: &WriteRing, settings: &Settings, delay: &D, runtime: &dyn RuntimeOps, @@ -75,13 +85,11 @@ where D: Delay, { let inbound: Channel = Channel::new(); - // Four slots of a pointer each: enough that a burst of small packets does - // not park the loop, and cheap because the bytes live on the heap. - let outbound: Channel, 4> = Channel::new(); + ring.drain(); let session = client_loop( &inbound, - &outbound, + ring, connection_settings, subscribe_topics, events, @@ -91,7 +99,7 @@ where runtime, ); - match select3(read_into(rx, &inbound), write_out(tx, &outbound), session).await { + match select3(read_into(rx, &inbound), ring.write_out(tx), session).await { Either3::First(error) => error, Either3::Second(error) => error, Either3::Third(Err(error)) => error, @@ -122,20 +130,6 @@ async fn read_into( } } -/// Drain encoded packets to the socket. Never cancelled, which is what keeps -/// the non-cancel-safe `write_all` out of a `select` arm. -async fn write_out( - mut tx: W, - outbound: &Channel, 4>, -) -> Error { - loop { - let packet = outbound.receive().await; - if tx.write_all(&packet).await.is_err() || tx.flush().await.is_err() { - return Error::Client(ClientError::PacketWrite(PacketWriteError::ConnectionSend)); - } - } -} - fn receive_failed() -> Error { Error::Client(ClientError::PacketRead(PacketReadError::ConnectionReceive)) } @@ -144,7 +138,7 @@ fn receive_failed() -> Error { #[allow(clippy::too_many_arguments)] async fn client_loop( inbound: &Channel, - outbound: &Channel, 4>, + ring: &WriteRing, connection_settings: &ConnectionSettings<'static>, subscribe_topics: &[(&str, QualityOfService)], events: &EventChannel, @@ -189,7 +183,7 @@ async fn client_loop( properties, ); state.connect(&connect).map_err(client_error)?; - queue(outbound, encode(&connect)?).await; + ring.put(&connect, 0).await?; } loop { @@ -212,9 +206,16 @@ async fn client_loop( if connected && now.saturating_sub(last_ping_ms) >= ping_interval { last_ping_ms = now; let ping = state.send_ping().map_err(client_error)?; - // The one packet worth dropping rather than waiting for — see - // `queue_lossy`. - queue_lossy(outbound, encode(&ping)?); + // The one packet worth dropping rather than waiting for. A ping + // carries no state — `send_ping` bumps a counter but arms no + // response deadline — so a dropped one costs nothing and the next + // deadline tries again; if the link really is gone, the liveness + // window closes the session. Parking on a ping would leave the + // loop that has to notice a dead link stuck. + if !ring.try_put(&ping)? { + #[cfg(feature = "defmt")] + defmt::warn!("MQTT: write ring full, ping dropped"); + } } // Subscriptions go out one at a time: `ClientStateNoQueue` tracks a @@ -222,7 +223,7 @@ async fn client_loop( if connected && !state.waiting_for_responses() && next_topic < subscribe_topics.len() { let (topic, qos) = subscribe_topics[next_topic]; let packet = state.subscribe_packet(topic, qos).map_err(client_error)?; - queue(outbound, encode(&packet)?).await; + ring.put(&packet, 0).await?; state.subscribe_update(&packet).map_err(client_error)?; next_topic += 1; // `continue` skips the bottom-of-loop bookkeeping, so arm the @@ -236,18 +237,26 @@ async fn client_loop( // The action arm is armed only when a publish can actually be sent: // connected, nothing awaiting acknowledgement (the client state holds - // one in-flight slot), every subscription placed, and room to queue the - // bytes. This - // is what replaces the old inline wait for a PUBACK — the ping and - // liveness deadlines keep running while it is parked. - let action_ready = connected - && !state.waiting_for_responses() - && next_topic >= subscribe_topics.len() - && !outbound.is_full(); + // one in-flight slot), every subscription placed, and room in the ring + // for the largest PUBLISH plus its reserve. This is what replaces the + // old inline wait for a PUBACK — the ping and liveness deadlines keep + // running while it is parked. Room is checked on every poll, not once + // here: PUBACKs and pings can take ring space while the arm is parked. + let action_ready = + connected && !state.waiting_for_responses() && next_topic >= subscribe_topics.len(); + let publish_room = ring.max_publish() + CONTROL_RESERVE; let action_arm = async { if !action_ready { core::future::pending::<()>().await; } + poll_fn(|cx| { + if ring.poll_room(publish_room, cx.waker()) { + Poll::Ready(()) + } else { + Poll::Pending + } + }) + .await; actions.receive().await }; @@ -265,7 +274,7 @@ async fn client_loop( drain_packets( &mut reader, &mut state, - outbound, + ring, events, runtime, &mut last_ack_ms, @@ -274,7 +283,7 @@ async fn client_loop( .await?; } Either3::Second(action) => { - perform(action, &mut state, outbound).await?; + perform(action, &mut state, ring).await?; } // The timer fired: the top of the loop re-evaluates every deadline. Either3::Third(()) => {} @@ -293,39 +302,44 @@ async fn client_loop( async fn drain_packets( reader: &mut PacketReader, state: &mut ClientStateNoQueue, - outbound: &Channel, 4>, + ring: &WriteRing, events: &EventChannel, runtime: &dyn RuntimeOps, last_ack_ms: &mut u64, connected: &mut bool, ) -> Result<(), Error> { while let Some(total) = reader.framed_len().map_err(client_error)? { + // A burst of QoS 1 publishes needs a PUBACK each, so room for one is + // waited for here, before parsing: the parsed packet is too large to + // hold across an await in the session future. + ring.wait_room(PUBACK_ROOM).await; + // The packet borrows the reader's buffer, so everything that outlives - // it — the response bytes, the application event — is made owned inside - // this scope. `consume` can then take `&mut`. - let (response, received) = { + // it — the application event — is made owned inside this scope. + // `consume` can then take `&mut`. + let received = { let packet: PacketGeneric<'_, MAX_PROPERTIES, 0, 0> = reader.parse(total).map_err(client_error)?; // Produce the PUBACK before the state update, as upstream does, so - // the two cannot disagree about what was acknowledged. - let response = match state + // the two cannot disagree about what was acknowledged. It borrows + // the client state, so it goes into the ring before + // `state.receive` takes `&mut`. The room was waited for above and + // nothing else writes the ring in between. + if let Some(puback) = state .receive_produce_response(&packet) .map_err(client_error)? { - Some(puback) => Some(encode(&puback)?), - None => None, - }; + if !ring.try_put(&puback)? { + return Err(client_error(PacketWriteError::Overflow)); + } + } let event = state.receive(packet).map_err(client_error)?; - (response, Received::of(event)?) + Received::of(event)? }; reader.consume(total); - if let Some(bytes) = response { - queue(outbound, bytes).await; - } - // Every packet the state accepted proves the broker is alive. *last_ack_ms = now_ms(runtime); @@ -395,7 +409,7 @@ impl Received { async fn perform( action: AimdbMqttAction, state: &mut ClientStateNoQueue, - outbound: &Channel, 4>, + ring: &WriteRing, ) -> Result<(), Error> { match action { AimdbMqttAction::Publish { @@ -425,7 +439,20 @@ async fn perform( ); }) .map_err(client_error)?; - queue(outbound, encode(&packet)?).await; + // A frame larger than the ring can always grant would wait + // forever; skip it before the state commits to it. + let len = encoded_len(&packet)?; + if len > ring.max_publish() { + #[cfg(feature = "defmt")] + defmt::warn!( + "MQTT: dropping publish to {}: {} bytes exceed the {}-byte write ring limit", + topic.as_str(), + len, + ring.max_publish() + ); + return Ok(()); + } + ring.put(&packet, CONTROL_RESERVE).await?; state.publish_update(&packet).map_err(client_error)?; } AimdbMqttAction::Subscribe { topic, qos } => { @@ -438,55 +465,13 @@ async fn perform( defmt::warn!("MQTT: dropping subscribe to {}: {}", topic.as_str(), _e); }) .map_err(client_error)?; - queue(outbound, encode(&packet)?).await; + ring.put(&packet, 0).await?; state.subscribe_update(&packet).map_err(client_error)?; } } Ok(()) } -/// Encode a packet to exactly its own length: a counting pass, then a real -/// one, so no fixed buffer is sized for the largest packet anyone might send. -fn encode(packet: &P) -> Result, Error> { - let mut len_writer = MqttLenWriter::new(); - len_writer.put(packet).map_err(write_error)?; - - let mut bytes = alloc::vec![0u8; len_writer.position()]; - let mut writer = MqttBufWriter::new(&mut bytes); - writer.put(packet).map_err(write_error)?; - Ok(bytes) -} - -/// Queue encoded bytes for the write half, waiting for a slot. -/// -/// Everything the protocol obliges us to send goes through here: CONNECT, -/// SUBSCRIBE, PUBLISH and the PUBACKs answering QoS 1 delivery. None of those -/// can be dropped — the state machine has already committed to them, so a -/// discarded packet leaves our state and the wire disagreeing, with nothing to -/// resync on. -/// -/// Waiting cannot deadlock: [`write_out`] is this channel's only consumer and -/// is a sibling arm of the same `select`, so parking here is what lets it run. -/// It is also the backpressure — a peer that stops reading stops us encoding. -async fn queue(outbound: &Channel, 4>, bytes: Vec) { - outbound.send(bytes).await; -} - -/// Queue encoded bytes only if the write half is keeping up, dropping them if -/// it is not. -/// -/// For pings alone. A ping carries no state — `send_ping` bumps a counter but -/// arms no response deadline — so a dropped one costs nothing and the next -/// ping deadline tries again; if the link really is gone, the liveness window -/// closes the session. Parking on a ping would be worse than skipping it: the -/// loop that has to notice the link is gone would be the thing stuck. -fn queue_lossy(outbound: &Channel, 4>, bytes: Vec) { - if outbound.try_send(bytes).is_err() { - #[cfg(feature = "defmt")] - defmt::warn!("MQTT: write queue full, ping dropped"); - } -} - /// Milliseconds to sleep before the earliest armed deadline. fn next_deadline( now: u64, @@ -509,13 +494,10 @@ fn client_error(error: impl Into) -> Error { Error::Client(error.into()) } -fn write_error(error: PacketWriteError) -> Error { - Error::Client(ClientError::PacketWrite(error)) -} - #[cfg(test)] mod tests { use super::*; + use alloc::vec::Vec; /// Halves that never do anything: enough to build the session future and /// measure it without polling it. @@ -572,6 +554,7 @@ mod tests { let settings = Settings::default(); let connection_settings = ConnectionSettings::unauthenticated("size-probe"); let runtime = aimdb_core::executor::test_support::NoopRuntimeOps; + let ring = WriteRing::new(64); // Built, never polled: `size_of_val` on the future is the whole point. let session = run_session( @@ -581,6 +564,7 @@ mod tests { &[], &events, &actions, + &ring, &settings, &NullDelay, &runtime, @@ -664,6 +648,20 @@ mod tests { ); } + #[test] + fn puback_room_covers_the_puback_the_client_state_produces() { + use mountain_mqtt::data::packet_identifier::PacketIdentifier; + use mountain_mqtt::data::reason_code::PublishReasonCode; + use mountain_mqtt::packets::puback::Puback; + + let puback: Puback<'_, MAX_PROPERTIES> = Puback::new( + PacketIdentifier(u16::MAX), + PublishReasonCode::Success, + heapless::Vec::new(), + ); + assert!(encoded_len(&puback).unwrap() <= PUBACK_ROOM); + } + #[test] fn a_deadline_in_the_past_still_sleeps_a_tick() { // Never zero: a zero-length sleep would spin the loop. diff --git a/aimdb-mqtt-connector/src/embedded/tls.rs b/aimdb-mqtt-connector/src/embedded/tls.rs index 4b088d2e..65571de7 100644 --- a/aimdb-mqtt-connector/src/embedded/tls.rs +++ b/aimdb-mqtt-connector/src/embedded/tls.rs @@ -24,6 +24,7 @@ use embedded_tls::{ use crate::embedded::manager::Settings; use crate::embedded::session_loop::run_session; +use crate::embedded::write_ring::{WriteRing, DEFAULT_WRITE_BUFFER}; use mountain_mqtt::client::ConnectionSettings; use mountain_mqtt::data::quality_of_service::QualityOfService; @@ -343,6 +344,9 @@ where .map(|topic| (topic.as_str(), QualityOfService::Qos1)) .collect(); + // Allocated once and reused by every session. + let ring = WriteRing::new(DEFAULT_WRITE_BUFFER); + loop { // Certificate validity needs real time. Take it from the runtime when // it has a wall clock; otherwise wait for whatever feeds `WallClock` @@ -410,6 +414,7 @@ where &subscribe_topics, &events, &actions, + &ring, &settings, &delay, runtime.as_ref(), diff --git a/aimdb-mqtt-connector/src/embedded/write_ring.rs b/aimdb-mqtt-connector/src/embedded/write_ring.rs new file mode 100644 index 00000000..56da78c5 --- /dev/null +++ b/aimdb-mqtt-connector/src/embedded/write_ring.rs @@ -0,0 +1,358 @@ +//! The session's write queue: one byte ring, allocated once per connector. +//! +//! The session loop encodes every packet straight into a grant of the ring; +//! [`WriteRing::write_out`] sends committed bytes to the socket and releases +//! them. Both ends run in the same task under the session's `select3`, so the +//! ring needs no lock. Two wakers carry the signals the ring's own notifier +//! does not: `data` from a commit to `write_out`, and `room` from a release to +//! whatever waits for space. + +use core::future::poll_fn; +use core::task::Poll; + +use aimdb_core::session::ByteWrite; +use bbqueue::traits::coordination::cas::AtomicCoord; +use bbqueue::traits::notifier::polling::Polling; +use bbqueue::traits::storage::BoxedSlice; +use bbqueue::BBQueue; +use embassy_sync::waitqueue::AtomicWaker; +use mountain_mqtt::client::ClientError; +use mountain_mqtt::codec::mqtt_writer::{MqttBufWriter, MqttLenWriter, MqttWriter}; +use mountain_mqtt::codec::write::Write; +use mountain_mqtt::error::PacketWriteError; + +use crate::embedded::manager::Error; + +/// Default ring size, the same as the inbound side's `BUFFER_SIZE`. +pub(crate) const DEFAULT_WRITE_BUFFER: usize = 4096; + +/// Bytes kept free behind a PUBLISH, so the control packet after it (a +/// PUBACK, a ping) still finds room. +pub(crate) const CONTROL_RESERVE: usize = 64; + +/// One connector's write ring. +pub(crate) struct WriteRing { + queue: BBQueue, + data: AtomicWaker, + room: AtomicWaker, +} + +impl WriteRing { + pub(crate) fn new(capacity: usize) -> Self { + Self { + queue: BBQueue::new_with_storage(BoxedSlice::new(capacity)), + data: AtomicWaker::new(), + room: AtomicWaker::new(), + } + } + + /// The largest PUBLISH frame the ring always has room for eventually. + /// + /// A bipbuffer grants only contiguous space: once its pointers have both + /// moved to `k`, an empty ring grants at most `max(capacity − k, k − 1)` + /// bytes, about half its capacity in the worst case. A frame plus its + /// reserve must fit in that half, or it could wait forever. + pub(crate) fn max_publish(&self) -> usize { + (self.queue.capacity() / 2).saturating_sub(CONTROL_RESERVE) + } + + /// Whether a contiguous grant of `n` bytes exists right now. + /// + /// bbqueue has no free-space query, so this takes the grant and drops it + /// uncommitted. A probe that wraps commits an early wraparound (the + /// unused tail is skipped until the reader passes it); no data changes. + pub(crate) fn has_room(&self, n: usize) -> bool { + self.queue.stream_producer().grant_exact(n).is_ok() + } + + /// [`has_room`](Self::has_room), registering `waker` for the next release + /// when there is none. + pub(crate) fn poll_room(&self, n: usize, waker: &core::task::Waker) -> bool { + if self.has_room(n) { + return true; + } + self.room.register(waker); + // A release between the probe and the registration woke nobody. + self.has_room(n) + } + + /// Wait until a contiguous grant of `n` bytes exists. + pub(crate) async fn wait_room(&self, n: usize) { + poll_fn(|cx| { + if self.poll_room(n, cx.waker()) { + Poll::Ready(()) + } else { + Poll::Pending + } + }) + .await + } + + /// Encode `packet` into the ring, waiting for room. `reserve` more bytes + /// are granted behind it and left free. + /// + /// Waiting cannot deadlock: [`write_out`](Self::write_out) is the ring's + /// only consumer and a sibling arm of the session's `select3`, so parking + /// here is what lets it run. It is also the backpressure: a peer that + /// stops reading stops the session encoding. + pub(crate) async fn put(&self, packet: &P, reserve: usize) -> Result<(), Error> { + let len = encoded_len(packet)?; + let producer = self.queue.stream_producer(); + let mut grant = poll_fn(|cx| match producer.grant_exact(len + reserve) { + Ok(grant) => Poll::Ready(grant), + Err(_) => { + self.room.register(cx.waker()); + match producer.grant_exact(len + reserve) { + Ok(grant) => Poll::Ready(grant), + Err(_) => Poll::Pending, + } + } + }) + .await; + let mut writer = MqttBufWriter::new(&mut grant[..len]); + writer.put(packet).map_err(write_error)?; + grant.commit(len); + self.data.wake(); + Ok(()) + } + + /// Encode `packet` only if there is room now. `Ok(false)`: dropped. + pub(crate) fn try_put(&self, packet: &P) -> Result { + let len = encoded_len(packet)?; + let Ok(mut grant) = self.queue.stream_producer().grant_exact(len) else { + return Ok(false); + }; + let mut writer = MqttBufWriter::new(&mut grant[..len]); + writer.put(packet).map_err(write_error)?; + grant.commit(len); + self.data.wake(); + Ok(true) + } + + /// Discard every byte still queued, so nothing from an old session + /// reaches a new socket ahead of its CONNECT. + pub(crate) fn drain(&self) { + let consumer = self.queue.stream_consumer(); + // Two reads at most: the tail before a wraparound, then the head. + while let Ok(grant) = consumer.read() { + let n = grant.len(); + grant.release(n); + } + self.room.wake(); + } + + /// Send committed bytes to the socket, releasing them once written and + /// flushed. Returns only on a write error. Never cancelled mid-write: + /// it is a `select3` arm that ends the session when it returns. + pub(crate) async fn write_out(&self, mut tx: W) -> Error { + let consumer = self.queue.stream_consumer(); + loop { + let grant = poll_fn(|cx| match consumer.read() { + Ok(grant) => Poll::Ready(grant), + Err(_) => { + self.data.register(cx.waker()); + match consumer.read() { + Ok(grant) => Poll::Ready(grant), + Err(_) => Poll::Pending, + } + } + }) + .await; + if tx.write_all(&grant).await.is_err() || tx.flush().await.is_err() { + return Error::Client(ClientError::PacketWrite(PacketWriteError::ConnectionSend)); + } + let n = grant.len(); + grant.release(n); + self.room.wake(); + } + } +} + +/// Bytes `packet` encodes to. +pub(crate) fn encoded_len(packet: &P) -> Result { + let mut len_writer = MqttLenWriter::new(); + len_writer.put(packet).map_err(write_error)?; + Ok(len_writer.position()) +} + +fn write_error(error: PacketWriteError) -> Error { + Error::Client(ClientError::PacketWrite(error)) +} + +#[cfg(test)] +mod tests { + use super::*; + use alloc::sync::Arc; + use alloc::task::Wake; + use alloc::vec::Vec; + use core::cell::RefCell; + + use core::future::Future; + use core::pin::pin; + use core::sync::atomic::{AtomicUsize, Ordering}; + use core::task::{Context, Waker}; + use embassy_sync::blocking_mutex::CriticalSectionMutex; + + use mountain_mqtt::codec::mqtt_writer::MqttWriter; + + /// A "packet" of `n` copies of one byte. + struct Bytes(usize, u8); + + impl Write for Bytes { + fn write<'a, W: MqttWriter<'a>>(&self, writer: &mut W) -> Result<(), PacketWriteError> { + writer.put_slice(&alloc::vec![self.1; self.0]) + } + } + + /// Records everything written. + #[derive(Clone)] + struct Sink(Arc>>>); + + impl Sink { + fn new() -> Self { + Self(Arc::new(CriticalSectionMutex::new( + RefCell::new(Vec::new()), + ))) + } + + fn written(&self) -> Vec { + self.0.lock(|w| w.borrow().clone()) + } + } + + impl ByteWrite for Sink { + async fn write_all(&mut self, buf: &[u8]) -> aimdb_core::session::TransportResult<()> { + self.0.lock(|w| w.borrow_mut().extend_from_slice(buf)); + Ok(()) + } + + async fn flush(&mut self) -> aimdb_core::session::TransportResult<()> { + Ok(()) + } + } + + #[derive(Default)] + struct Count(AtomicUsize); + + impl Wake for Count { + fn wake(self: Arc) { + self.0.fetch_add(1, Ordering::Relaxed); + } + } + + /// Poll a future once with `waker`. + fn poll_once(f: core::pin::Pin<&mut F>, waker: &Waker) -> Poll { + f.poll(&mut Context::from_waker(waker)) + } + + /// Everything committed, in read order, released as it is read. + fn read_all(ring: &WriteRing) -> Vec { + let consumer = ring.queue.stream_consumer(); + let mut out = Vec::new(); + while let Ok(grant) = consumer.read() { + out.extend_from_slice(&grant); + let n = grant.len(); + grant.release(n); + } + out + } + + /// Write `n` bytes and read them back, moving both pointers to `n`. + fn advance(ring: &WriteRing, n: usize) { + if n > 0 { + assert!(ring.try_put(&Bytes(n, 0)).unwrap()); + assert_eq!(read_all(ring).len(), n); + } + } + + fn ready(f: impl Future>) { + let f = pin!(f); + assert!(matches!(poll_once(f, Waker::noop()), Poll::Ready(Ok(())))); + } + + #[test] + fn a_probe_that_wraps_changes_no_data() { + let ring = WriteRing::new(16); + advance(&ring, 6); + ready(ring.put(&Bytes(6, b'b'), 0)); + // 12..16 is too short for 5 bytes; the probe wraps to the start. + assert!(ring.has_room(5)); + ready(ring.put(&Bytes(3, b'c'), 0)); + assert_eq!(read_all(&ring), b"bbbbbbccc"); + } + + #[test] + fn an_empty_ring_always_grants_half_its_capacity() { + const CAPACITY: usize = 64; + for offset in 0..CAPACITY { + let ring = WriteRing::new(CAPACITY); + advance(&ring, offset); + assert!(ring.has_room(CAPACITY / 2), "offset {offset}"); + } + // Drained at the middle, one byte more than half never fits. + let ring = WriteRing::new(CAPACITY); + advance(&ring, CAPACITY / 2); + assert!(!ring.has_room(CAPACITY / 2 + 1)); + } + + #[test] + fn max_publish_leaves_room_for_the_reserve_in_half_the_ring() { + let ring = WriteRing::new(DEFAULT_WRITE_BUFFER); + assert_eq!(ring.max_publish(), 1984); + assert_eq!( + ring.max_publish() + CONTROL_RESERVE, + DEFAULT_WRITE_BUFFER / 2 + ); + } + + #[test] + fn drain_discards_an_old_sessions_bytes() { + let ring = WriteRing::new(64); + ready(ring.put(&Bytes(10, b'a'), 0)); + ring.drain(); + ready(ring.put(&Bytes(4, b'b'), 0)); + assert_eq!(read_all(&ring), b"bbbb"); + } + + #[test] + fn a_puback_waits_for_room_and_goes_out_in_order() { + let ring = WriteRing::new(256); + ready(ring.put(&Bytes(200, b'p'), 0)); + ready(ring.put(&Bytes(40, b'q'), 0)); + + // 16 bytes left, and nothing has been read: the PUBACK waits. + let woken = Arc::new(Count::default()); + let waker = Waker::from(woken.clone()); + let mut puback = pin!(ring.put(&Bytes(20, b'r'), 0)); + assert!(poll_once(puback.as_mut(), &waker).is_pending()); + + // `write_out` sends what is queued and releases it. + let sink = Sink::new(); + let mut write_out = pin!(ring.write_out(sink.clone())); + assert!(poll_once(write_out.as_mut(), Waker::noop()).is_pending()); + assert_eq!(sink.written().len(), 240); + assert!( + woken.0.load(Ordering::Relaxed) > 0, + "the release woke the PUBACK" + ); + + assert!(matches!( + poll_once(puback.as_mut(), &waker), + Poll::Ready(Ok(())) + )); + assert!(poll_once(write_out.as_mut(), Waker::noop()).is_pending()); + let written = sink.written(); + assert_eq!(written.len(), 260); + assert!(written[..200].iter().all(|&b| b == b'p')); + assert!(written[200..240].iter().all(|&b| b == b'q')); + assert!(written[240..].iter().all(|&b| b == b'r')); + } + + #[test] + fn try_put_drops_rather_than_waits() { + let ring = WriteRing::new(16); + ready(ring.put(&Bytes(14, b'a'), 0)); + assert!(!ring.try_put(&Bytes(4, b'p')).unwrap()); + assert_eq!(read_all(&ring).len(), 14); + } +} From dc9a3a4d2947e909802d63ad1306be755fe84d25 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alexander=20Schn=C3=B6rch?= Date: Sun, 4 Oct 2026 18:42:46 +0000 Subject: [PATCH 2/2] fix(mqtt): fail instead of hang on a packet the write ring can never grant put() now returns Overflow when a frame plus its reserve exceeds half the ring, the most a bipbuffer can always grant contiguously. A CONNECT or SUBSCRIBE that large used to park forever outside the session's select. perform() reuses its encoded length through put_sized. Adds the review's size-limit proofs as regression tests, the two CONNECT cases flipped to the new behaviour, and runs write_ring_proofs in make test and make clippy. Co-Authored-By: Claude Opus 5.5 --- Makefile | 4 + .../src/embedded/session_loop.rs | 127 ++++++++++++- .../src/embedded/write_ring.rs | 111 +++++++++++- .../tests/write_ring_proofs.rs | 168 ++++++++++++++++++ 4 files changed, 400 insertions(+), 10 deletions(-) create mode 100644 aimdb-mqtt-connector/tests/write_ring_proofs.rs diff --git a/Makefile b/Makefile index fe75335b..15881391 100644 --- a/Makefile +++ b/Makefile @@ -253,6 +253,8 @@ test: 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" cargo test --package aimdb-mqtt-connector --no-default-features --features "_test-tokio-broker" --test session_loop + @printf "$(YELLOW) → Testing MQTT connector (embedded packet-size limits against a fake broker)$(NC)\n" + cargo test --package aimdb-mqtt-connector --no-default-features --features "_test-tokio-broker" --test write_ring_proofs @printf "$(YELLOW) → Testing MQTT connector (the same criteria over mqtts://)$(NC)\n" cargo test --package aimdb-mqtt-connector --no-default-features --features "_test-tls-broker" --test tls_session @printf "$(YELLOW) → Testing MQTT connector (no_std unit tests: framing, deadlines, TLS duplex)$(NC)\n" @@ -402,6 +404,8 @@ clippy: 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" cargo clippy --package aimdb-mqtt-connector --no-default-features --features "_test-tokio-broker" --test session_loop -- -D warnings + @printf "$(YELLOW) → Clippy on MQTT connector (embedded packet-size limits)$(NC)\n" + cargo clippy --package aimdb-mqtt-connector --no-default-features --features "_test-tokio-broker" --test write_ring_proofs -- -D warnings @printf "$(YELLOW) → Clippy on MQTT connector (the same criteria over mqtts://)$(NC)\n" cargo clippy --package aimdb-mqtt-connector --no-default-features --features "_test-tls-broker" --test tls_session -- -D warnings @printf "$(YELLOW) → Clippy on MQTT connector (no_std unit tests)$(NC)\n" diff --git a/aimdb-mqtt-connector/src/embedded/session_loop.rs b/aimdb-mqtt-connector/src/embedded/session_loop.rs index 542d227b..9bf39b2f 100644 --- a/aimdb-mqtt-connector/src/embedded/session_loop.rs +++ b/aimdb-mqtt-connector/src/embedded/session_loop.rs @@ -442,7 +442,7 @@ async fn perform( // A frame larger than the ring can always grant would wait // forever; skip it before the state commits to it. let len = encoded_len(&packet)?; - if len > ring.max_publish() { + if !ring.fits(len, CONTROL_RESERVE) { #[cfg(feature = "defmt")] defmt::warn!( "MQTT: dropping publish to {}: {} bytes exceed the {}-byte write ring limit", @@ -452,7 +452,7 @@ async fn perform( ); return Ok(()); } - ring.put(&packet, CONTROL_RESERVE).await?; + ring.put_sized(&packet, len, CONTROL_RESERVE).await?; state.publish_update(&packet).map_err(client_error)?; } AimdbMqttAction::Subscribe { topic, qos } => { @@ -668,3 +668,126 @@ mod tests { assert_eq!(next_deadline(1_000, true, 100, 500, None), 1); } } + +#[cfg(test)] +mod proofs { + //! Where the embedded backend's size limits sit today: the largest + //! PUBLISH it sends, what it does with a larger one (skipped without an + //! error and without a count), and the largest packet it receives. + use super::*; + use crate::embedded::write_ring::DEFAULT_WRITE_BUFFER; + use alloc::vec::Vec; + use core::future::Future; + use core::pin::pin; + use core::task::{Context, Waker}; + use mountain_mqtt::data::reason_code::ConnectReasonCode; + use mountain_mqtt::packets::connack::Connack; + + fn connected() -> ClientStateNoQueue { + let mut state = ClientStateNoQueue::new(); + let connect: Connect<'_, 1, 0> = + Connect::new(60, None, None, "proof", true, None, heapless::Vec::new()); + state.connect(&connect).unwrap(); + let connack: Connack<'_, MAX_PROPERTIES> = + Connack::new(false, ConnectReasonCode::Success, heapless::Vec::new()); + state + .receive(PacketGeneric::<'_, MAX_PROPERTIES, 0, 0>::Connack(connack)) + .unwrap(); + state + } + + /// Payload length whose QoS 1 PUBLISH to `t` encodes to exactly `frame`. + fn payload_for(frame: usize) -> usize { + let mut probe = connected(); + (0..frame) + .rev() + .find(|&n| { + let payload = alloc::vec![0u8; n]; + let packet = probe + .publish_packet("t", &payload, QualityOfService::Qos1, false) + .unwrap(); + encoded_len(&packet).unwrap() == frame + }) + .unwrap() + } + + fn publish(n: usize) -> AimdbMqttAction { + AimdbMqttAction::Publish { + topic: "t".into(), + payload: alloc::vec![b'x'; n], + qos: QualityOfService::Qos1, + retain: false, + } + } + + fn run( + action: AimdbMqttAction, + state: &mut ClientStateNoQueue, + ring: &WriteRing, + ) -> Poll> { + let f = pin!(perform(action, state, ring)); + f.poll(&mut Context::from_waker(Waker::noop())) + } + + #[test] + fn proof_a_1984_byte_publish_goes_out() { + let ring = WriteRing::new(DEFAULT_WRITE_BUFFER); + assert_eq!(ring.max_publish(), 1984); + let mut state = connected(); + let n = payload_for(1984); + assert!(matches!( + run(publish(n), &mut state, &ring), + Poll::Ready(Ok(())) + )); + assert!(!ring.has_room(DEFAULT_WRITE_BUFFER), "bytes were queued"); + assert!(state.waiting_for_responses(), "QoS 1 publish in flight"); + } + + #[test] + fn proof_a_1985_byte_publish_vanishes_without_an_error() { + let ring = WriteRing::new(DEFAULT_WRITE_BUFFER); + let mut state = connected(); + let n = payload_for(1985); + assert!(matches!( + run(publish(n), &mut state, &ring), + Poll::Ready(Ok(())) + )); + assert!(ring.has_room(DEFAULT_WRITE_BUFFER), "nothing was queued"); + assert!(!state.waiting_for_responses(), "nothing in flight"); + // No counter exists to assert on: the only trace is a defmt warn. + } + + /// A QoS 0 PUBLISH to `t` that is `total` bytes on the wire. + fn inbound_publish(total: usize) -> Vec { + let remaining = total - 3; // header byte + 2-byte varint + let mut bytes = alloc::vec![ + 0x30, + (remaining % 128) as u8 | 0x80, + (remaining / 128) as u8, + 0x00, + 0x01, + b't', + 0x00, + ]; + bytes.resize(total, b'x'); + bytes + } + + fn receive(total: usize) -> Result { + let mut reader = PacketReader::::new(); + for chunk in inbound_publish(total).chunks(RX_CHUNK) { + reader.feed(chunk)?; + if let Some(n) = reader.framed_len()? { + return Ok(n); + } + } + unreachable!("the whole packet was fed") + } + + #[test] + fn proof_the_receive_side_takes_3584_and_refuses_3585() { + assert_eq!(PACKET_BUFFER_SIZE, 3584); + assert_eq!(receive(3584), Ok(3584)); + assert_eq!(receive(3585), Err(PacketReadError::PacketTooLargeForBuffer)); + } +} diff --git a/aimdb-mqtt-connector/src/embedded/write_ring.rs b/aimdb-mqtt-connector/src/embedded/write_ring.rs index 56da78c5..36eb0003 100644 --- a/aimdb-mqtt-connector/src/embedded/write_ring.rs +++ b/aimdb-mqtt-connector/src/embedded/write_ring.rs @@ -47,13 +47,18 @@ impl WriteRing { } /// The largest PUBLISH frame the ring always has room for eventually. + pub(crate) fn max_publish(&self) -> usize { + (self.queue.capacity() / 2).saturating_sub(CONTROL_RESERVE) + } + + /// Whether a `len`-byte frame plus `reserve` always fits eventually. /// /// A bipbuffer grants only contiguous space: once its pointers have both /// moved to `k`, an empty ring grants at most `max(capacity − k, k − 1)` - /// bytes, about half its capacity in the worst case. A frame plus its - /// reserve must fit in that half, or it could wait forever. - pub(crate) fn max_publish(&self) -> usize { - (self.queue.capacity() / 2).saturating_sub(CONTROL_RESERVE) + /// bytes, about half its capacity in the worst case. A larger grant could + /// wait forever, even with nothing queued. + pub(crate) fn fits(&self, len: usize, reserve: usize) -> bool { + len + reserve <= self.queue.capacity() / 2 } /// Whether a contiguous grant of `n` bytes exists right now. @@ -89,7 +94,8 @@ impl WriteRing { } /// Encode `packet` into the ring, waiting for room. `reserve` more bytes - /// are granted behind it and left free. + /// are granted behind it and left free. A packet that does not + /// [`fit`](Self::fits) fails with `Overflow` instead of waiting forever. /// /// Waiting cannot deadlock: [`write_out`](Self::write_out) is the ring's /// only consumer and a sibling arm of the session's `select3`, so parking @@ -97,6 +103,21 @@ impl WriteRing { /// stops reading stops the session encoding. pub(crate) async fn put(&self, packet: &P, reserve: usize) -> Result<(), Error> { let len = encoded_len(packet)?; + if !self.fits(len, reserve) { + return Err(write_error(PacketWriteError::Overflow)); + } + self.put_sized(packet, len, reserve).await + } + + /// [`put`](Self::put) for a packet whose encoded length is already known + /// and [fits](Self::fits). + pub(crate) async fn put_sized( + &self, + packet: &P, + len: usize, + reserve: usize, + ) -> Result<(), Error> { + debug_assert!(self.fits(len, reserve)); let producer = self.queue.stream_producer(); let mut grant = poll_fn(|cx| match producer.grant_exact(len + reserve) { Ok(grant) => Poll::Ready(grant), @@ -317,8 +338,9 @@ mod tests { #[test] fn a_puback_waits_for_room_and_goes_out_in_order() { let ring = WriteRing::new(256); - ready(ring.put(&Bytes(200, b'p'), 0)); - ready(ring.put(&Bytes(40, b'q'), 0)); + // Filled with `try_put`: 200 bytes is more than `put` accepts. + assert!(ring.try_put(&Bytes(200, b'p')).unwrap()); + assert!(ring.try_put(&Bytes(40, b'q')).unwrap()); // 16 bytes left, and nothing has been read: the PUBACK waits. let woken = Arc::new(Count::default()); @@ -351,8 +373,81 @@ mod tests { #[test] fn try_put_drops_rather_than_waits() { let ring = WriteRing::new(16); - ready(ring.put(&Bytes(14, b'a'), 0)); + assert!(ring.try_put(&Bytes(14, b'a')).unwrap()); assert!(!ring.try_put(&Bytes(4, b'p')).unwrap()); assert_eq!(read_all(&ring).len(), 14); } } + +#[cfg(test)] +mod proofs { + //! A control packet larger than the ring can always grant fails instead + //! of parking forever; one that fits goes out from any offset. + use super::*; + use core::future::Future; + use core::pin::pin; + use core::task::{Context, Waker}; + use mountain_mqtt::packets::connect::Connect; + + fn connect_with_password(password: &[u8]) -> Connect<'_, 1, 0> { + Connect::new( + 60, + Some("user"), + Some(password), + "proof", + true, + None, + heapless::Vec::new(), + ) + } + + /// Move both pointers to `n` with nothing left queued, as a session that + /// sent `n` bytes and then dropped leaves the ring after `drain`. + fn leave_at(ring: &WriteRing, n: usize) { + struct Filler(usize); + impl Write for Filler { + fn write<'a, W: MqttWriter<'a>>(&self, w: &mut W) -> Result<(), PacketWriteError> { + w.put_slice(&alloc::vec![0u8; self.0]) + } + } + assert!(ring.try_put(&Filler(n)).unwrap()); + ring.drain(); + } + + fn poll(f: core::pin::Pin<&mut F>) -> Poll { + f.poll(&mut Context::from_waker(Waker::noop())) + } + + fn is_overflow(r: &Poll>) -> bool { + matches!( + r, + Poll::Ready(Err(Error::Client(ClientError::PacketWrite( + PacketWriteError::Overflow + )))) + ) + } + + #[test] + fn a_connect_over_half_the_ring_fails_at_any_offset() { + let password = alloc::vec![b'p'; 2100]; + let connect = connect_with_password(&password); + for offset in [0, DEFAULT_WRITE_BUFFER / 2] { + let ring = WriteRing::new(DEFAULT_WRITE_BUFFER); + leave_at(&ring, offset); + let mut put = pin!(ring.put(&connect, 0)); + assert!(is_overflow(&poll(put.as_mut())), "offset {offset}"); + } + } + + #[test] + fn a_connect_that_fits_goes_out_from_offset_2048() { + // Just under half the ring once the CONNECT's own fields are added. + let password = alloc::vec![b'p'; 2000]; + let connect = connect_with_password(&password); + assert!(encoded_len(&connect).unwrap() <= DEFAULT_WRITE_BUFFER / 2); + let ring = WriteRing::new(DEFAULT_WRITE_BUFFER); + leave_at(&ring, DEFAULT_WRITE_BUFFER / 2); + let mut put = pin!(ring.put(&connect, 0)); + assert!(matches!(poll(put.as_mut()), Poll::Ready(Ok(())))); + } +} diff --git a/aimdb-mqtt-connector/tests/write_ring_proofs.rs b/aimdb-mqtt-connector/tests/write_ring_proofs.rs new file mode 100644 index 00000000..ad352078 --- /dev/null +++ b/aimdb-mqtt-connector/tests/write_ring_proofs.rs @@ -0,0 +1,168 @@ +//! The embedded backend's packet-size behaviour against a fake broker +//! (`_test-tokio-broker`). +#![cfg(feature = "_test-tokio-broker")] + +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use tokio::net::TcpListener; + +mod common; +use common::{fake_broker, Seen}; + +// 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") +} +// This binary must define `_defmt_timestamp` itself. `embassy-time` would — +// `defmt-timestamp-uptime` is enabled here, as it is for `embassy_broker` — +// but nothing in this test references `embassy-time`, so its object never +// reaches the link and the symbol would be undefined. `embassy_broker` pulls +// it in through `embassy-net` and therefore must *not* define one. +defmt::timestamp!("{=u64:us}", 0); + +/// Real wall-clock time; the session loop's delays are `embassy_time`'s until +/// it takes core's `Delay`. +struct HostClock; +impl embassy_time_driver::Driver for HostClock { + fn now(&self) -> u64 { + use std::sync::OnceLock; + use std::time::Instant; + static START: OnceLock = 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); + +use aimdb_core::buffer::BufferCfg; +use aimdb_core::AimDbBuilder; +use aimdb_mqtt_connector::MqttConnector; +use aimdb_tokio_adapter::net::TokioNet; +use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; + +/// A 3,000-byte publish never reaches the broker, the session never errors, +/// and the small publishes around it keep flowing. Nothing counts the drop. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn proof_an_oversize_publish_is_dropped_silently() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let seen = Arc::new(Mutex::new(Seen::default())); + + let connector = MqttConnector::new(format!("mqtt://127.0.0.1:{port}")) + .transport(TokioNet::tcp()) + .with_client_id("proof-drop"); + let mut builder = AimDbBuilder::new() + .runtime(Arc::new(TokioAdapter)) + .with_connector(connector); + builder.configure::("blob", |reg| { + reg.buffer(BufferCfg::SingleLatest) + .source(|_ctx, producer| async move { + loop { + producer.produce(1u64); // 3,000 bytes + tokio::time::sleep(Duration::from_millis(100)).await; + producer.produce(2u64); // 1 byte + tokio::time::sleep(Duration::from_millis(100)).await; + } + }) + .link_to("mqtt://sensors/blob") + .with_serializer(|_ctx, v: &u64| { + Ok(if *v == 1 { + vec![b'x'; 3000] + } else { + b"2".to_vec() + }) + }) + .finish(); + }); + let (_db, runner) = builder.build().await.expect("build db"); + let broker = fake_broker(listener, seen.clone(), 0, None); + + let until_small_ones = async { + while seen.lock().unwrap().published.len() < 5 { + tokio::time::sleep(Duration::from_millis(5)).await; + } + }; + tokio::select! { + _ = runner.run() => panic!("the session loop returned"), + _ = broker => panic!("the broker returned"), + _ = until_small_ones => {} + _ = tokio::time::sleep(Duration::from_secs(30)) => panic!("watchdog"), + } + + let seen = seen.lock().unwrap(); + assert!( + seen.published.iter().all(|(_, p)| p == b"2"), + "only the 1-byte payloads reached the broker" + ); + assert_eq!(seen.connects, 1, "the session never errored"); +} + +/// Build a client subscribed to `sensors/temperature` against a broker that +/// pushes `payload_len` bytes after every SUBACK, as it would a retained +/// message, and count connections after `wait`. +async fn connects_with_push(payload_len: usize, wait: Duration) -> usize { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let seen = Arc::new(Mutex::new(Seen::default())); + + let connector = MqttConnector::new(format!("mqtt://127.0.0.1:{port}")) + .transport(TokioNet::tcp()) + .with_client_id("proof-retained"); + let mut builder = AimDbBuilder::new() + .runtime(Arc::new(TokioAdapter)) + .with_connector(connector); + builder.configure::("temperature", |reg| { + reg.buffer(BufferCfg::SingleLatest) + .link_from("mqtt://sensors/temperature") + .with_deserializer(|_ctx, data: &[u8]| Ok::(data.len() as u64)) + .finish(); + }); + let (_db, runner) = builder.build().await.expect("build db"); + + let payload = vec![b'x'; payload_len]; + let broker = fake_broker( + listener, + seen.clone(), + 0, + Some(("sensors/temperature", payload.as_slice())), + ); + tokio::select! { + _ = runner.run() => panic!("the session loop returned"), + _ = broker => panic!("the broker returned"), + _ = tokio::time::sleep(wait) => {} + } + let connects = seen.lock().unwrap().connects; + connects +} + +/// A 3,000-byte message fits the receive buffer. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn proof_a_3000_byte_retained_message_is_received_once() { + assert_eq!(connects_with_push(3000, Duration::from_secs(5)).await, 1); +} + +/// A 4,000-byte message ends every session it reaches, so a retained one +/// reconnects the client forever (one cycle per 2 s reconnection delay): the +/// CONNECT does not tell the broker the largest packet the client accepts. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn proof_a_4000_byte_retained_message_reconnects_forever() { + let connects = connects_with_push(4000, Duration::from_secs(5)).await; + eprintln!("connects in 5 s: {connects}"); + assert!( + connects >= 3, + "expected a reconnect loop, saw {connects} connects" + ); +}