feat(052): runtime-neutral MQTT connector

[lxsaah]

Wave C of design 052, plus the event-driven session rewrite that came out of it.

The split is now std vs no_std, not Tokio vs Embassy: the embedded backend runs on any target whose adapter supplies a StreamDialer, so a FreeRTOS port costs one adapter crate and no change here.

Why MQTT keeps two protocol clients

Unifying on mountain-mqtt was considered and rejected: it does not support QoS 2 (Qos2NotSupported; the codec has pubrec/pubrel/pubcomp, the state machine does not use them). rumqttc cannot ride the neutral layer either — its Transport is a closed enum, so no stream can be injected.

So one MqttConnector<B> over two backends, which keeps unification a later, small change: if QoS 2 lands in the fork, deleting Native is a deletion, not a rewrite.

Backend	Client	QoS	TLS
Native (no transport supplied)	rumqttc (std)	0–2	rustls
Embedded<D> (.transport(..))	mountain-mqtt (no_std)	0–1	embedded-tls

The session is event-driven — nothing polls

The old loop woke every 10 ms to ask three sources whether they had work, which on a battery node is the only state that normally runs. Worse, the "non-blocking peek" it polled with could block indefinitely, parking the loop and with it the pings, the liveness check and every queued publish.

The stream is now split into halves driven by three futures in one select: a reader lifting bytes off the socket, a writer draining encoded packets, and the session itself on two channels and one timer. tests/session_loop.rs asserts an idle session wakes fewer than 30 times in 3 s where the poll cost ~300, and that a QoS 1 publish no longer spins waiting inline for its PUBACK.

TLS runs that same loop. mqtts:// used to need a loop of its own because the client wanted a readiness peek a TLS session cannot answer honestly — bytes on the wire may decrypt to no application data at all. Nothing peeks any more, so the bespoke Connection, the readiness probe onto the raw socket beneath the TLS session, and the RefCell wrapping the whole socket so both could reach it are gone. What replaces them is one lock per direction behind a cloneable handle.

mountain-mqtt is now the codec alone — default-features = false, no features, on aimdb-mountain-mqtt 0.5.1 (upstream main, zero-line source delta). The driver half lives here. A Makefile cargo tree guard asserts the subtree stays codec-only and that no executor, network stack, adapter or logger reaches the embedded graph.

Breaking

Features rename. std carries rumqttc (tokio-runtime is a deprecated alias, still works); embedded carries mountain-mqtt with alloc only; embedded-tls is runtime-neutral TLS; embassy-runtime and embassy-tls are convenience bundles over those. critical-section-std-impl links a critical-section impl for std binaries.

Modules rename with no compatibility re-export: tokio_client → native, embassy_client → embedded. A shim would have been theatre — the builders those modules held are gone too, so the old import fails either way.

The Tokio path is otherwise unchanged — MqttConnector::new(&url) still. Embedded callers supply the transport:

// mqtt:// — the adapter owns the socket
MqttConnector::new(&url).transport(EmbassyNet::tcp(*stack, MQTT_RX.init(..), MQTT_TX.init(..)))

// mqtts:// — same transport; the dialer resolves the host, so TLS needs no stack
MqttConnector::new(&url).tls(EmbassyNet::tcp(..), TlsOptions::new(..))

.tls(dialer, options) replaces .tls(stack, options). The certificate-validity clock comes from RuntimeOps::unix_time(); SNTP is opt-in via TlsOptions::with_sntp for a runtime with no wall clock of its own. MqttConnectorBuilder and the Tokio*/Embassy* aliases are gone; with_client_id/with_credentials now work on either backend.

The largest MQTT packet the session can receive is 3328 bytes (was 4096). Reassembly, read scratch and the inbound slot are carved out of one BUFFER_SIZE rather than added to it, with one read chunk held in reserve so a chunk completing one packet can still carry the head of the next. Outbound packets are encoded to exactly their own size on the heap, so they gain no fixed cap.

Core and adapters

ByteStream::split, with ByteRead/ByteWrite (aimdb-core) — borrows a stream into independently pollable halves, which a single &mut stream cannot express. Both adapters implement it. The cancellation contract is written down with it: read is cancel-safe on both adapters AimDB ships, write_all is cancel-safe nowhere and must never sit in a select arm.

⚠️ Action required — EmbassyTcpDialer resolves hostnames, so the adapter's net feature now enables embassy-net/dns. embassy_net::new adds the resolver socket itself, so every application using net must grow its StackResources<N> by one; too small an N panics at stack construction. Previously a connector handing through a hostname dialed fine on Tokio and failed forever on Embassy.

aimdb-tokio-adapter gains an embedded-io feature — the embedded-io-async trio on the net streams, including a non-destructive ReadReady via poll_peek. That is what lets the embedded backend and embedded-tls run on a host, so both are covered by real tests rather than a cross-compile alone.

Both dialers also implement Delay, so a connector generic over one needs no separate clock handle.

Fixed during review

• PUBACKs were dropped under an inbound burst. The 4-slot write queue was try_send-and-forget, so a coalesced burst lost every response past the fourth — measured at 19 of 40 acknowledged. Unrecoverable: the client state had already retired the message, and pings kept the liveness watchdog satisfied, so the node went deaf on inbound QoS 1 with no error and no reconnect. Protocol obligations now wait for a slot; only pings are still dropped, since a ping arms no response deadline. every_qos1_push_is_acknowledged is the regression test.
• qos=2 diverged silently between backends — Native honours it, Embedded downgrades to QoS 1. The build now names each such route in a warning.
• Two test harnesses asserted nothing: they set QoS via a ?qos= query on the link URL, which LinkAddress::parse strips, so every test ran at the default regardless. Both now use .with_qos(..).
• The documented packet-size ceiling was a chunk too high; embedded-hal-async was a dead manifest entry; the TLS locking comment pointed at a test file that did not exist.

Not addressed

• A malformed route option (with_config("qos", "two")) falls to the default on both backends identically. Consistent, so not a divergence — worth its own pass if route-URL typos bite.
• An over-limit retained message reconnect-loops the connector, since retain replays on every resubscribe and clean-start does not clear it. Documented in the CHANGELOG; the fix needs a "discard N more bytes" state in the reader.

Verification

Locally: all connector tests across six binaries, the aimdb-core session suite, clippy -D warnings on the std, embedded, embedded,tracing and embassy-runtime,embassy-tls,defmt legs (thumbv7em where applicable), rustdoc -D warnings per feature leg, and cargo fmt --check. The full matrix is CI's.

🤖 Generated with Claude Code