feat: MQTT 5.0 outbound (broker-assigned) Topic Alias (#840)

.github/workflows/mqtt-compat.yml

@@ -100,7 +100,7 @@ jobs:
         if: always()
         run: kill "${BROKER_PID}" 2>/dev/null || true
 
-      - name: Warn on unexpected passes
+      - name: Fail on unexpected passes
         if: always()
         run: |
           [ -f result.json ] || exit 0
@@ -112,10 +112,13 @@ jobs:
                    if t.get("unexpected_pass")]
           if xpass:
               names = ", ".join(xpass)
-              print(f"::warning title=MQTT compat: update EXPECTED_GAPS::"
+              print(f"::error title=MQTT compat: update EXPECTED_GAPS::"
                     f"{len(xpass)} expected-gap test(s) now pass: {names}. "
                     f"Remove them from EXPECTED_GAPS in "
                     f"tools/mqtt-compat/run_compat.py.")
+              # Fail the job: a stale gap list (a now-implemented feature still
+              # marked as an expected gap) must not slip through on a green check.
+              raise SystemExit(1)
           PY
 
       - name: Job summary

aedes.js

@@ -30,6 +30,14 @@ const defaultOptions = {
   // MQTT 5.0: maximum Topic Alias value the broker accepts from a client.
   // 0 disables inbound topic aliases (the value advertised in CONNACK).
   topicAliasMaximum: 0,
+  // MQTT 5.0: broker-side ceiling on how many Topic Aliases the broker assigns per
+  // connection on OUTBOUND PUBLISHes (the effective max is min(this, the client's
+  // advertised Topic Alias Maximum)). Bounds the never-evicting per-connection
+  // table. Defaults to 0 (disabled, OPT-IN) to match the inbound `topicAliasMaximum`
+  // sibling: enabling it changes wire behaviour (clients start receiving empty-topic
+  // PUBLISHes) and adds per-connection memory, so it should not turn on silently on
+  // upgrade. Set it (e.g. 64) to enable. [#840]
+  outboundTopicAliasMaximum: 0,
   // MQTT 5.0: maximum size (bytes) of a packet the broker accepts. 0 = no
   // limit (and nothing advertised in CONNACK).
   maximumPacketSize: 0,
@@ -83,7 +91,23 @@ export class Aedes extends EventEmitter {
     this.maxClientsIdLength = opts.maxClientsIdLength
     // clamp to a safe [1, 100] range; see MAX_TOPIC_LEVELS
     this.maxTopicLevels = Math.min(Math.max(opts.maxTopicLevels, 1), MAX_TOPIC_LEVELS)
-    this.topicAliasMaximum = opts.topicAliasMaximum
+    // Clamp like its outbound sibling below: it is advertised as an int16 CONNACK
+    // property, so a non-integer / out-of-range value (65536, Infinity, a JSON-config
+    // string) would make mqtt-packet stream.destroy() every v5 CONNACK — a
+    // self-inflicted v5 outage — and Infinity would also lift the `alias > max` bound
+    // in resolveTopicAlias. A bad value disables inbound aliasing (0).
+    this.topicAliasMaximum = Number.isInteger(opts.topicAliasMaximum) && opts.topicAliasMaximum > 0
+      ? Math.min(opts.topicAliasMaximum, 65535)
+      : 0
+    // Coerce a non-integer / negative value to 0 (disabled), not to the default:
+    // "bad value" must never mean "silently enabled". A string '0' from env/JSON
+    // config, -1, or 0.5 all disable outbound aliasing rather than turning it on.
+    // Clamp the top to 65535 — Topic Alias is a Two Byte Integer, so a larger value
+    // (e.g. via a preConnect hook raising the client's advertised max) would make
+    // Math.min yield an out-of-range alias mqtt-packet's int16 writer can't encode.
+    this.outboundTopicAliasMaximum = Number.isInteger(opts.outboundTopicAliasMaximum) && opts.outboundTopicAliasMaximum > 0
+      ? Math.min(opts.outboundTopicAliasMaximum, 65535)
+      : 0
     this.maximumPacketSize = opts.maximumPacketSize
     this.receiveMaximum = opts.receiveMaximum
     this.sessionExpiryIntervalLimit = opts.sessionExpiryIntervalLimit

docs/Aedes.md

@@ -24,6 +24,7 @@
   - [Event: willDropped](#event-willdropped)
   - [Event: sessionExpired](#event-sessionexpired)
   - [Event: sessionLimitReached](#event-sessionlimitreached)
+  - [Event: outboundTopicAliasExhausted](#event-outboundtopicaliasexhausted)
   - [aedes.handle (stream)](#aedeshandle-stream)
   - [aedes.subscribe (topic, deliverfunc, callback)](#aedessubscribe-topic-deliverfunc-callback)
   - [aedes.unsubscribe (topic, deliverfunc, callback)](#aedesunsubscribe-topic-deliverfunc-callback)
@@ -52,7 +53,8 @@
   - `id` `<string>` aedes broker unique identifier. __Default__: `uuidv4()`
   - `connectTimeout` `<number>` maximum waiting time in milliseconds waiting for a [`CONNECT`][CONNECT] packet. __Default__: `30000`
   - `keepaliveLimit` `<number>` maximum client keep alive time allowed, 0 means no limit. For MQTT 5.0 clients exceeding this, the broker sends a `Server Keep Alive` in the CONNACK and uses it instead of rejecting the connection. __Default__: `0`
-  - `topicAliasMaximum` `<number>` MQTT 5.0 only. Maximum inbound Topic Alias value the broker accepts from a client, advertised in the CONNACK. `0` disables inbound topic aliases. __Default__: `0`
+  - `topicAliasMaximum` `<number>` MQTT 5.0 only. Maximum __inbound__ Topic Alias value the broker accepts from a client, advertised in the CONNACK. `0` disables inbound topic aliases. A non-integer or negative value coerces to `0` (disabled) and a value above `65535` clamps to `65535` (the Two Byte Integer ceiling), so a bad value never silently enables a broken maximum. __Default__: `0`. (See also `outboundTopicAliasMaximum` for the reverse direction, and the callout under [`Event: connackSent`](#event-connacksent).)
+  - `outboundTopicAliasMaximum` `<number>` MQTT 5.0 only. Broker-side ceiling on how many Topic Aliases the broker assigns per connection on the __outbound__ PUBLISHes it sends to a v5 client. The effective per-connection maximum is `min(this, the client's advertised Topic Alias Maximum)`. Bounds the never-evicting per-connection alias table (keyed by full topic strings), so an untrusted client can't drive it toward 65535 entries. Opt-in (like `topicAliasMaximum`): `0` disables the broker's alias __assignment__ entirely. Set it (e.g. `64`) to enable — enabling it changes wire behaviour (clients start receiving empty-topic PUBLISHes) and adds per-connection memory. A non-integer or negative value (e.g. the string `'64'` arriving from env/JSON config) coerces to `0` (disabled — a bad value must never mean "silently enabled") and a value above `65535` clamps to `65535`. __Default__: `0`. Note that whatever this is set to, a caller-supplied `topicAlias` on an outbound PUBLISH (via `client.publish()`) and on a Will is always stripped — the broker owns the outbound alias namespace (see the callout under [`Event: connackSent`](#event-connacksent)).
   - `maximumPacketSize` `<number>` MQTT 5.0 only. Maximum size in bytes of a packet the broker accepts, advertised in the CONNACK. Enforced: an oversized inbound frame is rejected as soon as its declared length is known — before the rest of its payload is buffered — with a `DISCONNECT` (reason code `0x95`, Packet too large) for connected v5 clients, or a dropped connection with a `connectionError`/`clientError` for pre-auth or v3/v4 clients. `0` means no limit. __Default__: `0`
   - `receiveMaximum` `<number>` MQTT 5.0 only. Maximum number of in-flight QoS 1/2 PUBLISH packets advertised to the client in the CONNACK. __Advisory only__: the value is advertised but the broker does not currently enforce an inbound in-flight window (existing `drainTimeout` transport backpressure applies instead); enforcement is planned for a follow-up. `0` means it is not advertised (clients assume the protocol default of `65535`). __Default__: `0`
   - `sessionExpiryIntervalLimit` `<number>` MQTT 5.0 only. Upper bound in seconds on the Session Expiry Interval a client may request. A larger requested value — including `0xFFFFFFFF` ("never expires") — is clamped to this, both in CONNECT and in a DISCONNECT that updates the interval. Bounds how long per-client session-expiry and will-delay timers (and their persisted session state) live, limiting memory accumulation from a single source cycling client identities. `0` means no cap. __Default__: `0`
@@ -233,6 +235,10 @@ Emitted when server sends an acknowledge to `client`. Please refer to the MQTT s
 
 For MQTT 3.1/3.1.1 the packet carries a `returnCode` (`0` = success). For MQTT 5.0 it instead carries a `reasonCode` (`0x00` = success; the v3/v4 return codes map to the equivalent v5 reason codes) and may carry a `properties` object advertising negotiated capabilities (e.g. `topicAliasMaximum`, `maximumPacketSize`, `receiveMaximum`, `serverKeepAlive`, `assignedClientIdentifier`, `sharedSubscriptionAvailable`).
 
+> __MQTT 5.0 outbound Topic Alias:__ the broker's alias __assignment__ is opt-in via the [`outboundTopicAliasMaximum`](#new-aedesoptions) broker option (default `0` = off). When it is set __and__ a v5 client advertises its own `Topic Alias Maximum` in CONNECT, the broker assigns Topic Aliases on the PUBLISHes it sends to that client — full topic + alias on first use of a topic, empty topic + the same alias thereafter — to save bandwidth. Mappings are per Network Connection (reset on reconnect, never carried across, per §3.3.2.3.4) and the per-connection table never evicts. The number of aliases assigned is `min(client's advertised maximum, outboundTopicAliasMaximum)`. Full documentation is under the [`outboundTopicAliasMaximum`](#new-aedesoptions) option.
+>
+> One thing is __not__ gated by the option: a caller-supplied `topicAlias` — set on an outbound PUBLISH passed to `client.publish()`, or carried on a client's Will — is always stripped before the packet reaches another client, even when `outboundTopicAliasMaximum` is `0`. The outbound alias namespace is broker-owned; forwarding a caller value verbatim would emit an alias the receiving client never negotiated ([MQTT-3.1.2-27]) or one that disagrees with the broker's own table. So enabling the option adds broker-assigned aliasing; it does not turn on (or off) that scrub.
+
 ## Event: closed
 
 Emitted when server is closed.
@@ -257,6 +263,13 @@ MQTT 5.0 only. Emitted when a session that outlived its connection reaches its S
 
 MQTT 5.0 only. Emitted when the [`pendingSessionsLimit`](#new-aedesoptions) DoS guard trips: a disconnecting session that would otherwise be retained (including a never-expiring one) is instead expired immediately (`reason: 'sessionExpiry'`), or a delayed Will is published immediately instead of being held (`reason: 'willDelay'`). Listen for this to detect identity-cycling abuse — without it the guard would degrade behavior silently.
 
+## Event: outboundTopicAliasExhausted
+
+- `client` [`<Client>`](./Client.md)
+- `info` `<object>` `{ max: number }` — the effective per-connection alias maximum (`min(client's advertised Topic Alias Maximum, outboundTopicAliasMaximum)`) that has been filled
+
+MQTT 5.0 only. Emitted once per connection, the first time the broker sends a PUBLISH for a __new__ topic that can no longer be aliased because that connection's [`outboundTopicAliasMaximum`](#new-aedesoptions) table is full. The broker keeps working — it falls back to sending the full topic name (the table never evicts) — but throughput for further distinct topics loses the alias saving. Listen for this to detect a `outboundTopicAliasMaximum` set too low for a client's topic cardinality; without it the degradation is silent.
+
 ## aedes.handle (stream)
 
 - stream: `<net.Socket>` | `<stream.Duplex>`

lib/client.js

@@ -36,7 +36,20 @@ class Client {
 
     // MQTT 5.0 inbound Topic Alias map (alias number -> topic name), scoped
     // to this connection. Populated as the client registers aliases.
-    this._topicAliases = new Map()
+    this._inboundTopicAliases = new Map()
+    // MQTT 5.0 outbound Topic Alias state (broker -> client), negotiated at
+    // CONNECT. Allocated once here — like its inbound sibling `_inboundTopicAliases`
+    // above — rather than lazily in connect.js: write()'s hot path then always reads
+    // a Map (never null↔Map polymorphism), and teardown `.clear()`s it instead of
+    // hand-maintaining a "max > 0 ⇒ Map" invariant. `_outboundTopicAliasMaximum`
+    // stays the on/off switch: 0 (the default) means write() short-circuits before
+    // it ever touches the map, so an empty map on a non-aliasing client is inert.
+    this._outboundTopicAliasMaximum = 0
+    this._outboundTopicAliases = new Map()
+    // One-shot latch for the `outboundTopicAliasExhausted` telemetry event: set the
+    // first time a new topic can't be aliased because the per-connection table is
+    // full, so the event fires once per connection rather than per PUBLISH.
+    this._outboundTopicAliasExhausted = false
 
     this.broker = broker
     this.conn = conn
@@ -418,6 +431,20 @@ class Client {
     conn.removeAllListeners('readable')
 
     this._parser._queue = null
+    // Release the per-connection Topic Alias tables promptly (like _parser._queue),
+    // shortening the retention window if anything still references a closed client
+    // (conn.client backref, in-flight mqemitter callbacks, a takeover race). [#840]
+    //
+    // `connected` isn't cleared until finish() runs (after the async
+    // handleUnsubscribe), so an in-flight deliver0/deliverQoS (both setImmediate)
+    // can still reach write() in this window. Zero the outbound MAXIMUM first so
+    // withOutboundTopicAlias short-circuits on `!max` before it touches the table,
+    // then `.clear()` both maps (kept as Maps, never nulled — a .get() on an empty
+    // map safely returns undefined, so an in-flight write can't throw on a stale
+    // reference). Mirrors the inbound map's teardown exactly.
+    this._outboundTopicAliasMaximum = 0
+    this._outboundTopicAliases.clear()
+    this._inboundTopicAliases.clear()
 
     if (this._keepaliveTimer) {
       clearTimeout(this._keepaliveTimer)

lib/handlers/connect.js

@@ -303,6 +303,31 @@ function init (client, packet, done) {
   // (Consistent posture; enforcement is a planned follow-up.)
   client.maximumPacketSize = packet.properties?.maximumPacketSize // bytes, broker->client
   client.receiveMaximum = packet.properties?.receiveMaximum ?? RECEIVE_MAXIMUM_DEFAULT // in-flight QoS 1/2
+  // [#840] MQTT 5.0 outbound Topic Alias: the client's Topic Alias Maximum is the
+  // highest alias value it will accept from the broker (§3.1.2.11.5). When > 0,
+  // the broker MAY alias outbound PUBLISH topics; track the per-connection map.
+  // The effective per-connection maximum is min(client-advertised, broker option)
+  // — the broker's `outboundTopicAliasMaximum` bounds the never-evicting,
+  // client-keyed table (0 disables outbound aliasing entirely). Require a positive
+  // INTEGER: mqtt-packet decodes a duplicated property to an array and a preConnect
+  // hook can set a fractional/negative value on the mutable packet; the `> 0`
+  // integer gate matches write.js's `!max` so a bad value yields 0 at both sites.
+  //
+  // A duplicated Topic Alias Maximum (§3.1.2.11.5 Protocol Error) is deliberately
+  // handled LENIENTLY here — the non-integer array simply disables outbound aliasing
+  // for the connection rather than refusing the CONNECT with 0x82. Outbound aliasing
+  // is a broker-side MAY optimisation the client can live without, so degrading it is
+  // safer than dropping an otherwise-valid connection over a malformed optional
+  // property; the receiveMaximum/authenticationData Protocol Errors above stay strict
+  // because those affect correctness, not a bandwidth optimisation.
+  const advertisedTopicAliasMax = packet.properties?.topicAliasMaximum
+  // Just the on/off switch + effective cap; the Map itself is allocated once in the
+  // Client constructor and stays empty until write() assigns the first alias. `max`
+  // is min(client's advertised Topic Alias Maximum, broker option), 0 when either is
+  // absent (feature off for this connection).
+  client._outboundTopicAliasMaximum = packet.protocolVersion === 5 && Number.isInteger(advertisedTopicAliasMax) && advertisedTopicAliasMax > 0
+    ? Math.min(advertisedTopicAliasMax, client.broker.outboundTopicAliasMaximum)
+    : 0
   // MQTT 5.0 Request Problem Information [MQTT-3.1.2-29]: when false the broker
   // must not send a Reason String or User Properties on packets other than
   // PUBLISH/CONNACK/DISCONNECT. mqtt-packet decodes the byte property as a
@@ -313,6 +338,19 @@ function init (client, packet, done) {
   // false — the broker only returns it on explicit request.
   client._requestResponseInformation = packet.properties?.requestResponseInformation === true
   client._will = packet.will
+  // [#840] A Will carries client-supplied properties, and mqtt-packet round-trips
+  // Topic Alias (0x23) among them. A Topic Alias is meaningless on a stored Will
+  // (it names an entry in a since-gone connection's alias table) and, once the Will
+  // fans out to other subscribers, an attacker-chosen value would hijack an alias
+  // slot in their tables. The write path scrubs it defensively, but drop it at the
+  // source too — the inbound-PUBLISH analogue lives at publish.js resolveTopicAlias.
+  // Assign `undefined` rather than `delete`, matching the sibling scrubs at
+  // publish.js:180 / write.js:29: the will `properties` object is reference-shared
+  // into every fanout packet, and `delete` would transition it into dictionary mode,
+  // deopting property access for all recipients.
+  if (client._will?.properties?.topicAlias !== undefined) {
+    client._will.properties.topicAlias = undefined
+  }
 
   runSeries(
     new ClientPacketStatus(client, packet),

lib/handlers/publish.js

@@ -150,7 +150,12 @@ function authorizePublish (packet, done) {
 function resolveTopicAlias (client, packet) {
   const alias = packet.properties.topicAlias
   const max = client.broker.topicAliasMaximum
-  if (alias < 1 || alias > max) {
+  // A duplicated Topic Alias property decodes to an array (mqtt-packet), and
+  // `[1,2] < 1` / `> max` are both false via NaN — so a non-integer would slip the
+  // range check below and `.set()` a fresh array key per PUBLISH, growing the map
+  // without bound (bypassing topicAliasMaximum entirely). Require an integer first;
+  // this is the inbound sibling of the guard connect.js applies to the outbound max.
+  if (!Number.isInteger(alias) || alias < 1 || alias > max) {
     return new Error(`topic alias ${alias} is out of range (broker topicAliasMaximum is ${max})`)
   }
 
@@ -159,10 +164,10 @@ function resolveTopicAlias (client, packet) {
     // store a wildcard/invalid topic in the connection's alias map. An invalid
     // topic still falls through to handlePublish's checks, which reject it.
     if (packet.topic.indexOf('#') === -1 && packet.topic.indexOf('+') === -1) {
-      client._topicAliases.set(alias, packet.topic)
+      client._inboundTopicAliases.set(alias, packet.topic)
     }
   } else {
-    const topic = client._topicAliases.get(alias)
+    const topic = client._inboundTopicAliases.get(alias)
     if (topic === undefined) {
       return new Error(`unknown topic alias ${alias}`)
     }