Internals: MQTT

How @opentf/esrun-mqtt is put together, and why each rule it enforces is there. The decision and what was rejected are in docs/DECISIONS.md (D172).

A package, not a runtime module

Everything the client needs is already in the runtime: runtime:net opens a TCP or TLS socket and hands back web streams. So the client is JavaScript over that API, as the database drivers are — no new code in esrun, and no new capability: reaching the broker is --allow-net, scoped to its host like any other socket.

That also decides what it cannot do. It reaches only hosts the run was granted, and TLS is verified by the runtime's own verifier, with no option here to switch it off. It speaks MQTT 5.0, and falls back to 3.1.1 for a broker that does not — MQTT over WebSockets, and the 3.1 wire format below 3.1.1, are different framing and are out of scope.

The layers

Each layer is usable alone; the ones below it decide what the ones above can assume. The client never touches a socket — it is handed a Transport — which is why the handshake and the packet routing can be tested against a scripted broker in memory, with no port and no capability granted.

The packet boundary

MQTT is a stream: a fixed header (a type byte and its flags), a Remaining Length in one to four octets, and a variable header and payload of exactly that many bytes. A socket delivers it in whatever chunks it likes — a fixed header that arrives alone, a length split across two reads, several packets in one chunk are all possible.

So the connection accumulates bytes and is what decides a packet is complete. A buffer too short to say is not an error, just a packet not yet arrived; only a length that is malformed — a fifth octet, which no amount of waiting fixes — is raised. This is solved once, in the connection, rather than by every packet. The accumulation buffer grows by doubling and is compacted in place: copying everything buffered on every chunk made a large message cost the square of its size.

A CONNACK is judged by its own level's rule. A 5.0 reason code refuses from 0x80 up; a 3.1.1 return code refuses with anything but 0, and its refusals are 1–5. Judged by the 5.0 rule, a 3.1.1 broker refusing the password looked like one accepting the connection, and the client found out only when the socket closed.

Why a lost socket is latched

If the socket is lost, or the bytes are not MQTT, the stream is at an unknown boundary and every later packet would be misread. So such a failure is latched once, and every caller — now and later — gets that one cause rather than a new symptom of the same dead connection. There is no resynchronisation: a stream that lost a byte cannot be repaired, only replaced.

The connection does not repair itself; the client replaces it (below).

Version negotiation

A broker that will not speak 5.0 says so one of two ways: a CONNACK reason code of 0x84 (Unsupported Protocol Version), or an older CONNACK whose return code has no 5.0 meaning. Both mean the same thing — dial again as 3.1.1 — so the negotiation is a loop in one place, not a flag threaded through the codec. Any other refusal (bad credentials, a banned client) is reported, not retried, because a second protocol level would not change it.

The two levels share a codec but not every rule: 5.0 properties exist only on a 5.0 connection, and asking for one on 3.1.1 is refused rather than emitted as a packet the broker would reject. The same is true of request/reply, which is the 5.0 response-topic convention and nothing else.

Acknowledging on ownership

An inbound QoS 1 message is acknowledged when the application takes it, not when it arrives. The specification says the receiver sends PUBACK "having accepted ownership of the Application Message" (4.3.2), and the broker's send quota is replenished only by PUBACK and PUBCOMP (4.9) — so acknowledging on arrival, as mqtt.js does, claims ownership the application does not have yet and leaves memory as the only bound on a fast publisher. With the Receive Maximum the client advertises (250 by default), acknowledging on ownership is what holds the broker to the application's pace; on 3.1.1, the broker's own in-flight window does the same.

QoS 2 is delivered on its PUBLISH and de-duplicated by packet identifier: a DUP retransmission before the application has taken it is the same message, and one after (PUBREC sent, PUBREL not yet seen) only draws the PUBREC again. The identifiers awaiting PUBREL are session state, kept across reconnects, because a resumed session's broker resends PUBREL for exactly those. The rest of the exchange — PUBREL for our own QoS 2 publishes, PUBCOMP for theirs — needs no decision, and the connection sends it.

QoS 0 has no acknowledgement and cannot be flow-controlled, so each subscription holds at most maxBuffered of them and drops the oldest past that.

Reconnecting, and what a session allows

mqtt.js reconnects by default, and so does this client — but where it would guess, the specification decides (DECISIONS D178). On a reconnect with Clean Start 0 and a session present, both sides "MUST resend any unacknowledged PUBLISH packets (where QoS > 0) and PUBREL packets using their original Packet Identifiers… [and] MUST NOT resend messages at any other time" (4.4). So a resumed session resends what was in flight, with DUP set; a fresh one rejects it, since republishing it as a new message is exactly what 4.4 forbids and the receiver could not tell the copy from the original. A fresh session has every subscription sent again; a resumed one keeps the broker's.

Retries back off exponentially with jitter, so a restarted broker's clients do not return in lockstep. Session taken over normally ends the client — another client now holds the id, and reconnecting would make the two evict each other forever — with one exception found in testing: behind Docker's port proxy, an attempt made while the broker restarted saw its handshake fail and was abandoned, but the proxy delivered its CONNECT once the broker was back, after a later attempt had connected, and the broker handed the session to that ghost. Attempts whose socket opened and whose handshake failed are counted, and a takeover while one is outstanding reconnects instead of ending the client.

Keep-alive

The connection sends a PINGREQ every half the keep-alive interval, and treats silence for twice the interval as a lost connection. The interval is what was negotiated: a broker's serverKeepAlive (5.0) overrides what was asked for.

The keep-alive timer is unreferenced: it does not keep a script alive that has otherwise finished, so a program exits when its work is done rather than waiting out the interval.

A PINGRESP carries no identifier, so two PINGREQs in flight cannot be told apart: either reply answers either, and a reply to a ping already given up on would answer the next one instead. The connection therefore keeps at most one exchange at a time. Concurrent ping calls join the one in flight, keep-alive sends nothing while a caller's ping is outstanding, and a ping's timeout is treated as a lost connection rather than a stray reply — after it, the next PINGRESP could no longer be attributed, so the connection is failed and every caller told the same cause.

Writing is serialised

A packet may be several writes — a header, then a body — and two publishes issued at once must not interleave halfway. So a connection's writes go through one chain: each write waits for the last to finish. The cost is that a connection's throughput is serialised; the alternative is a packet the broker reads as two.

Timeouts and abandoned promises

An acknowledgement waiter arms its timeout when it is created, and the write that should provoke the acknowledgement follows. Because writes are serialised, a write can take longer than the timeout. Two things follow from that. First, the call has to settle on the timeout rather than wait out a write that may never finish, so the write and the acknowledgement are awaited together and whichever fails first ends the call. Second, the waiter's rejection can land before the caller awaits it, so it carries a handler from creation — the failure waits for the caller's await rather than reaching the runtime as an unhandled rejection, which this runtime treats as fatal. A local failure that never reaches the socket — an unencodable packet — is the same shape: the waiter is withdrawn immediately, since the connection, never having seen the bytes, will not clear it.

When the timeout wins a write that is still in flight, the connection is failed rather than left alive. The bytes are already with the transport, so the packet may yet reach the broker, and the stuck write still holds the serialised chain — every later packet would queue behind one that may never finish. Failing closes the socket, releases the chain, and means the failure the caller was told about cannot still be delivered behind it. There is no way to un-send bytes already handed to a transport, so the honest answer is to stop trusting the connection.

A deliberate close is not that case. A DISCONNECT queued behind a write the peer has stopped reading would wait for it, so a close that finds a write in flight aborts the transport instead — skipping the DISCONNECT — and settles the in-flight operation with Closed. That is a stop the caller asked for, not a failure, so nothing is latched: failure stays null. #closing draws the line, which is also why the read loop's own end during a close is not read as a lost connection. A close with nothing in flight still sends the DISCONNECT and then closes the socket.

Tearing the connection down has to reach the socket, not the writer. On the socket transport the writer is closed before the socket, so the peer sees a tidy half-close — but a writer's close waits for a pending write, and a write the peer has stopped reading never finishes. Closing the writer would therefore hang, and the socket close that interrupts the write would never be reached. A failure, and a close with a write in flight, close the socket directly instead: that tears the connection down and settles the write underneath it. A close with nothing pending still goes through the writer, so the tidy half-close is kept where it is safe.

Request/reply, without a reply queue

The usual RPC shape costs a queue per caller. MQTT 5.0's convention removes it: the request carries a response topic and correlation data, the responder publishes its answer to that topic carrying the same data, and the caller matches on it. request subscribes to a private response topic (esrun/reply/<token>), publishes, and resolves the first reply whose correlation data matches — so a stray or duplicated reply is dropped rather than mistaken for the answer. It is 5.0-only, and refuses on 3.1.1 rather than silently turning into a fire-and-forget publish. The topic is deliberately not under $: that space belongs to the broker, and the specification says a broker SHOULD stop clients exchanging messages on it (4.7.2), so a $ reply topic is one a compliant broker may silently starve. Its token and the correlation data come from crypto.getRandomValues, because a client that could guess the topic could answer in the responder's place.

The broker's limits are kept, not discovered

A 5.0 CONNACK says what the broker will accept: a Receive Maximum of QoS 1 and 2 publishes in flight, a Maximum QoS (AWS IoT Core allows 1), whether retained messages and wildcard, shared or identified subscriptions exist, and a Maximum Packet Size. The specification makes each a MUST for the client, and the broker's answer to a violation is to disconnect — so one bad publish took every other operation on the connection with it. They are checked before anything is written: a publish past the Receive Maximum waits for a slot, within its timeout, and everything else is refused with ERR_MQTT_UNSUPPORTED.

Nothing is dropped before a subscription exists

The read loop starts with the CONNACK, and a resumed session's backlog follows it at once — before connect() has returned, let alone before the application subscribed again. Those messages were owed an acknowledgement on arrival, so a client that dropped them lost them for good. They are now held until the client exists, and anything no subscription matches goes to onUnmatchedMessage, which can be given at connect time for exactly this reason. It plays the part of mqtt.js's client-wide message event, without giving up per-subscription iteration.

Testing without a port

esrun grants nothing by default, and a unit test should not be the reason to start a broker. So the Transport is an interface, and the connection and client sit on top of it: the unit suites drive a scripted broker in memory — it speaks the same packets, grants subscriptions and acknowledges publishes — with no socket and no --allow-net. The end-to-end suite then runs the same client against a real Mosquitto, where the framing either matches or nothing passes.

What it costs

The client is JavaScript, so the wire codec — the connection's buffer, each packet, each message — lives on the JavaScript heap. That is the same line runtime:db draws: Rust owns transport and the embedded engine, and JavaScript owns every network protocol, so a networked backend adds no runtime code. The cost is memory and the collector's time over it, not a capability — nothing here needs a grant beyond the socket it already had. Moving the codec into Rust later would not change what this package exports.

Not here

MQTT over WebSockets, the 3.1 wire format below 3.1.1, enhanced authentication (the AUTH packet), and persisting in-flight messages across a process restart: the session state the client keeps lives in memory, so a resumed session after a restart of the process itself relies on the broker alone.