From af818c1cc4c6ec1a2561394a5ce08979f6805f98 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alexander=20Schn=C3=B6rch?= Date: Sun, 4 Oct 2026 20:08:21 +0000 Subject: [PATCH] =?UTF-8?q?feat(mqtt):=20pull=20embedded=20outbound=20publ?= =?UTF-8?q?ishes=20from=20OutboundRoutes=20(design=20054=20=C2=A74.7)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- aimdb-core/src/outbound/routes.rs | 28 +- aimdb-mqtt-connector/src/connector.rs | 41 +- aimdb-mqtt-connector/src/embedded/manager.rs | 10 +- aimdb-mqtt-connector/src/embedded/mod.rs | 327 ++++++-------- aimdb-mqtt-connector/src/embedded/session.rs | 11 +- .../src/embedded/session_loop.rs | 422 ++++++++++++------ aimdb-mqtt-connector/src/embedded/tls.rs | 11 +- .../src/embedded/write_ring.rs | 8 +- aimdb-mqtt-connector/src/lib.rs | 3 + aimdb-mqtt-connector/src/publish_opts.rs | 116 +++++ aimdb-mqtt-connector/tests/session_loop.rs | 97 +++- aimdb-mqtt-connector/tests/tokio_broker.rs | 131 ++++++ .../tests/write_ring_proofs.rs | 6 +- aimdb-tokio-adapter/tests/outbound_routes.rs | 11 + 14 files changed, 844 insertions(+), 378 deletions(-) create mode 100644 aimdb-mqtt-connector/src/publish_opts.rs diff --git a/aimdb-core/src/outbound/routes.rs b/aimdb-core/src/outbound/routes.rs index 9f06061d..871e7cef 100644 --- a/aimdb-core/src/outbound/routes.rs +++ b/aimdb-core/src/outbound/routes.rs @@ -72,8 +72,12 @@ impl OutboundPayload<'_> { /// Values taken from one route's buffer, by outcome. #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub struct RouteStats { - /// Staged for the connector. + /// Staged and handed to the connector, including any it then rejected. pub sent: u64, + /// Handed to the connector, which could not send them (for example, + /// larger than its transport accepts). Reported with + /// [`OutboundRoutes::reject`]. + pub rejected: u64, /// Missed because the reader fell behind. pub lagged: u64, /// Skipped: the written topic did not fit. @@ -358,6 +362,15 @@ impl OutboundRoutes { }) } + /// Count a message from route `id` that the connector took but could not + /// send. The connector logs why; this keeps the count beside the route's + /// other outcomes. + pub fn reject(&mut self, id: RouteId) { + if let Some(stats) = self.stats.get_mut(id) { + stats.rejected += 1; + } + } + /// [`poll_stage`](Self::poll_stage), then [`take_staged`](Self::take_staged), /// for hand-written `poll` code. pub fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll>> { @@ -394,4 +407,17 @@ mod tests { let moved = payload.into_vec(); assert_eq!(moved.as_ptr(), ptr, "moved, not copied"); } + + #[tokio::test] + async fn reject_counts_beside_the_routes_other_outcomes() { + let (db, _runner) = crate::AimDbBuilder::new() + .runtime(Arc::new(crate::executor::test_support::NoopRuntimeOps)) + .build() + .await + .expect("empty database"); + let mut routes = OutboundRoutes::new(&db, "mqtt").unwrap(); + // No routes: an unknown id is ignored rather than a panic. + routes.reject(0); + assert_eq!(routes.stats(0), None); + } } diff --git a/aimdb-mqtt-connector/src/connector.rs b/aimdb-mqtt-connector/src/connector.rs index c410350b..a6b56fde 100644 --- a/aimdb-mqtt-connector/src/connector.rs +++ b/aimdb-mqtt-connector/src/connector.rs @@ -47,6 +47,7 @@ pub struct Native; #[cfg(feature = "embedded")] pub struct Embedded { pub(crate) dialer: D, + pub(crate) write_buffer: usize, } /// The `mountain-mqtt` backend over `embedded-tls`, on the same @@ -55,6 +56,7 @@ pub struct Embedded { pub struct EmbeddedTls { pub(crate) dialer: D, pub(crate) options: crate::embedded::TlsSlot, + pub(crate) write_buffer: usize, } /// An MQTT connector over the backend `B`. @@ -90,7 +92,10 @@ impl MqttConnector { client_id: self.client_id, credentials: self.credentials, keep_alive: self.keep_alive, - backend: Embedded { dialer }, + backend: Embedded { + dialer, + write_buffer: crate::embedded::DEFAULT_WRITE_BUFFER, + }, } } @@ -110,6 +115,7 @@ impl MqttConnector { backend: EmbeddedTls { dialer, options: crate::embedded::TlsSlot::new(options), + write_buffer: crate::embedded::DEFAULT_WRITE_BUFFER, }, } } @@ -143,6 +149,38 @@ impl MqttConnector { } } +/// The write buffer's documentation, shared by both embedded backends. +#[cfg(feature = "embedded")] +macro_rules! write_buffer_doc { + () => { + "Size the session's write ring, in bytes (default 4,096). Allocated once \ + and reused across reconnects.\n\n\ + An outbound PUBLISH frame plus a 64-byte reserve must fit in half the \ + ring (1,984 bytes of frame at the default). `build()` fails for a \ + route whose largest frame does not fit, and for a CONNECT or \ + SUBSCRIBE that does not; an owned payload over the limit at runtime \ + is skipped and counted as rejected in the route's `RouteStats`." + }; +} + +#[cfg(feature = "embedded")] +impl MqttConnector> { + #[doc = write_buffer_doc!()] + pub fn with_write_buffer(mut self, bytes: usize) -> Self { + self.backend.write_buffer = bytes; + self + } +} + +#[cfg(feature = "embedded-tls")] +impl MqttConnector> { + #[doc = write_buffer_doc!()] + pub fn with_write_buffer(mut self, bytes: usize) -> Self { + self.backend.write_buffer = bytes; + self + } +} + /// Whole seconds for the wire, or the reason this keep-alive cannot be used. fn keep_alive_secs(keep_alive: Duration) -> DbResult { let secs = keep_alive.as_secs(); @@ -228,6 +266,7 @@ where credentials, keep_alive_secs, &self.dialer, + self.write_buffer, ) } } diff --git a/aimdb-mqtt-connector/src/embedded/manager.rs b/aimdb-mqtt-connector/src/embedded/manager.rs index 12fa5852..ec0259f7 100644 --- a/aimdb-mqtt-connector/src/embedded/manager.rs +++ b/aimdb-mqtt-connector/src/embedded/manager.rs @@ -1,19 +1,13 @@ -//! Session cadence and the channel a session takes actions from. +//! Session cadence and the reasons a session ends. //! -//! The channel uses `CriticalSectionRawMutex`, so it is `Sync` and the sink -//! needs no force-`Send` wrapper. Time comes from core's +//! Time comes from core's //! [`aimdb_core::session::Delay`], so nothing here names an executor. use core::time::Duration; use aimdb_core::RuntimeOps; -use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex; -use embassy_sync::channel::Channel; use mountain_mqtt::client::ClientError; -/// The action channel: `pump_sink` to broker session. -pub(crate) type ActionChannel = Channel; - /// Monotonic milliseconds. Only differences are meaningful. pub(crate) fn now_ms(runtime: &dyn RuntimeOps) -> u64 { runtime.now_nanos() / 1_000_000 diff --git a/aimdb-mqtt-connector/src/embedded/mod.rs b/aimdb-mqtt-connector/src/embedded/mod.rs index 5933aa9e..ef958e89 100644 --- a/aimdb-mqtt-connector/src/embedded/mod.rs +++ b/aimdb-mqtt-connector/src/embedded/mod.rs @@ -1,9 +1,9 @@ //! The `mountain-mqtt` backend: broker session plus the data-plane bridges. //! -//! Outbound publishes ride core's [`pump_sink`] — the action channel is -//! `Sync`, so nothing force-`Send` stands between it and the runner. Inbound -//! publishes are dispatched into their records by the session loop itself, -//! through an [`InboundDispatch`](aimdb_core::InboundDispatch). +//! The session task drives both directions itself: it pulls outbound messages +//! from an [`OutboundRoutes`](aimdb_core::OutboundRoutes) when it can send, +//! and dispatches inbound publishes into their records through an +//! [`InboundDispatch`](aimdb_core::InboundDispatch). Core runs no pump for it. //! //! See the crate docs for a usage example. @@ -25,8 +25,6 @@ pub mod tls; extern crate alloc; use aimdb_core::connector::ConnectorUrl; -use aimdb_core::session::pump_sink; -use aimdb_core::transport::{ConnectorConfig, PublishError}; use alloc::boxed::Box; use alloc::format; use alloc::string::{String, ToString}; @@ -39,18 +37,19 @@ use core::pin::Pin; use aimdb_embassy_adapter::connectors::into_box_future; use mountain_mqtt::client::ConnectionSettings; -use mountain_mqtt::data::quality_of_service::QualityOfService; use crate::embedded::manager::Settings; +use crate::embedded::session_loop::{connect_packet, publish_frame_len, subscribe_len}; +use crate::embedded::write_ring::{encoded_len, fits_ring, CONTROL_RESERVE}; +use crate::publish_opts::PublishOpts; + +pub(crate) use crate::embedded::write_ring::DEFAULT_WRITE_BUFFER; #[cfg(feature = "embedded-tls")] pub use crate::embedded::tls::TlsOptions; #[cfg(feature = "embedded-tls")] use crate::embedded::tls::{host_ip_literal, READ_BUF_MIN, WRITE_BUF_MIN}; -/// Maximum number of pending MQTT actions -pub(crate) const CHANNEL_SIZE: usize = 32; - /// Buffer size for MQTT packets (4KB) pub(crate) const BUFFER_SIZE: usize = 4096; @@ -62,78 +61,6 @@ pub(crate) const MAX_PROPERTIES: usize = 32; /// The runner's collected future type. type EmbassyBoxFuture = Pin + Send + 'static>>; -/// What a transport's setup hands back: the action channel `pump_sink` rides, -/// plus the tasks that serve it. -type ManagerSetup = (Arc, Vec); - -/// Outbound publishes and subscriptions: pumps to broker session. -pub(crate) type ActionChannel = - crate::embedded::manager::ActionChannel; - -/// What the pumps ask the session to put on the wire. -/// -/// The session encodes each of these itself against the MQTT client state, so -/// an action is data rather than a call (`session_loop::perform`). -#[derive(Clone)] -pub enum AimdbMqttAction { - /// Publish a message to a topic - Publish { - topic: String, - payload: Vec, - qos: QualityOfService, - retain: bool, - }, - /// Subscribe to a topic - Subscribe { - topic: String, - qos: QualityOfService, - }, -} - -// =========================================================================== -// Data-plane bridge — `pump_sink` drives it directly. The action channel is -// `Sync` (its mutex is `CriticalSectionRawMutex`), so no force-`Send` wrapper -// stands between it and the runner. -// =========================================================================== - -/// Turns a `pump_sink` publish into an `AimdbMqttAction::Publish` on the -/// session's action channel. -struct MqttSink { - actions: Arc, -} - -impl aimdb_core::transport::Connector for MqttSink { - fn publish( - &self, - destination: &str, - config: &ConnectorConfig, - payload: &[u8], - ) -> Pin> + Send + '_>> { - // `qos`/`retain` arrive via the URL query (passed through in - // `protocol_options`); default to QoS 1 (legacy behaviour), no retain. - let qos = match opt_u8(config, "qos").map(map_qos) { - Some(Ok(qos)) => qos, - Some(Err(e)) => return Box::pin(async move { Err(e) }), - None => QualityOfService::Qos1, - }; - let retain = opt_bool(config, "retain").unwrap_or(false); - let topic = destination.to_string(); - let payload = payload.to_vec(); - - Box::pin(async move { - self.actions - .send(AimdbMqttAction::Publish { - topic, - payload, - qos, - retain, - }) - .await; - Ok(()) - }) - } -} - /// Force-`Send + Sync` slot for the TLS materials: [`TlsOptions`] holds /// `&'static mut` exclusive resources, so it is neither `Sync` nor takeable /// through the `&self` that [`ConnectorBuilder::build`] receives. @@ -148,6 +75,7 @@ pub(crate) fn build_plain<'a, D>( credentials: Option<&'a (String, String)>, keep_alive_secs: u16, dialer: &'a D, + write_buffer: usize, ) -> Pin>> + Send + 'a>> where D: aimdb_core::session::StreamDialer @@ -160,24 +88,28 @@ where Box::pin(async move { let inbound = aimdb_core::InboundDispatch::new(db, "mqtt", &crate::MqttGrammar)?; let topics = inbound_topics(&inbound); - warn_unsupported_qos(db); let broker = parse_broker_url(broker_url)?; if broker.tls { return Err(build_err("mqtts:// broker URLs require .tls(...)")); } let connection_settings = static_connection_settings(client_id, credentials, broker.credentials.as_ref()); + let settings = Settings::from_keep_alive_secs(keep_alive_secs); + let (outbound, opts) = + prepare_outbound(db, write_buffer, &settings, &connection_settings, &topics)?; - let (actions, manager_tasks) = setup_manager( + setup_manager( &broker, connection_settings, dialer.clone(), topics, inbound, - Settings::from_keep_alive_secs(keep_alive_secs), + outbound, + opts, + write_buffer, + settings, db.runtime_ops(), - )?; - Ok(collect_pumps(db, actions, manager_tasks)) + ) }) } @@ -202,7 +134,6 @@ where Box::pin(async move { let inbound = aimdb_core::InboundDispatch::new(db, "mqtt", &crate::MqttGrammar)?; let topics = inbound_topics(&inbound); - warn_unsupported_qos(db); let broker = parse_broker_url(broker_url)?; if !broker.tls { return Err(build_err(".tls(...) requires an mqtts:// broker URL")); @@ -213,18 +144,28 @@ where .ok_or_else(|| build_err("TLS materials already taken; build() ran twice"))?; let connection_settings = static_connection_settings(client_id, credentials, broker.credentials.as_ref()); + let settings = Settings::from_keep_alive_secs(keep_alive_secs); + let (outbound, opts) = prepare_outbound( + db, + backend.write_buffer, + &settings, + &connection_settings, + &topics, + )?; - let (actions, manager_tasks) = setup_tls_manager( + setup_tls_manager( &broker, options, connection_settings, backend.dialer.clone(), topics, inbound, - Settings::from_keep_alive_secs(keep_alive_secs), + outbound, + opts, + backend.write_buffer, + settings, db.runtime_ops(), - )?; - Ok(collect_pumps(db, actions, manager_tasks)) + ) }) } @@ -242,16 +183,76 @@ fn inbound_topics(inbound: &aimdb_core::InboundDispatch) -> Vec { topics } -/// Outbound publishes ride core's `pump_sink`; the session tasks, which also -/// dispatch inbound publishes, join them. -fn collect_pumps( +/// Build the outbound routes and parse each route's options, and check that +/// every packet the session must be able to send fits the write ring: the +/// largest PUBLISH of every route, the CONNECT and each SUBSCRIBE. One that +/// does not would fail every session, and the connector would reconnect +/// forever. Warns once per route asking for `qos=2`, which this client sends +/// at QoS 1. +fn prepare_outbound( db: &aimdb_core::builder::AimDb, - actions: Arc, - manager_tasks: Vec, -) -> Vec { - let mut futures = pump_sink(db, "mqtt", Arc::new(MqttSink { actions })); - futures.extend(manager_tasks); - futures + write_buffer: usize, + settings: &Settings, + connection_settings: &ConnectionSettings<'static>, + topics: &[String], +) -> Result<(aimdb_core::OutboundRoutes, Vec), aimdb_core::DbError> { + let outbound = aimdb_core::OutboundRoutes::new(db, "mqtt")?; + let size = |len: Result| { + len.map_err(|_| build_err("a packet could not be sized")) + }; + let mut problems: Vec = Vec::new(); + let mut opts = Vec::with_capacity(outbound.routes().len()); + + for route in outbound.routes() { + match PublishOpts::parse(route) { + Ok(opt) => { + if opt.qos == 2 { + aimdb_core::log_warn!( + "MQTT: route '{}' asks for qos=2; this backend publishes it at QoS 1 (at-least-once). The std backend honours qos=2 on the same URL.", + route.default_topic + ); + #[cfg(feature = "defmt")] + defmt::warn!( + "MQTT: route '{}' asks qos=2; publishing at QoS 1 (at-least-once)", + &*route.default_topic + ); + } + opts.push(opt); + } + Err(e) => problems.push(e), + } + let topic_len = route.default_topic.len().max(route.topic_capacity); + let frame = size(publish_frame_len(topic_len, route.payload_capacity))?; + if !fits_ring(write_buffer, frame, CONTROL_RESERVE) { + problems.push(format!( + "route '{}': its largest PUBLISH is {frame} bytes, which needs a write buffer of at least {} bytes; it is {write_buffer}", + route.default_topic, + 2 * (frame + CONTROL_RESERVE) + )); + } + } + + let connect = size(encoded_len(&connect_packet(settings, connection_settings)))?; + if !fits_ring(write_buffer, connect, 0) { + problems.push(format!( + "the CONNECT (client id and credentials) is {connect} bytes, which needs a write buffer of at least {} bytes; it is {write_buffer}", + 2 * connect + )); + } + for topic in topics { + let subscribe = size(subscribe_len(topic))?; + if !fits_ring(write_buffer, subscribe, 0) { + problems.push(format!( + "the SUBSCRIBE to '{topic}' is {subscribe} bytes, which needs a write buffer of at least {} bytes; it is {write_buffer}", + 2 * subscribe + )); + } + } + + if !problems.is_empty() { + return Err(build_err(&problems.join("; "))); + } + Ok((outbound, opts)) } /// Parsed broker endpoint: transport + authority. @@ -335,18 +336,21 @@ fn static_connection_settings( } } -/// Set up the plain-TCP broker session loop, returning the action channel -/// (outbound), the event channel (inbound), and the task future. Synchronous — -/// no `.await` — so the caller's `build` future stays `Send`. +/// Set up the plain-TCP broker session task, which drives both directions. +/// Synchronous — no `.await` — so the caller's `build` future stays `Send`. +#[allow(clippy::too_many_arguments)] fn setup_manager( broker: &BrokerUrl, connection_settings: ConnectionSettings<'static>, dialer: D, topics: Vec, inbound: aimdb_core::InboundDispatch, + outbound: aimdb_core::OutboundRoutes, + opts: Vec, + write_buffer: usize, settings: Settings, runtime: Arc, -) -> Result +) -> aimdb_core::DbResult> where D: aimdb_core::session::StreamDialer + aimdb_core::session::Delay @@ -355,18 +359,16 @@ where + Sync + 'static, { - let actions: Arc = Arc::new(ActionChannel::new()); - // The dialer is both the transport and the clock the session runs on. let host = broker.host.clone(); let port = broker.port; // SAFETY: every value the session holds is `Send` — `StreamDialer` - // guarantees `Stream: Send`, the channels are `CriticalSectionRawMutex` - // and the state cell is a blocking mutex. See `SendSession`. + // guarantees `Stream: Send`, `InboundDispatch` and `OutboundRoutes` are + // `Send`, and the session's channel is `CriticalSectionRawMutex`. See + // `SendSession`. let manager_task: EmbassyBoxFuture = Box::pin(unsafe { crate::embedded::session::SendSession::new({ - let actions = actions.clone(); async move { #[cfg(feature = "defmt")] defmt::info!("MQTT background task starting"); @@ -379,7 +381,9 @@ where connection_settings, settings, inbound, - actions, + outbound, + opts, + write_buffer, runtime, ) .await @@ -387,7 +391,7 @@ where }) }); - Ok((actions, alloc::vec![manager_task])) + Ok(alloc::vec![manager_task]) } /// Set up the TLS broker manager ([`run_tls`]) plus the SNTP time-source task. @@ -401,9 +405,12 @@ fn setup_tls_manager( dialer: D, topics: Vec, inbound: aimdb_core::InboundDispatch, + outbound: aimdb_core::OutboundRoutes, + opts: Vec, + write_buffer: usize, settings: Settings, runtime: Arc, -) -> Result +) -> aimdb_core::DbResult> where D: aimdb_core::session::StreamDialer + aimdb_core::session::Delay @@ -439,8 +446,6 @@ where )); } - let actions: Arc = Arc::new(ActionChannel::new()); - let host = broker.host.clone(); let port = broker.port; #[cfg(feature = "embassy-tls")] @@ -448,12 +453,11 @@ where let delay = dialer.clone(); // SAFETY: as for the plain path — `StreamDialer` guarantees `Stream: Send`, - // the channels are `CriticalSectionRawMutex`, and `TlsOptions` is `Send` - // (its RNG carries the bound). See `session::SendSession`. + // `InboundDispatch` and `OutboundRoutes` are `Send`, and `TlsOptions` is + // `Send` (its RNG carries the bound). See `session::SendSession`. #[cfg_attr(not(feature = "embassy-tls"), allow(unused_mut))] let mut tasks: Vec = alloc::vec![Box::pin(unsafe { crate::embedded::session::SendSession::new({ - let actions = actions.clone(); async move { #[cfg(feature = "defmt")] defmt::info!("MQTT-TLS background task starting"); @@ -469,7 +473,9 @@ where connection_settings, settings, inbound, - actions, + outbound, + opts, + write_buffer, delay, runtime, ) @@ -490,86 +496,5 @@ where })); } - Ok((actions, tasks)) -} - -/// Map a QoS level to mountain-mqtt's `QualityOfService`. -/// -/// `2` downgrades to 1 — this client implements no exactly-once handshake, -/// where [`Native`](crate::connector::Native) honours the same route URL -/// exactly. [`warn_unsupported_qos`] is what says so, once per route at build. -/// Anything above 2 is rejected, as `Native` rejects it. -fn map_qos(qos: u8) -> Result { - match qos { - 0 => Ok(QualityOfService::Qos0), - 1 => Ok(QualityOfService::Qos1), - 2 => Ok(QualityOfService::Qos1), - _ => Err(PublishError::UnsupportedQoS), - } -} - -/// Name, at build, every outbound route asking for a QoS this backend cannot -/// give. -/// -/// Checked here rather than in [`map_qos`] because `map_qos` runs per publish: -/// warning there would repeat at the route's own rate for the life of the -/// process, and latching it to fire once would hide the message whenever the -/// first publish beats the logger into place. The route set is fixed at build, -/// so once per offending route — naming the route, while the caller is still -/// reading startup output — is both quieter and more use than either. -/// -/// Both facades fire: they are independent, and neither covers the other. -/// `log_warn!` reaches `tracing`/`log` when this backend runs on a host, -/// `defmt` reaches an MCU. -fn warn_unsupported_qos(db: &aimdb_core::builder::AimDb) { - for route in db.collect_outbound_routes("mqtt") { - let asked = route - .config - .iter() - .find(|(k, _)| k == "qos") - .and_then(|(_, v)| v.parse::().ok()); - - if asked == Some(2) { - aimdb_core::log_warn!( - "MQTT: route '{}' asks for qos=2; this backend publishes it at QoS 1 (at-least-once). The std backend honours qos=2 on the same URL.", - route.topic - ); - #[cfg(feature = "defmt")] - defmt::warn!( - "MQTT: route '{}' asks qos=2; publishing at QoS 1 (at-least-once)", - route.topic.as_str() - ); - } - } -} - -/// Read a `u8` option from the per-route `protocol_options` (URL query). -fn opt_u8(config: &ConnectorConfig, key: &str) -> Option { - config - .protocol_options - .iter() - .find(|(k, _)| k == key) - .and_then(|(_, v)| v.parse::().ok()) -} - -/// Read a `bool` option from the per-route `protocol_options` (URL query). -fn opt_bool(config: &ConnectorConfig, key: &str) -> Option { - config - .protocol_options - .iter() - .find(|(k, _)| k == key) - .and_then(|(_, v)| v.parse::().ok()) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_qos_mapping() { - assert!(matches!(map_qos(0), Ok(QualityOfService::Qos0))); - assert!(matches!(map_qos(1), Ok(QualityOfService::Qos1))); - assert!(matches!(map_qos(2), Ok(QualityOfService::Qos1))); // Downgrades to QoS 1 - assert!(matches!(map_qos(99), Err(PublishError::UnsupportedQoS))); // Not a QoS level - } + Ok(tasks) } diff --git a/aimdb-mqtt-connector/src/embedded/session.rs b/aimdb-mqtt-connector/src/embedded/session.rs index 12c8c0b0..b41bec41 100644 --- a/aimdb-mqtt-connector/src/embedded/session.rs +++ b/aimdb-mqtt-connector/src/embedded/session.rs @@ -54,7 +54,9 @@ pub(crate) async fn run_sessions( connection_settings: mountain_mqtt::client::ConnectionSettings<'static>, settings: crate::embedded::manager::Settings, inbound: aimdb_core::InboundDispatch, - actions: alloc::sync::Arc, + mut outbound: aimdb_core::OutboundRoutes, + opts: alloc::vec::Vec, + write_buffer: usize, runtime: alloc::sync::Arc, ) -> ! where @@ -64,7 +66,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}; + use crate::embedded::write_ring::WriteRing; // Built once and borrowed for the loop; re-sent on every connection. let subscribe_topics: alloc::vec::Vec<(&str, QualityOfService)> = topics @@ -73,7 +75,7 @@ where .collect(); // Allocated once and reused by every session. - let ring = WriteRing::new(DEFAULT_WRITE_BUFFER); + let ring = WriteRing::new(write_buffer); loop { let mut stream = match dialer.connect(&host, port).await { @@ -95,7 +97,8 @@ where &connection_settings, &subscribe_topics, &inbound, - &actions, + &mut outbound, + &opts, &ring, &settings, &dialer, diff --git a/aimdb-mqtt-connector/src/embedded/session_loop.rs b/aimdb-mqtt-connector/src/embedded/session_loop.rs index acf4c1a4..b8708a9c 100644 --- a/aimdb-mqtt-connector/src/embedded/session_loop.rs +++ b/aimdb-mqtt-connector/src/embedded/session_loop.rs @@ -17,23 +17,27 @@ use core::task::Poll; use core::time::Duration; use aimdb_core::session::{ByteRead, ByteWrite, Delay}; -use aimdb_core::{InboundDispatch, RuntimeOps}; +use aimdb_core::{InboundDispatch, OutboundRoutes, RouteId, RuntimeOps}; 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::data::property::ConnectProperty; +use mountain_mqtt::data::packet_identifier::{PacketIdentifier, PublishPacketIdentifier}; +use mountain_mqtt::data::property::{ConnackProperty, ConnectProperty, Property}; use mountain_mqtt::data::quality_of_service::QualityOfService; use mountain_mqtt::error::{PacketReadError, PacketWriteError}; use mountain_mqtt::packets::connect::Connect; use mountain_mqtt::packets::packet_generic::PacketGeneric; +use mountain_mqtt::packets::publish::Publish; +use mountain_mqtt::packets::subscribe::{Subscribe, SubscriptionRequest}; use crate::embedded::manager::{now_ms, Error, Settings}; use crate::embedded::packet_reader::PacketReader; use crate::embedded::write_ring::{encoded_len, WriteRing, CONTROL_RESERVE}; -use crate::embedded::{ActionChannel, AimdbMqttAction, BUFFER_SIZE, MAX_PROPERTIES}; +use crate::embedded::{BUFFER_SIZE, MAX_PROPERTIES}; +use crate::publish_opts::PublishOpts; /// Bytes lifted off the socket at a time, and the size of one `inbound` slot. const RX_CHUNK: usize = 256; @@ -60,17 +64,18 @@ type Chunk = heapless::Vec; /// Drive one MQTT session over a split stream until an error ends it. /// -/// Connects, subscribes `subscribe_topics`, then performs actions and -/// dispatches every inbound publish into its records through `dispatch`. -/// Returns only on failure — the caller reconnects. +/// Connects, subscribes `subscribe_topics`, then publishes what `outbound` +/// stages and dispatches every inbound publish into its records through +/// `dispatch`. 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. +/// **At most once**: a message is taken from its record buffer before it is +/// written, so the one in flight when a session ends is lost. Everything still +/// in the record buffers survives (what survives depends on each buffer's +/// type). #[allow(clippy::too_many_arguments)] pub(crate) async fn run_session( rx: R, @@ -78,7 +83,8 @@ pub(crate) async fn run_session( connection_settings: &ConnectionSettings<'static>, subscribe_topics: &[(&str, QualityOfService)], dispatch: &InboundDispatch, - actions: &ActionChannel, + outbound: &mut OutboundRoutes, + opts: &[PublishOpts], ring: &WriteRing, settings: &Settings, delay: &D, @@ -98,7 +104,8 @@ where connection_settings, subscribe_topics, dispatch, - actions, + outbound, + opts, settings, delay, runtime, @@ -147,7 +154,8 @@ async fn client_loop( connection_settings: &ConnectionSettings<'static>, subscribe_topics: &[(&str, QualityOfService)], dispatch: &InboundDispatch, - actions: &ActionChannel, + outbound: &mut OutboundRoutes, + opts: &[PublishOpts], settings: &Settings, delay: &D, runtime: &dyn RuntimeOps, @@ -168,28 +176,17 @@ async fn client_loop( let mut waiting_since: Option = Some(start); let mut connected = false; let mut next_topic = 0usize; + // The broker's Maximum Packet Size, from its CONNACK. Sending a larger + // packet is a protocol error that ends the session. + let mut broker_max: Option = None; + // Latched on `Ready(None)`: every route is closed, or there were none. + // Passing it through would resolve the arm on every iteration and the + // session would never yield. + let mut outbound_done = false; // CONNECT goes out first; its CONNACK is what flips `connected`. { - let mut properties = heapless::Vec::new(); - // Topic aliases are declined: honouring them would mean storing the - // server's topic names for the life of the connection. - let _ = properties.push(ConnectProperty::TopicAliasMaximum(0.into())); - let _ = properties.push(ConnectProperty::MaximumPacketSize( - (MAX_INBOUND_PACKET as u32).into(), - )); - // Ours, not `connection_settings.keep_alive()`: that field has no - // setter, so it is always mountain-mqtt's own 60 s constant. The - // cadence below is derived from the value we actually send. - let connect: Connect<'_, 2, 0> = Connect::new( - settings.keep_alive_secs, - *connection_settings.username(), - *connection_settings.password(), - connection_settings.client_id(), - true, - None, - properties, - ); + let connect = connect_packet(settings, connection_settings); state.connect(&connect).map_err(client_error)?; ring.put(&connect, 0).await?; } @@ -243,30 +240,32 @@ async fn client_loop( // --- park until something happens ---------------------------------- - // The action arm is armed only when a publish can actually be sent: + // The publish 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 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(); + // for the largest PUBLISH plus its reserve. 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. A value leaves its record buffer only when the arm + // resolves, so a losing arm takes nothing. + let publish_ready = connected + && !outbound_done + && !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; + let publish_arm = poll_fn(|cx| { + if !publish_ready || !ring.poll_room(publish_room, cx.waker()) { + return Poll::Pending; } - poll_fn(|cx| { - if ring.poll_room(publish_room, cx.waker()) { - Poll::Ready(()) - } else { + match outbound.poll_stage(cx) { + Poll::Ready(Some(id)) => Poll::Ready(id), + Poll::Ready(None) => { + outbound_done = true; Poll::Pending } - }) - .await; - actions.receive().await - }; + Poll::Pending => Poll::Pending, + } + }); let sleep_for = Duration::from_millis(next_deadline( now, @@ -276,7 +275,7 @@ async fn client_loop( waiting_since.map(|since| since + response_timeout), )); - match select3(inbound.receive(), action_arm, delay.sleep(sleep_for)).await { + match select3(inbound.receive(), publish_arm, delay.sleep(sleep_for)).await { Either3::First(chunk) => { reader.feed(&chunk).map_err(client_error)?; drain_packets( @@ -287,11 +286,12 @@ async fn client_loop( runtime, &mut last_ack_ms, &mut connected, + &mut broker_max, ) .await?; } - Either3::Second(action) => { - perform(action, &mut state, ring).await?; + Either3::Second(id) => { + publish_staged(outbound, id, opts, &mut state, ring, broker_max).await?; } // The timer fired: the top of the loop re-evaluates every deadline. Either3::Third(()) => {} @@ -316,6 +316,7 @@ async fn drain_packets( runtime: &dyn RuntimeOps, last_ack_ms: &mut u64, connected: &mut bool, + broker_max: &mut Option, ) -> 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 @@ -331,6 +332,13 @@ async fn drain_packets( let packet: PacketGeneric<'_, MAX_PROPERTIES, 0, 0> = reader.parse(total).map_err(client_error)?; + if let PacketGeneric::Connack(connack) = &packet { + *broker_max = connack.properties().iter().find_map(|p| match p { + ConnackProperty::MaximumPacketSize(max) => Some(max.value() as usize), + _ => None, + }); + } + // Produce the PUBACK before the state update, as upstream does, so // the two cannot disagree about what was acknowledged. It borrows // the client state, so it goes into the ring before @@ -407,76 +415,129 @@ fn deliver( Ok(()) } -/// Turn one queued action into a packet on the wire. +/// Publish the message `outbound` staged for route `id`. /// /// Sent before the state update, as upstream does: a state that believes a -/// publish is in flight when it is not parks the action arm forever. -async fn perform( - action: AimdbMqttAction, +/// publish is in flight when it is not parks the publish arm forever. A frame +/// the ring can never grant, or larger than the broker accepts, is skipped +/// before the state commits to it, logged, and counted as rejected in the +/// route's `RouteStats`. +async fn publish_staged( + outbound: &mut OutboundRoutes, + id: RouteId, + opts: &[PublishOpts], state: &mut ClientStateNoQueue, ring: &WriteRing, + broker_max: Option, ) -> Result<(), Error> { - match action { - AimdbMqttAction::Publish { - topic, - payload, - qos, - retain, - } => { + let opt = opts.get(id).copied().unwrap_or(PublishOpts { + qos: 1, + retain: false, + }); + // QoS 2 is not supported by this client; build() warned once per route. + let qos = if opt.qos == 0 { + QualityOfService::Qos0 + } else { + QualityOfService::Qos1 + }; + let rejected = { + let Some(msg) = outbound.take_staged() else { + return Ok(()); + }; + let packet = state + .publish_packet(msg.topic, msg.payload.as_slice(), qos, opt.retain) + .map_err(client_error)?; + let len = encoded_len(&packet)?; + let over_broker = broker_max.is_some_and(|max| len > max); + if !ring.fits(len, CONTROL_RESERVE) || over_broker { + aimdb_core::log_warn!( + "MQTT: skipping publish to '{}': {} bytes exceed the {} limit", + msg.topic, + len, + if over_broker { + "broker's" + } else { + "write ring's" + } + ); #[cfg(feature = "defmt")] - defmt::debug!( - "Publishing {} bytes to {} (QoS={:?})", - payload.len(), - topic.as_str(), - qos + defmt::warn!( + "MQTT: skipping publish to {}: {} bytes exceed the {} limit", + msg.topic, + len, + if over_broker { + "broker's" + } else { + "write ring's" + } ); - let packet = state - .publish_packet(&topic, &payload, qos, retain) - .inspect_err(|_e| { - // The action is already off the channel, so a failure here - // loses this message and ends the session — say which. - #[cfg(feature = "defmt")] - defmt::warn!( - "MQTT: dropping publish of {} bytes to {}: {}", - payload.len(), - topic.as_str(), - _e - ); - }) - .map_err(client_error)?; - // 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 !ring.fits(len, CONTROL_RESERVE) { - #[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(()); - } + true + } else { ring.put_sized(&packet, len, CONTROL_RESERVE).await?; state.publish_update(&packet).map_err(client_error)?; + false } - AimdbMqttAction::Subscribe { topic, qos } => { - #[cfg(feature = "defmt")] - defmt::info!("Subscribing to {} (QoS={:?})", topic.as_str(), qos); - let packet = state - .subscribe_packet(&topic, qos) - .inspect_err(|_e| { - #[cfg(feature = "defmt")] - defmt::warn!("MQTT: dropping subscribe to {}: {}", topic.as_str(), _e); - }) - .map_err(client_error)?; - ring.put(&packet, 0).await?; - state.subscribe_update(&packet).map_err(client_error)?; - } + }; + if rejected { + outbound.reject(id); } Ok(()) } +/// The CONNECT this session sends, also sized at build. +pub(crate) fn connect_packet<'a>( + settings: &Settings, + connection_settings: &'a ConnectionSettings<'static>, +) -> Connect<'a, 2, 0> { + let mut properties = heapless::Vec::new(); + // Topic aliases are declined: honouring them would mean storing the + // server's topic names for the life of the connection. + let _ = properties.push(ConnectProperty::TopicAliasMaximum(0.into())); + let _ = properties.push(ConnectProperty::MaximumPacketSize( + (MAX_INBOUND_PACKET as u32).into(), + )); + // Ours, not `connection_settings.keep_alive()`: that field has no setter, + // so it is always mountain-mqtt's own 60 s constant. The cadence is + // derived from the value we actually send. + Connect::new( + settings.keep_alive_secs, + *connection_settings.username(), + *connection_settings.password(), + connection_settings.client_id(), + true, + None, + properties, + ) +} + +/// Bytes the SUBSCRIBE for `topic` encodes to. +pub(crate) fn subscribe_len(topic: &str) -> Result { + let packet: Subscribe<'_, 0, 0> = Subscribe::new( + PacketIdentifier(1), + SubscriptionRequest::new(topic, QualityOfService::Qos1), + heapless::Vec::new(), + heapless::Vec::new(), + ); + encoded_len(&packet) +} + +/// Bytes a QoS 1 PUBLISH with a `topic_len`-byte topic and a +/// `payload_len`-byte payload encodes to: the largest frame a route with +/// those capacities produces. +pub(crate) fn publish_frame_len(topic_len: usize, payload_len: usize) -> Result { + let topic = "x".repeat(topic_len); + let payload = alloc::vec![0u8; payload_len]; + let packet: Publish<'_, 0> = Publish::new( + false, + false, + &topic, + PublishPacketIdentifier::Qos1(PacketIdentifier(1)), + &payload, + heapless::Vec::new(), + ); + encoded_len(&packet) +} + /// Milliseconds to sleep before the earliest armed deadline. fn next_deadline( now: u64, @@ -554,7 +615,6 @@ mod tests { /// absorb codegen drift, but not loose enough to fit another buffer. #[test] fn the_session_future_has_not_outgrown_the_loop_it_replaced() { - let actions = ActionChannel::new(); let settings = Settings::default(); let connection_settings = ConnectionSettings::unauthenticated("size-probe"); let runtime = aimdb_core::executor::test_support::NoopRuntimeOps; @@ -566,6 +626,7 @@ mod tests { ) .expect("empty database"); let dispatch = InboundDispatch::new(&db, "mqtt", &crate::MqttGrammar).expect("no links"); + let mut outbound = OutboundRoutes::new(&db, "mqtt").expect("no links"); // Built, never polled: `size_of_val` on the future is the whole point. let session = run_session( @@ -574,7 +635,8 @@ mod tests { &connection_settings, &[], &dispatch, - &actions, + &mut outbound, + &[], &ring, &settings, &NullDelay, @@ -739,15 +801,14 @@ mod tests { #[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. + //! Where the embedded backend's size limits sit: the largest PUBLISH it + //! sends, what it does with a larger one (skipped and counted as rejected + //! in the route's `RouteStats`), and the largest packet it receives. use super::*; use crate::embedded::write_ring::DEFAULT_WRITE_BUFFER; + use alloc::boxed::Box; + use alloc::sync::Arc; 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; @@ -779,50 +840,123 @@ mod proofs { .unwrap() } - fn publish(n: usize) -> AimdbMqttAction { - AimdbMqttAction::Publish { - topic: "t".into(), - payload: alloc::vec![b'x'; n], - qos: QualityOfService::Qos1, - retain: false, + /// Lets `link_to("mqtt://…")` register; drives nothing. + struct NoTransport; + + impl aimdb_core::connector::ConnectorBuilder for NoTransport { + #[allow(clippy::type_complexity)] + fn build<'a>( + &'a self, + _db: &'a aimdb_core::AimDb, + ) -> core::pin::Pin< + Box< + dyn core::future::Future< + Output = aimdb_core::DbResult< + Vec + Send>>>, + >, + > + Send + + 'a, + >, + > { + Box::pin(async { Ok(Vec::new()) }) + } + fn scheme(&self) -> &str { + "mqtt" } } - 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())) + /// One route to `t` whose owned serializer emits as many bytes as the + /// value says, with a message of `payload_len` bytes staged. + async fn staged(payload_len: usize) -> (aimdb_core::AimDb, OutboundRoutes, RouteId) { + use aimdb_core::buffer::BufferCfg; + use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; + + let mut builder = aimdb_core::AimDbBuilder::new() + .runtime(Arc::new(TokioAdapter)) + .with_connector(NoTransport); + builder.configure::("blob", |reg| { + reg.buffer(BufferCfg::SingleLatest) + .link_to("mqtt://t") + .with_serializer(|_ctx, n: &usize| Ok(alloc::vec![b'x'; *n])) + .finish(); + }); + let (db, _runner) = builder.build().await.expect("build"); + let mut outbound = OutboundRoutes::new(&db, "mqtt").expect("routes"); + db.produce("blob", payload_len).expect("produce"); + let id = poll_fn(|cx| outbound.poll_stage(cx)) + .await + .expect("route open"); + (db, outbound, id) } - #[test] - fn proof_a_1984_byte_publish_goes_out() { + const QOS1: [PublishOpts; 1] = [PublishOpts { + qos: 1, + retain: false, + }]; + + #[tokio::test] + async fn 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(())) - )); + let (_db, mut outbound, id) = staged(payload_for(1984)).await; + publish_staged(&mut outbound, id, &QOS1, &mut state, &ring, None) + .await + .unwrap(); assert!(!ring.has_room(DEFAULT_WRITE_BUFFER), "bytes were queued"); assert!(state.waiting_for_responses(), "QoS 1 publish in flight"); + assert_eq!(outbound.stats(id).unwrap().rejected, 0); } - #[test] - fn proof_a_1985_byte_publish_vanishes_without_an_error() { + #[tokio::test] + async fn a_1985_byte_publish_is_skipped_and_counted() { 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(())) - )); + let (_db, mut outbound, id) = staged(payload_for(1985)).await; + publish_staged(&mut outbound, id, &QOS1, &mut state, &ring, None) + .await + .unwrap(); 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. + let stats = outbound.stats(id).unwrap(); + assert_eq!((stats.sent, stats.rejected), (1, 1)); + } + + #[tokio::test] + async fn a_publish_over_the_brokers_maximum_packet_size_is_skipped_and_counted() { + let ring = WriteRing::new(DEFAULT_WRITE_BUFFER); + let mut state = connected(); + let (_db, mut outbound, id) = staged(payload_for(200)).await; + publish_staged(&mut outbound, id, &QOS1, &mut state, &ring, Some(199)) + .await + .unwrap(); + assert!(ring.has_room(DEFAULT_WRITE_BUFFER), "nothing was queued"); + assert_eq!(outbound.stats(id).unwrap().rejected, 1); + + // At the limit it goes out. + let (_db, mut outbound, id) = staged(payload_for(200)).await; + publish_staged(&mut outbound, id, &QOS1, &mut state, &ring, Some(200)) + .await + .unwrap(); + assert_eq!(outbound.stats(id).unwrap().rejected, 0); + assert!(state.waiting_for_responses()); + } + + #[test] + fn publish_frame_len_matches_what_the_client_state_encodes() { + let mut state = connected(); + for (topic_len, payload_len) in [(1, 0), (1, 100), (30, 1900), (200, 3000)] { + let topic = "t".repeat(topic_len); + let payload = alloc::vec![0u8; payload_len]; + let packet = state + .publish_packet(&topic, &payload, QualityOfService::Qos1, true) + .unwrap(); + assert_eq!( + publish_frame_len(topic_len, payload_len).unwrap(), + encoded_len(&packet).unwrap(), + "{topic_len}-byte topic, {payload_len}-byte payload" + ); + } } /// A QoS 0 PUBLISH to `t` that is `total` bytes on the wire. diff --git a/aimdb-mqtt-connector/src/embedded/tls.rs b/aimdb-mqtt-connector/src/embedded/tls.rs index c98f77f3..c2c58f3e 100644 --- a/aimdb-mqtt-connector/src/embedded/tls.rs +++ b/aimdb-mqtt-connector/src/embedded/tls.rs @@ -24,7 +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 crate::embedded::write_ring::WriteRing; use mountain_mqtt::client::ConnectionSettings; use mountain_mqtt::data::quality_of_service::QualityOfService; @@ -322,7 +322,9 @@ pub(crate) async fn run_tls( connection_settings: ConnectionSettings<'static>, settings: Settings, inbound: aimdb_core::InboundDispatch, - actions: Arc, + mut outbound: aimdb_core::OutboundRoutes, + opts: Vec, + write_buffer: usize, delay: D, runtime: Arc, ) -> ! @@ -345,7 +347,7 @@ where .collect(); // Allocated once and reused by every session. - let ring = WriteRing::new(DEFAULT_WRITE_BUFFER); + let ring = WriteRing::new(write_buffer); loop { // Certificate validity needs real time. Take it from the runtime when @@ -413,7 +415,8 @@ where &connection_settings, &subscribe_topics, &inbound, - &actions, + &mut outbound, + &opts, &ring, &settings, &delay, diff --git a/aimdb-mqtt-connector/src/embedded/write_ring.rs b/aimdb-mqtt-connector/src/embedded/write_ring.rs index 36eb0003..d94fe078 100644 --- a/aimdb-mqtt-connector/src/embedded/write_ring.rs +++ b/aimdb-mqtt-connector/src/embedded/write_ring.rs @@ -58,7 +58,7 @@ impl WriteRing { /// 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 + fits_ring(self.queue.capacity(), len, reserve) } /// Whether a contiguous grant of `n` bytes exists right now. @@ -189,6 +189,12 @@ impl WriteRing { } } +/// Whether a `len`-byte frame plus `reserve` always fits eventually in a +/// `capacity`-byte ring; see [`WriteRing::fits`]. +pub(crate) fn fits_ring(capacity: usize, len: usize, reserve: usize) -> bool { + len + reserve <= capacity / 2 +} + /// Bytes `packet` encodes to. pub(crate) fn encoded_len(packet: &P) -> Result { let mut len_writer = MqttLenWriter::new(); diff --git a/aimdb-mqtt-connector/src/lib.rs b/aimdb-mqtt-connector/src/lib.rs index b5d1389f..dbe41722 100644 --- a/aimdb-mqtt-connector/src/lib.rs +++ b/aimdb-mqtt-connector/src/lib.rs @@ -127,6 +127,9 @@ pub mod native; #[cfg(feature = "embedded")] pub mod embedded; +#[cfg(feature = "embedded")] +pub(crate) mod publish_opts; + // SNTP wire codec — pure and feature-independent so it is unit-tested on the // host; only the TLS I/O task consumes it. #[cfg_attr(not(feature = "embassy-tls"), allow(dead_code))] diff --git a/aimdb-mqtt-connector/src/publish_opts.rs b/aimdb-mqtt-connector/src/publish_opts.rs new file mode 100644 index 00000000..4c992c04 --- /dev/null +++ b/aimdb-mqtt-connector/src/publish_opts.rs @@ -0,0 +1,116 @@ +//! Per-route publish options, parsed once when the connector builds. + +use alloc::format; +use alloc::string::String; + +use aimdb_core::RouteInfo; + +/// How every message of one outbound route is published. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct PublishOpts { + /// 0, 1 or 2, from `?qos=`. Defaults to 1. + pub qos: u8, + /// From `?retain=`. Defaults to `false`. + pub retain: bool, +} + +impl PublishOpts { + /// Parse `route`'s `qos` and `retain`. A value that does not parse is an + /// error naming the route, not a silent fallback to the default. + pub(crate) fn parse(route: &RouteInfo) -> Result { + let option = |key: &str| { + route + .config + .protocol_options + .iter() + .find(|(k, _)| k == key) + .map(|(_, v)| v.as_str()) + }; + let qos = match option("qos") { + None => 1, + Some("0") => 0, + Some("1") => 1, + Some("2") => 2, + Some(other) => { + return Err(format!( + "route '{}': qos must be 0, 1 or 2, got '{other}'", + route.default_topic + )) + } + }; + let retain = match option("retain") { + None | Some("false") => false, + Some("true") => true, + Some(other) => { + return Err(format!( + "route '{}': retain must be true or false, got '{other}'", + route.default_topic + )) + } + }; + Ok(Self { qos, retain }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use aimdb_core::transport::ConnectorConfig; + use alloc::string::ToString; + use alloc::sync::Arc; + use alloc::vec::Vec; + + fn route(query: &[(&str, &str)]) -> RouteInfo { + let query: Vec<(String, String)> = query + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect(); + RouteInfo { + id: 0, + default_topic: Arc::from("sensors/t"), + config: ConnectorConfig::from_query(&query), + topic_capacity: 0, + payload_capacity: 0, + } + } + + #[test] + fn defaults_are_qos_1_without_retain() { + assert_eq!( + PublishOpts::parse(&route(&[])), + Ok(PublishOpts { + qos: 1, + retain: false + }) + ); + } + + #[test] + fn every_valid_value_parses() { + for (qos, expected) in [("0", 0), ("1", 1), ("2", 2)] { + assert_eq!( + PublishOpts::parse(&route(&[("qos", qos)])).unwrap().qos, + expected + ); + } + assert!( + PublishOpts::parse(&route(&[("retain", "true")])) + .unwrap() + .retain + ); + assert!( + !PublishOpts::parse(&route(&[("retain", "false")])) + .unwrap() + .retain + ); + } + + #[test] + fn a_value_that_does_not_parse_names_the_route() { + for query in [[("qos", "3")], [("qos", "abc")], [("retain", "yes")]] { + let err = PublishOpts::parse(&route(&query)).unwrap_err(); + assert!(err.contains("route 'sensors/t'"), "{err}"); + assert!(err.contains(query[0].1), "{err}"); + } + } +} diff --git a/aimdb-mqtt-connector/tests/session_loop.rs b/aimdb-mqtt-connector/tests/session_loop.rs index e0d256e5..82c778fa 100644 --- a/aimdb-mqtt-connector/tests/session_loop.rs +++ b/aimdb-mqtt-connector/tests/session_loop.rs @@ -403,11 +403,10 @@ async fn every_qos1_push_is_acknowledged() { // The build-time QoS warning can actually see what it warns about. // --------------------------------------------------------------------------- -/// `warn_unsupported_qos` scans `collect_outbound_routes("mqtt")` for a `qos` -/// entry in each route's query config. That scan is the part that can silently -/// find nothing — a scheme filter that does not match, or a config key that -/// never lands — leaving a warning that compiles and never fires. This asserts -/// the shape it depends on, mirroring the private function exactly. +/// `build()` warns once per route whose `RouteInfo` carries `qos=2`. That +/// lookup is the part that can silently find nothing — a scheme that does not +/// match, or a config key that never lands — leaving a warning that compiles +/// and never fires. This asserts the shape it depends on. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn a_qos2_route_is_visible_to_the_build_time_scan() { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); @@ -416,20 +415,17 @@ async fn a_qos2_route_is_visible_to_the_build_time_scan() { let dialer = CountingDialer::new(); let (db, _runner) = build_db(port, dialer, Some((Duration::from_secs(60), 2))).await; - let routes = db.collect_outbound_routes("mqtt"); - assert!( - !routes.is_empty(), - "the mqtt scheme must match, or the scan sees no routes at all" - ); - + let routes = aimdb_core::OutboundRoutes::new(&db, "mqtt").expect("routes"); let flagged: Vec<(&str, &str)> = routes + .routes() .iter() .filter_map(|route| { route .config + .protocol_options .iter() .find(|(k, _)| k == "qos") - .map(|(_, v)| (route.topic.as_str(), v.as_str())) + .map(|(_, v)| (&*route.default_topic, v.as_str())) }) .collect(); @@ -439,3 +435,80 @@ async fn a_qos2_route_is_visible_to_the_build_time_scan() { "the scan must see the route's topic and its qos option; got {flagged:?}" ); } + +// --------------------------------------------------------------------------- +// Outbound values wait in their record buffers, not in a connector queue. +// --------------------------------------------------------------------------- + +/// While the broker holds a PUBACK, the session cannot publish, and values +/// produced meanwhile stay in their record buffer. A single-latest record +/// keeps only the newest, so the publish after the stall carries `9`, not the +/// nine values a connector queue would have held. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn after_a_stall_a_single_latest_record_sends_only_its_newest_value() { + use aimdb_core::buffer::BufferCfg; + use aimdb_core::AimDbBuilder; + use aimdb_mqtt_connector::{MqttConnector, MqttLinkExt}; + use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; + + const ACK_DELAY: Duration = Duration::from_millis(1500); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let log = Arc::new(Mutex::new(Log::default())); + + let connector = MqttConnector::new(format!("mqtt://127.0.0.1:{port}")) + .transport(CountingDialer::new()) + .with_client_id("outage") + .with_keep_alive(TEST_KEEP_ALIVE); + let mut builder = AimDbBuilder::new() + .runtime(Arc::new(TokioAdapter)) + .with_connector(connector); + builder.configure::("uptime", |reg| { + reg.buffer(BufferCfg::SingleLatest) + .link_to("mqtt://sensors/uptime") + .with_qos(1) + .with_serializer(|_ctx, value: &u64| Ok(value.to_string().into_bytes())) + .finish(); + }); + let (db, runner) = builder.build().await.expect("build db"); + let producer = db.producer::("uptime").expect("producer"); + + let published = |log: &Mutex| -> Vec { + log.lock() + .unwrap() + .publishes + .iter() + .map(|(_, p)| String::from_utf8_lossy(p).into_owned()) + .collect() + }; + let drive = async { + producer.produce(0); + while published(&log).is_empty() { + tokio::time::sleep(Duration::from_millis(5)).await; + } + // The PUBACK for `0` is held: produce the rest during the stall, + // spaced so that anything reading the buffer meanwhile sees each one. + for n in 1..=9 { + producer.produce(n); + tokio::time::sleep(Duration::from_millis(20)).await; + } + while published(&log).len() < 2 { + tokio::time::sleep(Duration::from_millis(5)).await; + } + // Nothing else may follow. + tokio::time::sleep(ACK_DELAY + Duration::from_millis(500)).await; + }; + + tokio::select! { + _ = runner.run() => panic!("the session loop returned"), + _ = serve_one(listener, log.clone(), Script::SlowPuback { delay: ACK_DELAY }) => { + panic!("the broker returned") + } + _ = drive => {} + _ = tokio::time::sleep(Duration::from_secs(30)) => { + panic!("watchdog: published {:?}", published(&log)) + } + } + + assert_eq!(published(&log), ["0", "9"]); +} diff --git a/aimdb-mqtt-connector/tests/tokio_broker.rs b/aimdb-mqtt-connector/tests/tokio_broker.rs index 32e7c303..1295ad27 100644 --- a/aimdb-mqtt-connector/tests/tokio_broker.rs +++ b/aimdb-mqtt-connector/tests/tokio_broker.rs @@ -377,3 +377,134 @@ async fn the_embedded_backend_dispatches_inbound_on_its_session_task() { let futures = connector().build(&db).await.expect("build connector"); assert_eq!(futures.len(), 1, "the session task, and no inbound pump"); } + +// --------------------------------------------------------------------------- +// Build-time checks: per-route options and what the write ring can carry. +// --------------------------------------------------------------------------- + +/// An embedded-backend connector that never dials. +macro_rules! embedded { + () => { + aimdb_mqtt_connector::MqttConnector::new("mqtt://127.0.0.1:1") + .transport(aimdb_tokio_adapter::net::TokioNet::tcp()) + }; +} + +/// Builds a database with `connector` and the records `configure` adds. +async fn build( + connector: impl aimdb_core::connector::ConnectorBuilder + 'static, + configure: impl FnOnce(&mut aimdb_core::AimDbBuilder), +) -> Result<(), String> { + let mut builder = aimdb_core::AimDbBuilder::new() + .runtime(Arc::new(aimdb_tokio_adapter::TokioAdapter)) + .with_connector(connector); + configure(&mut builder); + builder.build().await.map(|_| ()).map_err(|e| e.to_string()) +} + +/// One outbound record on `sensors/out` with `config` on its link and, given +/// a capacity, a scratch serializer. +fn outbound( + config: &'static [(&'static str, &'static str)], + scratch: Option, +) -> impl FnOnce(&mut aimdb_core::AimDbBuilder) { + move |builder| { + use aimdb_tokio_adapter::TokioRecordRegistrarExt; + builder.configure::("out", move |reg| { + let mut link = reg + .buffer(aimdb_core::buffer::BufferCfg::SingleLatest) + .link_to("mqtt://sensors/out") + .with_serializer(|_ctx, v: &u64| Ok(v.to_le_bytes().to_vec())); + for (k, v) in config { + link = link.with_config(k, v); + } + if let Some(capacity) = scratch { + link = link.with_serializer_into(capacity, |_ctx, v: &u64, out| { + out[..8].copy_from_slice(&v.to_le_bytes()); + Ok(8) + }); + } + link.finish(); + }); + } +} + +#[tokio::test] +async fn an_invalid_qos_or_retain_fails_the_build() { + for (key, value, expected) in [ + ("qos", "3", "qos must be 0, 1 or 2, got '3'"), + ("qos", "abc", "qos must be 0, 1 or 2, got 'abc'"), + ("retain", "yes", "retain must be true or false, got 'yes'"), + ] { + let config: &'static [(&str, &str)] = Box::leak(Box::new([(key, value)])); + let err = build(embedded!(), outbound(config, None)) + .await + .unwrap_err(); + assert!(err.contains("route 'sensors/out'"), "{err}"); + assert!(err.contains(expected), "{err}"); + } +} + +#[tokio::test] +async fn a_route_too_large_for_the_write_ring_fails_the_build() { + let err = build(embedded!(), outbound(&[], Some(3000))) + .await + .unwrap_err(); + assert!( + err.contains("route 'sensors/out': its largest PUBLISH is"), + "{err}" + ); + assert!(err.contains("it is 4096"), "{err}"); + + build( + embedded!().with_write_buffer(8192), + outbound(&[], Some(3000)), + ) + .await + .expect("a larger ring carries it"); +} + +#[tokio::test] +async fn a_connect_too_large_for_the_write_ring_fails_the_build() { + let password = "p".repeat(2100); + let err = build(embedded!().with_credentials("user", &password), |_| {}) + .await + .unwrap_err(); + assert!( + err.contains("the CONNECT (client id and credentials) is"), + "{err}" + ); + + build( + embedded!() + .with_credentials("user", &password) + .with_write_buffer(8192), + |_| {}, + ) + .await + .expect("a larger ring carries it"); +} + +#[tokio::test] +async fn a_subscribe_too_large_for_the_write_ring_fails_the_build() { + let topic = format!("sensors/{}", "t".repeat(2100)); + let inbound = |topic: String| { + move |builder: &mut aimdb_core::AimDbBuilder| { + use aimdb_tokio_adapter::TokioRecordRegistrarExt; + builder.configure::("in", move |reg| { + reg.buffer(aimdb_core::buffer::BufferCfg::SingleLatest) + .link_from(&format!("mqtt://{topic}")) + .with_deserializer(|_ctx, data: &[u8]| Ok::(data.len() as u64)) + .finish(); + }); + } + }; + let err = build(embedded!(), inbound(topic.clone())) + .await + .unwrap_err(); + assert!(err.contains("the SUBSCRIBE to 'sensors/"), "{err}"); + + build(embedded!().with_write_buffer(8192), inbound(topic)) + .await + .expect("a larger ring carries it"); +} diff --git a/aimdb-mqtt-connector/tests/write_ring_proofs.rs b/aimdb-mqtt-connector/tests/write_ring_proofs.rs index a2574298..ea9ced11 100644 --- a/aimdb-mqtt-connector/tests/write_ring_proofs.rs +++ b/aimdb-mqtt-connector/tests/write_ring_proofs.rs @@ -55,9 +55,11 @@ 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. +/// and the small publishes around it keep flowing. The skip is counted as +/// rejected in the route's `RouteStats` (asserted in the session loop's unit +/// tests, which hold the `OutboundRoutes`). #[tokio::test(flavor = "multi_thread", worker_threads = 4)] -async fn proof_an_oversize_publish_is_dropped_silently() { +async fn an_oversize_publish_is_skipped_and_the_session_stays_up() { 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())); diff --git a/aimdb-tokio-adapter/tests/outbound_routes.rs b/aimdb-tokio-adapter/tests/outbound_routes.rs index 471de39a..a713225d 100644 --- a/aimdb-tokio-adapter/tests/outbound_routes.rs +++ b/aimdb-tokio-adapter/tests/outbound_routes.rs @@ -496,3 +496,14 @@ async fn every_route_closes_when_the_database_is_dropped() { .expect("closes"); assert!(end.is_none()); } + +#[tokio::test] +async fn a_rejected_message_is_counted_beside_sent() { + let db = db(vec![spmc(0, 16)]).await; + let mut o = OutboundRoutes::new(&db, "test").unwrap(); + produce(&db, 0, 1); + let id = pull(&mut o).await.unwrap().0; + o.reject(id); + let stats = o.stats(0).unwrap(); + assert_eq!((stats.sent, stats.rejected), (1, 1)); +}