feat(mqtt): pull embedded outbound publishes from OutboundRoutes (design 054 §4.7)

[lxsaah]

Stage 10 of the 054 implementation plan. The embedded MQTT session now pulls outbound messages from OutboundRoutes when it can send. The action channel, MqttSink and the per-route pump_sink tasks are gone, so the embedded connector contributes exactly one future: its session task. This stage also lands four follow-ups from the #283 review: counting skipped publishes (task 1), with_write_buffer (task 2), size checks at build (task 6), and the broker's CONNACK limit (task 7).

Change

Core (outbound/routes.rs):

• RouteStats::rejected and OutboundRoutes::reject(id): a connector reports a message it took but could not send.
• sent now reads "staged and handed to the connector, including any it then rejected".

New publish_opts.rs:

• PublishOpts::parse(&RouteInfo) reads qos (0/1/2, default 1) and retain (true/false, default false).
• Behaviour change: a value that does not parse is now a build error naming the route. It used to fall back silently to the default.
• Gated on embedded for now. Stage 11 shares it with the native backend.

connector.rs: with_write_buffer(bytes) on MqttConnector<Embedded<_>> and MqttConnector<EmbeddedTls<_>>, default 4,096. The size travels through build_*, setup_* and run_sessions/run_tls to the ring.

embedded/mod.rs — new prepare_outbound, run in build():

• Builds OutboundRoutes::new(db, "mqtt") and parses each route's options. Routes asking for qos=2 get a warning once each (log_warn! and defmt); this client sends them at QoS 1, as before.
• Fails the build, with every problem listed and both sizes named, when one of these doesn't fit the ring (frame plus reserve ≤ capacity / 2):
  • a route's largest PUBLISH (overhead + max(default topic, topic capacity) + payload capacity)
  • the CONNECT (client id and credentials)
  • any SUBSCRIBE
• Removed: AimdbMqttAction (public, so breaking), ActionChannel, CHANNEL_SIZE, MqttSink, collect_pumps, the pump_sink call, map_qos, opt_u8/opt_bool, and warn_unsupported_qos (with its use of collect_outbound_routes).
• The manager's ActionChannel alias is removed too.

session_loop.rs:

• Publish arm (design §4.7): gated on connected, nothing in flight, all subscriptions placed, and room for the largest PUBLISH plus its reserve (re-checked on every poll through the room waker). It then calls poll_stage. Ready(None) is latched in outbound_done, and a losing arm takes nothing.
• publish_staged: after the select, it takes the staged message, builds the PUBLISH with the route's qos/retain, then writes it into the ring and updates the state.
  • It skips the message, logs it and calls outbound.reject(id) when the frame doesn't fit the ring, or exceeds the broker's Maximum Packet Size from its CONNACK. The CONNACK property is read in drain_packets and kept per session.
  • Sending a larger packet would be a protocol error (DISCONNECT 0x95).
• connect_packet is shared by the session and the build check. subscribe_len and publish_frame_len size packets at build. write_ring::fits_ring is the same rule without a ring.
• perform is gone. At most once still holds: a message is taken from its record buffer before it is written.

Diff: +844 / −378 overall, including core, the new parser and tests. In aimdb-mqtt-connector/src excluding the parser it's +470 / −363.

Tests

New:

• after_a_stall_a_single_latest_record_sends_only_its_newest_value (session_loop): during a held PUBACK, values 1–9 are produced 20 ms apart, and the broker then receives ["0", "9"].
  • Run against the previous code (src/ stashed), the same test sees ["0", "1", "2", …].
  • An earlier version of the test produced the values in a tight loop. It passed on both versions, because single-latest collapses them before the old pump ran, so I spaced them out.
• an_invalid_qos_or_retain_fails_the_build: qos=3, qos=abc and retain=yes each fail the build, naming the route.
• a_route_too_large_for_the_write_ring_fails_the_build, a_connect_too_large_… and a_subscribe_too_large_…: each fails at the default ring and builds with with_write_buffer(8192).
• Publish-cap proofs moved to publish_staged on a real staged message:
  • 1,984 bytes goes out.
  • 1,985 bytes is skipped and counted (sent 1, rejected 1). This proof flipped.
  • A message over the broker's limit is skipped and counted; at the limit it goes out.
• publish_frame_len_matches_what_the_client_state_encodes.
• PublishOpts unit tests.
• Core: a_rejected_message_is_counted_beside_sent (Tokio adapter) and reject with an unknown id.

Updated:

• a_qos2_route_is_visible_to_the_build_time_scan reads OutboundRoutes::routes().
• The loopback proof_an_oversize_publish_is_dropped_silently is renamed an_oversize_publish_is_skipped_and_the_session_stays_up.
• an_idle_session_wakes_at_the_ping_cadence (inbound-only connector) still passes, so Ready(None) doesn't spin.

Suite	Passed
embedded-tls lib	47
session_loop	7
tokio_broker	9
embassy_broker	1
tls_session	3
tls_broker	3
backend_parity	8
write_ring_proofs	1
--features std	40
aimdb-core	261
aimdb-tokio-adapter	87

Not done here

• Outage semantics on Embassy buffers (design §8). The plan wanted the outage case in embassy_broker too. That test runs its own executor and fake broker over embassy-net, so a held PUBACK would need a second scripted broker. The question concerns the buffers, not MQTT, so I propose a follow-up test of OutboundRoutes directly over the Embassy adapter's buffers, like outage_semantics_per_buffer_type for Tokio.
• Reading RouteStats from outside the connector. OutboundRoutes lives inside the session task, so only the connector can call stats(id). The new count is asserted at unit level. Users need a read path, for example AimDb::outbound_stats(scheme) backed by shared counters. That's a core API decision, so I've left it open.

Verification

• All of the above.
• make clippy (whole workspace, including embedded targets), every core and MQTT doc leg (-D warnings) and cargo fmt --all --check: clean.
• embassy-mqtt-connector-demo and weather-station-gamma build for thumbv8m.main-none-eabihf.
• CI does not run on PRs into feat/054-connector-boundary.

🤖 Generated with Claude Code