fix(binding-mqtt-kafka): account for the retained stream when granting the publish window

[jfallows]

Description

Fixes #2518

When a route has a retained topic configured, MqttKafkaPublishFactory maintains two downstream Kafka streams per publish connection: messages (the main topic — every publish goes here) and retained (the retained-messages topic — even a non-retained publish still issues a doKafkaFlush on this stream carrying the message's full reserved bytes, so its budget shrinks in lockstep with messages). doMqttWindow combines both streams' windows into the one granted upstream to the MQTT client, and had two independent defects:

1. Wrong gate. The retained-aware branch was guarded by hasPublishFlagRetained(publishFlags) — a per-message flag, true only while a message carrying RETAIN is being processed and reset after every FIN. In steady state that branch was effectively dead code. Fixed to gate on retainAvailable, a per-stream capability decided once at BEGIN, so the retained stream's budget is factored in for the stream's whole life.

2. Wrong arithmetic. Even when reachable, the computation was min(messages.ack, retained.ack) paired with max(messages.max, retained.max). But messages holds its ack pinned at zero and grows its max, while retained advances its ack against a fixed max — two different representations of "room left". Minimizing ack and maximizing max independently combines the ack of one stream with the max of the other, describing neither stream's actual budget, and (since retained.max never changes) permanently caps the connection's lifetime publishable bytes once it exceeds messages.max, rather than bounding bytes in flight. Fixed to compute each stream's own available budget uniformly as max - (seq - ack), take the smaller of the two, and express it against the chosen max via a monotonic ack.

Credit to community contributor @sfr-oc, who diagnosed and fixed this as part of the combined #2523. This PR cherry-picks both commits standalone (unmodified, 98148a89 and 208f10f6), since #2523 bundles seven independently-scoped defects across three bindings into one PR — each deserves its own focused review, and this one is ready on its own.

Test coverage

The first commit's own scenario (publish.many.messages.retain.available) still passed with the second commit reverted: both messages and retained are accepted children of the same accept "zilla://streams/kafka0" in the k3po test stub, so they always share the identical declared window and messages.max can never diverge from retained.max. I can show algebraically that the pre-fix formula reduces to the exact same expression as the fix whenever max_m == max_r, regardless of how far the ack values diverge — so the defect was structurally unreachable from that scenario as written.

Reproduced the divergence using existing k3po options, no new harness code: option zilla:update "handshake" on the shared accept stops the automatic acknowledge a STREAM-mode DATA read would otherwise trigger, so messages' own ack stays pinned at its initial handshake value once its max is explicitly grown via read option zilla:window. retained's flush-triggered acknowledge is unconditional regardless of update mode, so it keeps advancing normally against its original, unchanged max — reproducing exactly the "messages holds ack, grows max; retained advances ack, fixed max" mismatch the second commit's own description names.

• Modified scenario: specs/binding-mqtt-kafka.spec's streams/kafka/publish.many.messages.retain.available/server.rpt (2 additions, no new scenario needed — the existing runtime IT MqttKafkaPublishProxyIT#shouldPublishManyMessagesRetainAvailable and self-consistency KafkaIT#shouldPublishManyMessagesRetainAvailable both exercise it)

Verification

• Reverted just the second commit and confirmed the runtime IT now fails for the right reason: it times out publishing partway through (the pre-fix budget understates the real combined credit until the client can no longer make progress) — then restored the fix and confirmed it passes in ~1.3s
• Confirmed the self-consistency KafkaIT scenario still passes independent of the engine, so the two scripts remain mutually consistent
• Ran the full binding-mqtt-kafka and binding-mqtt-kafka.spec IT suites — both clean

🤖 Generated with Claude Code

https://claude.ai/code/session_015YVNaqKvEXZVzmg3HoGVnt

---

Generated by Claude Code