feat(mqtt): encode embedded packets into one bbqueue write ring (design 054 §4.7) - #283
Merged
Merged
Conversation
…gn 054 §4.7) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…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 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Stage 8 of the 054 implementation plan, the first connector stage. The embedded MQTT session no longer allocates a
Vecper outbound packet: every packet is encoded straight into onebbqueuering, allocated once per connector. The action and event channels stay for now (stages 9 and 10).Change
aimdb-mqtt-connector/src/embedded/write_ring.rs(new):WriteRingBBQueue<BoxedSlice, AtomicCoord, Polling>, plus two embassy-syncAtomicWakers:data: a commit wakeswrite_out.room: a release wakes whatever waits for space.bbqueue's polling notifier wakes nothing, so these carry the signals.
CONTROL_RESERVE64, which givesmax_publish()1,984 bytes. That is half the ring minus the reserve: the most a bipbuffer can always grant contiguously, whatever its offset.put(packet, reserve): sizes the packet withMqttLenWriter, waits forgrant_exact(len + reserve), encodes withMqttBufWriterover the grant, and commits onlylen.try_put: the lossy variant, for pings.has_room(n)is a probe grant dropped uncommitted.poll_roomandwait_roomare the waker-registering forms.drain()discards leftover bytes.write_out(tx)writes read grants, flushes, releases them and wakesroom.session_loop.rsChannel<Vec<u8>, 4>,encode,queue,queue_lossyand the oldwrite_outare gone. CONNECT and SUBSCRIBE useput(…, 0), PUBLISH usesput(…, CONTROL_RESERVE), and PINGREQ usestry_put(still lossy).run_sessiontakes&WriteRingand drains it first, so no bytes from an old session reach a new socket ahead of its CONNECT.max_publish() + CONTROL_RESERVE, checked on every poll through theroomwaker instead of!outbound.is_full()computed once.max_publish()is skipped with adefmtwarning before the client state commits to it.drain_packetswaits forPUBACK_ROOM(16 bytes) before parsing each packet, then encodes the PUBACK withtry_putbeforestate.receivetakes&mut.putinside the parse scope. That version kept the parsedPacketGeneric(32 property slots) inside the session future and grew it to 10,488 bytes, which failed the existing ≤ 8,192 size test.PUBACK_ROOM.session.rs,tls.rs: each creates the ring once, before its reconnect loop, and passes it to every session.Cargo.toml:bbqueue = { version = "0.7", default-features = false, features = ["alloc"] }on theembeddedfeature (MIT OR Apache-2.0;cargo deny check licensesok).Tests
New unit tests in
write_ring.rs:max_publishplus the reserve is exactly half the default ring.draindiscards an old session's bytes.write_out's release, and goes out in order.try_putdrops instead of waiting.Existing suites, unchanged and passing:
session_loop(idle ping cadence, outbound under inbound flood)tokio_broker(reconnect and resubscribe)embassy_brokertls_sessiontls_brokerbackend_parity--features stdembedded-tlslib testsReview follow-up (
dc9a3a4)putnow fails withOverflowwhen a frame plus its reserve exceeds half the ring. Before, a CONNECT or SUBSCRIBE that large (for example a JWT used as the password) parked forever outside the session'sselect, with no deadline and no error. Such a packet now ends the session. Refusing it at build is planned for stage 10.fits(len, reserve)andput_sizedletperformreuse its encoded length instead of computing it twice.mainthat advertises Maximum Packet Size)tests/write_ring_proofs.rs(a fake broker) runs inmake testandmake clippy.putbeyond half their size; they now fill withtry_put.a_connect_over_half_the_ring_fails_at_any_offset.Verification
cargo clippy -p aimdb-mqtt-connector --target thumbv7em-none-eabihf --no-default-features --features "embassy-runtime,defmt" -- -D warnings: clean.make clippy(whole workspace) andcargo fmt --all --check: clean.feat/054-connector-boundary.🤖 Generated with Claude Code