(2.15) [IMPROVED] MQTT: Pipeline QoS2 PUBLISH and PUBREL processing

[levb]

QoS2 ingest blocked the readLoop on four synchronous JetStream round-trips per message: the PUBLISH staged its copy and waited (one), and the PUBREL loaded that copy back, deleted it waiting for the ack, and stored it for delivery (three more).

This extends the QoS1 pipeline treatment to QoS2. Where QoS1 has one acknowledged JetStream interaction per message (PUBLISH -> store -> PUBACK), QoS2 has two: PUBLISH -> staged store -> PUBREC, then PUBREL -> delivery store + staged-copy discard -> PUBCOMP. Both are now submitted asynchronously and acknowledged from the per-connection ack pipeline as the JetStream acks arrive, in packet order. The discard is fire-and-forget: by stream sequence where a load supplied one, else by the unique staging subject - needing no sequence is what made dropping the load possible.

Submitting asynchronously costs the readLoop its read-your-own-writes view of JetStream, so the connection keeps the state it needs for deduplication while a JS interaction is in flight, in one PI-keyed map of its QoS2 exchanges (qos2Exchanges):

• A non-nil entry is a copy of a pending PUBLISH, staged store to PUBREL. Duplicate PUBLISHes dedupe against it (first copy wins, mirroring the staging stream's max-msgs-per-subject; each still gets its PUBREC), and the delivery on PUBREL needs no JetStream load at all. Copies are bounded by the pipeline window; extras are not kept.
• A nil entry marks a released exchange whose staged-copy discard is still in flight: a retransmitted PUBREL is answered with just the PUBCOMP, where a JetStream load could race the discard and deliver a duplicate [MQTT-4.3.3-1]. A new PUBLISH on the PI replaces the entry (a new publication per the spec).

JetStream remains the source of truth: on a miss (a session resumed on a new connection, or past the cap) the PUBREL falls back to the JetStream load, as the synchronous code always did.

The pipeline entries now carry their response packet type (PUBACK, PUBREC or PUBCOMP); the single FIFO preserves the per-packet-type ordering of [MQTT-4.6]. One behavior change: a PUBREL that loads an invalid staged message now fails the connection without first emitting the PUBCOMP that the old code sent via defer, and discards the corrupt message, so the PUBREL the client retries converges to a clean PUBCOMP.

Added tests: PUBREC/PUBCOMP burst ordering with exactly-once delivery (TestMQTTQoS2AckPipelineOrder), and connection close with unreleased exchanges in flight - no leaked JSA reply registrations, released messages delivered, unreleased ones not
(TestMQTTQoS2AckPipelineConnClose); a corrupt staged message failing the connection once, then converging (TestMQTTQoS2PubRelInvalidStagedMessage); and back-to-back PI reuse across releases with the discard still in flight (TestMQTTQoS2PIReuseAfterRelease). PUBREL-retransmission dedupe and PI reuse are also exercised by the existing
TestMQTTQoS2RejectPublishDuplicates, which fails without the released marks.

Measured on a real 3-node R3 cluster (GCP c2d, crossover methodology, QoS0 control 1.00, per-pass ratios agreeing): single connection, QoS2, 256-message window, ~690/s to ~18.1K/s (26x); p99 ack latency 384ms to 19ms. Before is flat at ~600-690/s at every window (four serialized JetStream round-trips per message); after rises monotonically with the client's window (1.8x at 1, 10x at 10, 15x at 20). The window-1 gain is pipelining-free: dropping the load and the waited delete halves the JetStream round-trips per exchange, so even strict stop-and-wait clients - which the QoS1 change could not help - benefit; the rest of the curve is the pipelining. Aggregate many-connection QoS2 throughput improves 1.4-1.5x (~12-13K/s to ~17.8K/s at 100/200 connections): unlike QoS1, pipelining moves the aggregate too, because it also removes per-message JS API round-trips, not just waiting. QoS1 (sweep, aggregate, MQTT.js) and QoS0 are unaffected (~1.0x). The remaining gap to QoS1's aggregate (~18K/s vs ~68K/s) is structural: two stream appends plus a discard per message, all on the single $MQTT_msgs R3 stream.

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 6262eb7cfb

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[neilalexander]

@claude review

[synadia-claude-reviewer]

Claude finished @neilalexander's task in 6m 12s —— View job

---

Code Review

• Initialize review
• Read CLAUDE.md for repo guidelines
• Analyze PR diff (mqtt-qos1-pipeline...HEAD)
• Review server/mqtt.go changes
• Review server/mqtt_test.go changes
• Post comprehensive review findings

---

This PR extends the QoS1 ack-pipeline approach to QoS2, eliminating 2–3 synchronous JetStream round-trips per exchange and adding an in-memory dedupe layer via qos2Exchanges. The implementation is solid overall. Notes below ordered by severity.

Correctness of core invariants

State machine (qos2Exchanges)

The three states per PI — absent, non-nil (pending), nil (released) — are correctly maintained:

Transition	Method	Effect
First PUBLISH	mqttRecordQoS2Publish	absent → non-nil
Duplicate PUBLISH (same conn)	mqttRecordQoS2Publish	non-nil → non-nil (kept, returns early)
DUP PUBLISH (reconnect, unknown)	mqttRecordQoS2Publish	absent → absent (returns early, defer to JS)
PUBREL delivers	mqttMarkQoS2Released	non-nil → nil, count--
PUBREL retransmit	mqttTakeQoS2Publish	nil → (nil, released=true), short-circuit
New PUBLISH on released PI	mqttRecordQoS2Publish	nil → absent via delete, then → non-nil

All transitions look correct, including the reconnect case fixed in commit 603d31a. The else if pp.flags&mqttPubFlagDup != 0 guard (line 4836) correctly handles the Codex P2 concern: a retransmitted PUBLISH with DUP set arriving on a fresh connection has no in-memory entry, so it falls into this branch and the PUBREL falls back to JetStream as the authoritative source of bytes. TestMQTTQoS2RejectPublishDuplicatesAcrossReconnect adequately covers this.

Discard-before-PUBCOMP ordering

The PR claims the discard is FIFO-safe against PI reuse because it's "enqueued before the PUBCOMP that frees the PI." This holds because:

1. Discard is pushed to jsa.sendq (before mqttDeliverInboundMsg)
2. Delivery store is pushed to jsa.sendq (via storeMsgAsync in mqttPipelineStoreThenAck)
3. PUBCOMP is emitted only after the delivery store's JetStream ack arrives

Both the discard and delivery store travel over the same sendq → NATS connection → JetStream server pipeline. NATS delivers messages from a single connection in order, so the discard reaches JetStream before the delivery store. PUBCOMP can only be emitted after the delivery store acks, so it's always emitted after the discard is processed. The ordering is sound. ✓

deleteMsgBySubject using stream purge

A subject-filtered stream purge rather than a sequence-based delete is a neat trick that eliminates the load entirely. One nuance worth checking: if a deleteMsgBySubject purge races with a new PUBLISH's staging store on the same subject (fast PI reuse), the sendq ordering guarantees the purge is processed first (it's enqueued at PUBREL time, while the new PUBLISH can only arrive after PUBCOMP, which is after PUBCOMP is emitted, which is after the delivery store ack which is submitted after the purge). The test TestMQTTQoS2PIReuseAfterRelease exercises this.

Observations / minor points

qos2PendingCount >= mqttMaxAcksInFlight guard is never triggered in practice

mqttRecordQoS2Publish is called before mqttPipelineStoreThenAck. The latter blocks in mqttPipelinePush when the pipeline channel (capacity mqttMaxAcksInFlight) is full. So when qos2PendingCount would reach mqttMaxAcksInFlight, the readLoop is already blocked on the pipeline — no new mqttRecordQoS2Publish call is possible. The guard is still good defense-in-depth, but it's effectively unreachable under normal operation. A comment noting this would clarify intent.

Released-entry accumulation

nil entries remain in qos2Exchanges for the duration of the connection (see comment: "released entries last the connection — the discard is fire-and-forget, there is no safe moment to drop them"). This is correct but means a client cycling through all 65535 PIs would build a 65535-entry map. Bounded and acceptable in practice, but worth noting in the comment.

mqttMarkQoS2Released unconditionally initializes the map

func (c *client) mqttMarkQoS2Released(pi uint16) {
    if c.mqtt.qos2Exchanges == nil {
        c.mqtt.qos2Exchanges = make(map[uint16]*mqttPublish)
    } else if ...
    c.mqtt.qos2Exchanges[pi] = nil
}

This is called even on the JetStream fallback path for PIs that were never recorded. The nil entry is still correct (it screens subsequent retransmitted PUBRELs), and the allocation is a one-time cost, so this is fine as written.

exceptDuplicateError called in two places

Both mqttAckLoop (line 4602) and mqttPipelinePush's synchronous first-entry fast-path (line 4682) call exceptDuplicateError. This is correct and ensures consistent handling whether or not the pipeline goroutine has been started yet. The shared helper method makes the symmetry clear.

Test coverage assessment

Scenario	Test
PUBREC/PUBCOMP burst ordering, exactly-once	TestMQTTQoS2AckPipelineOrder
Connection close with in-flight exchanges	TestMQTTQoS2AckPipelineConnClose
Corrupt staged message → connection fail → retry converges	TestMQTTQoS2PubRelInvalidStagedMessage
PI reuse after release with discard in flight	TestMQTTQoS2PIReuseAfterRelease
DUP PUBLISH across reconnect (Codex P2 fix)	TestMQTTQoS2RejectPublishDuplicatesAcrossReconnect
Same-connection PUBREL retransmit dedupe	existing TestMQTTQoS2RejectPublishDuplicates

Coverage is comprehensive. One scenario not explicitly tested: a PUBREL arriving after the pipeline window is at mqttMaxAcksInFlight (triggering the overflow timeout path). This is an edge case and the pipeline behaviour there is inherited from the QoS1 code, so it may not need a dedicated test.

Summary

The implementation is well-designed and correct. The key invariants — FIFO discard-before-PUBCOMP ordering, the DUP reconnect fallback to JetStream, and the released-entry screen against retransmitted PUBRELs — all hold. No blocking issues found.

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 9ebd6fcc46

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[chatgpt-codex-connector]

Keep QoS2 release state until discard is durable

When a QoS2 publisher sends PUBREL and the connection drops before this fire-and-forget discard reaches JetStream, the only released marker is the connection-local qos2Exchanges entry and is lost on reconnect. If the client reconnects, especially to another server in a cluster where this send queue does not order the old discard before the new loadLastMsgFor, the retried PUBREL can load the still-staged copy and deliver it again even though the first connection already called mqttDeliverInboundMsg (immediately visible for QoS0 subscribers, or after the delivery store succeeds). The discard needs to be confirmed or the release marker made reconnect-visible before falling back to loading staged messages.

Useful? React with 👍 / 👎.