perf(mqtt): share payload and QoS 0 packet encoding across topic subscribers

[kriszyp]

What

When one message is published to an MQTT topic, every subscriber independently serialized the payload and generated its own PUBLISH packet — even though the output was byte-identical. This computes both once and reuses them across the fan-out.

• Payload — serialized once per (message, record version, negotiated content type).
• QoS 0 packet — additionally shared per (payload, topic, protocol version). A QoS 0 PUBLISH carries no message identifier, so the whole packet is identical across those subscribers.
• QoS 1/2 — packet generation stays per subscriber; the 2-byte message identifier in the variable header differs. A follow-up could pass a small per-subscriber header plus the shared payload as a 2-element array to writev (which already accepts a buffer list) to avoid the copy; out of scope here.

Why

Measured, not assumed. A V8 heap snapshot of one worker on a production node at ~127k cluster connections found 157,888 live Subscription objects across only 29 distinct topics — the busiest topics each had ~7,200 subscribers, i.e. ~7,200 structurally identical delivery pipelines per topic. A 6s CPU profile of a worker at 182k connections put essentially every non-GC sample on the per-delivery path (writev 16.4%, the delivery loop 9.8%, buffer _copyActual 9.3%, mqtt.js listener 5.6%, mqtt-packet generate/concat ~2%), with GC at 26.4% — largely the short-lived allocation churn from that duplicated encoding. Harper's CPU scales with deliveries (messages × subscribers), not connections.

Is sharing safe?

serializeMessage(data, request) output varies only on the negotiated content type. It resolves a serializer via findBestSerializer(request), which reads only the Accept header, and every registered serializer (JSONStringify, cbor encode, msgpackr pack) is a pure function of the message. Nothing in the MQTT delivery path applies per-user shaping: attribute-level permissions (attribute_permissions → target.select) are a REST query-time concern, and the checkPermission gate on subscribe is all-or-nothing eligibility, not per-message redaction. The value delivered is auditRecord.getValue(...) or entry.value straight from the store, unshaped by subscriber identity. So a shared payload cannot cross users.

Sharing is keyed on the message object's identity, which works because the fan-out has a single shared origin — transactionBroadcast dispatches one object to every subscription of a key. Because a custom Resource is not bound by that, the event's record version is part of the key: an entry whose version does not match is re-encoded, and an event with no version is not shared at all. So a Resource that reuses one mutable envelope is correct as long as the version advances, rather than merely being told not to.

Numbers

benchmarks/mqtt-fanout-encoding.js, 200 messages, ms per publish (JSON, MQTT v3.1.1):

subscribers	QoS 0 before	after		QoS 1 before	after
1	0.0167	0.0041		0.0023	0.0026
100	0.2044	0.0082		0.1613	0.0819
1000	1.3841	0.0310		1.4572	0.7574

QoS 0 per-publish cost goes from linear in N to roughly flat (44x at 1000 subscribers); QoS 1 gets ~2x from payload sharing alone. At N=1 it is break-even within noise. CBOR shows a larger win (~80-130x at 1000). The benchmark measures encoding wall-time only — the socket writes that follow are unchanged and per-subscriber either way.

Retention

Sharing needs the encoding to outlive one delivery but not the message object — for record events that object is the store's cached record, so an unbounded memo would keep serialized copies and PUBLISH buffers resident for as long as the record stays cached. Two rotating WeakMap generations bound it in time (10s interval, unref'd, stopped when nothing is retained), and a 4 MB per-generation byte budget bounds it in volume.

Test notes

• Unit (unitTests/server/serverHelpers/sharedMessageEncoding.test.js, 36 tests) — sharing per content type, content-type isolation, the version key invalidating a whole chain, pass-through messages skipping negotiation, retry after a failed serialization, rotation and exact byte accounting, and a shared buffer surviving ws.send to five real clients unmutated.
• Premise (unitTests/resources/subscriptionValueIdentity.test.js) — two subscribers of a record write receive the same value object, and consecutive versions receive distinct objects. Without this, a change making each subscription decode its own copy would silently revert the optimization with every other test green.
• Integration (integrationTests/mqtt/mqtt.test.ts, 5 new) — including one that proves the server serializes once per publish regardless of subscriber count, via a fixture content type that stamps the server's serialization counter into the payload. Byte equality holds whether or not sharing works, so this is the only test that can catch a regression to per-subscriber encoding; verified it fails with sharing disabled and passes with it on, while the other four pass either way.
• Suites: test:unit:main, test:unit:resources (1349), server + serverHelpers (531), test:integration:all (1640 pass / 0 fail). Pre-existing environment failures unrelated to this change are noted in the review.

Review

Seven rounds of independent cross-model review (Codex + Gemini + Grok + Harper-domain adjudication) ran pre-push; the final round returned no majors. Findings fixed along the way included a 406 regression for pass-through messages, a memoized rejection that turned one transient serialization failure permanent, unbounded retention, and a benchmark that had stopped measuring the shared path. Remaining accepted trade-offs are in the self-review.

Dispatched task: no linked issue.

🤖 Generated with Claude Code
\n\n### CI follow-up\n\nBun 1.3.14 kept a private resolver cache across component redeploys and its CommonJS resolver could not resolve import-only package exports. The follow-up tracks resolution candidates and package entries explicitly. The focused redeploy suite passes 18/18 under Node and Bun; the exact Bun 4/6 shard passes 138 with 0 failures.

[gemini-code-assist]

Code Review

This pull request introduces a shared message encoding mechanism to optimize MQTT fan-out delivery by serializing messages once per content type and sharing QoS 0 PUBLISH packets across subscribers of a topic. It includes comprehensive benchmarks, integration tests, and unit tests to validate the performance improvements and ensure message identity contracts. The review feedback suggests two improvements: using a more robust check for thenable objects in sharedMessageEncoding.ts to prevent false positives, and adhering to the repository style guide in subscriptionValueIdentity.test.js by using the node: prefix for builtin imports and including file extensions on relative imports.

[gemini-code-assist]

When detecting a Promise or thenable object, use typeof value?.then === 'function' instead of a simple truthiness check on value?.then to prevent false positives from non-Promise objects that have a truthy then property.

Suggested change

	if ((payload as any)?.then)
	if (typeof (payload as any)?.then === 'function')

References

1. When detecting a Promise or thenable object, use typeof value?.then === 'function' instead of a simple truthiness check on value?.then to prevent false positives from non-Promise objects that have a truthy then property.

[gemini-code-assist]

Node builtins must use the node: prefix as per the repository style guide constraints. Additionally, relative imports must include the file extension.

Suggested change

	const assert = require('assert');
	const { setupTestDBPath } = require('../testUtils');
	const assert = require('node:assert');
	const { setupTestDBPath } = require('../testUtils.js');

References

1. Node builtins use the node: prefix. These are constraints, not style choices. Relative imports must include the file extension. ^(link)

[claude]

Reviewed; no blockers found.