feat: MQTT binary publish and topic-scoped binary receive

[genmon]

Prompted by a request from the hawthorn MQTT broker hardening work, which wants voice audio to ride the MQTT connection the device already holds so firmware can drop its second WebSocket (and with it a permanent TLS session).

The bug this starts from

esp_mqtt_client_publish does if (len <= 0 && data != NULL) len = strlen(data), and every MqttTransport::publish overload passed 0. So:

• PCM16 is full of NUL bytes — silence is entirely zeros — and audio frames were truncated at the first zero sample.
• Worse, a buffer with no terminator at all is read past the end of its allocation until a zero byte turns up, and whatever it found gets published onto a topic.

publishBinary(topic, data, len, qos, retain) passes a real length and rejects nullptr, len == 0 and len > INT_MAX rather than arranging for IDF's happy path to be taken. A binary caller passing zero length is a bug, and silently emitting an empty publish would hide it at the one place best positioned to catch it. An intentionally empty payload is a text publish: publish(topic, "", qos, retain).

Binary receive, now rather than later

MQTT 3.1.1 has no content-type on the wire, so the subscriber declares the lane:

mqtt.subscribeBinary("devices/+/voice/audio");
mqtt.onBinary([](const char* topic, const uint8_t* data, size_t len) { ... });

Payloads on a binary-declared filter reach onBinary only — never onMessage, never Client's JSON lane, so nothing deserializeJsons them once MQTT is the default transport. Config::binaryTopics declares the same at construction.

The justification for building this before there is a caller is narrow: retrofitting a receive hook onto a released transport is a breaking change, whereas adding it before the release is free. That argument covers the seam, not the capacity — see Known limits.

Two things this needed that aren't obvious:

• Wildcard filters. subscribeBinary registers a filter but the broker delivers a concrete topic, so an exact-string lookup never matches and the frame falls back into the JSON lane — precisely the failure the opt-in exists to prevent. Hence MqttTransport::topicMatches, public because callers dispatching inside onMessage need the same rule.
• Where classification happens. At loop() drain time, on the task that owns the subscription list — not on the ESP-IDF event task, which would have meant a cross-task read of a std::vector per inbound message.

Publishing from a second task

esp-mqtt is thread-safe for concurrent API calls: publish, subscribe_multiple, unsubscribe, start and stop each take an internal MQTT_API_LOCK. What it does not protect is the handle's lifetime — esp_mqtt_client_destroy takes no lock and frees the transport list, the outbox, the buffers, the API lock itself, and then the client. It structurally can't be lock-protected.

That is courier's actual gap: publish() reads _client and enters IDF while a reconnect escalation on the app task frees it. Lock.h adds two mutexes for that, with the deadlock rule against IDF's own API lock written down in the header. _clientLock is taken with a 250 ms bound by the publish path (returns false rather than waiting out a teardown, which stop() can stretch across a TLS connect) and unbounded by begin/disconnect/suspend/resume. _topicsLock covers _topics only and is read by the IDF task once per connect.

It is not a congestion signal. A lone publisher entering a busy client still waits on IDF's unbounded lock; no Courier-side lock can change that. Config::network_timeout_ms is the knob for that worst case.

Also

• Subscriptions are re-subscribed at their registered QoS after a reconnect. subscribe(topic, qos) applied the QoS on the initial call, but subscribeAll() re-subscribed everything at QoS 0.
• Config::out_buffer_size and Config::network_timeout_ms, both defaulting to 0 = leave the IDF value. A 1920-byte frame against the default 1024 buffer is two writes and two TLS records; sizing past the frame makes it one.
• Recent changelog entries rewritten to one line per change.

Known limits

The MQTT inbound path is a depth-8 SPSC queue with a heap-allocated topic per message, drained on the app task at loop() cadence. onBinary is the right shape for control-rate blobs. It is not sized for a sustained stream — a 16.7 fps audio downlink overflows it within half a second of a slow loop(). Documented in docs/api.md as a known boundary; a TTS-to-device downlink would be separate work on the inbound path, probably dispatching on the MQTT task rather than through the app-task queue.

Testing

• ./tools/run-tests.py unit — 181 pass. 18 new, including: embedded-NUL round trip asserted on .size() and never as a C string; the three rejection cases; binary-lane routing; wildcard routing; client-hook bypass; multi-chunk reassembly; survival across reconnect; and three threaded tests that park a second task inside a mocked publish to check the bounded lock refuses rather than blocks.
• ./tools/run-tests.py static-analysis — clean.
• ./tools/run-tests.py build — the two Arduino-framework examples compile MqttTransport.cpp and pass. The three arduino, espidf examples could not run locally: PlatformIO's bundled tool-ninja is an x86_64 binary and the dev machine is arm64 (bad CPU type in executable). Pre-existing and unrelated to this change, but it means the ESP-IDF-component builds are unverified outside CI.

The device-side locking is argued from esp-mqtt's source and exercised by host tests; it has not run on hardware with a real send task racing a reconnect. Worth watching in first bring-up.

🤖 Generated with Claude Code