fix(nats): stop losing work before it is queued (CLO-4771)

[levivannoort]

All six findings from CLO-4771 — Epic 1 of the NATS queue migration readiness audit. A publish that vanishes never becomes a message, so no consumer metric can see it; these are the paths where that happens.

Two commits: the three P0s (connection death detection), then the three P1/P2s (frame integrity and write semantics).

---

P0 — Reconnect on a connection-closing -ERR

handleError() fired onError and threw, but never set status, closed the transport, or reconnected — and ProtocolException extends NatsException, not ConnectionException, so processMessage()'s reconnect catch never saw it. The connection stayed marked connected over a dead socket, and the next publish() could write into it and return void success: silent loss, reported as a send.

Adds a pure Connection::closesConnection() classifier and recycleDeadConnection(), which reconnects when allowReconnect is set and otherwise marks the connection disconnected so the next call raises.

Judgment call: authorization and authentication failures also close the connection server-side, but are excluded deliberately — retrying with credentials the server just rejected turns a hard failure into a hot loop. Worth a second opinion.

P0 — Keepalive, now actually driven

checkPings() had exactly one call site, processMessage(), reachable only from request()/wait()/nextMessage(). Nothing in cloud calls wait(), and there was no timer — so it also ran after request() had already written into the dead socket.

It is now driven from ensureConnected(), so the keepalive runs before the write, and Connection::tick() is public for a holder that never reads.

tick() alone was not enough, and this is the part worth reviewing. checkPings() only writes; it is the read path that clears outstandingPings, and a pooled publisher has none. Ticking such a connection would therefore march it to maxPingsOut and declare a perfectly healthy socket stale — the keepalive causing the outage it exists to prevent. tick() now collects the PONGs it is owed, through a readMessage() split out of processMessage() so the drain does not re-enter the keepalive check and out-ping itself.

The clock lives in the queue Swoole adapter, the one layer in the stack that owns a scheduler; nats and pools stay free of any dependency on a runtime. Pool::maintain() sweeps idle resources and ticks the ones that can be ticked, discovered by method_exists the same way recover() already probes for reset()/reconnect() — so no change to the Adapter interface, and no breaking bump for external adapters.

Two things in there are load-bearing:

• tick() vs maintain(). tick() reads the socket and needs exclusive access; maintain() reaches only what is idle and is safe while a consume loop is mid-receive. The adapter sweep calls maintain() and never tick(), so it cannot double-bind the socket a receive loop is blocked on.
• Pool::maintain() drains before ticking rather than popping and pushing one at a time. Adapters choose their own order — Stack is LIFO, Swoole's channel FIFO — and on a LIFO adapter the one-at-a-time form hands the same resource straight back and leaves every other one unswept. Caught by the test, which found exactly that.

This closes the finding for any Swoole host. A plain-TCP host still calls tick() itself, which is inherent: nothing in a library can keep an idle socket alive without a clock, and the clock has to come from the host.

P0 — Closed stream resources

TcpTransport::ensureConnected() checked !== null but never is_resource(), and isTimedOut() took the resource by parameter instead of re-reading $this->stream. A close() concurrent with a yielded fread/fgets raised TypeError at TcpTransport.php:181 — and because that is an \Error, not a ConnectionException, it bypassed reconnect entirely.

ensureConnected() now checks is_resource(); isTimedOut() and a new isAtEof() re-read the property.

Scope note: the ticket names TcpTransport, but TlsTransport and WebSocketTransport carried the identical defect — same === null check, same !feof($this->stream) in isConnected(). All three are fixed; fixing one would have left TLS and WebSocket exposed while looking handled. SwooleTransport uses a Swoole Client object and is unaffected.

P1 — Parser fails closed mid-frame

readExactly() passed no timeout, so the payload read inherited whatever deadline the last readLine() left on the stream. A payload split across segments could raise TimeoutException after the header line was consumed — and processMessage() catches TimeoutException first and returns null, indistinguishable from "nothing arrived". The buffer was then sitting mid-frame, so every later read parsed payload bytes as protocol: permanently desynced, and the dropped frame is often the PubAck itself.

The timeout is now passed explicitly on every frame-body read, and the two frame parsers run through a guard that marks the parser poisoned and raises ConnectionException — which reaches the reconnect path, and connectToServer() builds a fresh Parser.

Design note: the timeout applies per read, not to the frame as a whole. A large payload legitimately arrives across several segments, and charging the whole body to one caller deadline would fail frames that are making steady progress — a regression risk the stricter reading would have introduced.

P1 — Transport failures are no longer rewritten as CAS conflicts

KeyValue::create/update and ObjectStore's two guarded meta publishes wrapped js->publish() in catch (\Throwable) and rethrew a semantic verdict, so a stale connection was reported as "you lost the race" — a conclusion the caller acts on by not retrying.

Only a JetStream API rejection carrying err_code 10071 (new JetStream::ERR_WRONG_LAST_SEQUENCE) is treated as a conflict; transport and timeout errors keep their own identity, and other JetStream API errors rethrow as themselves rather than being relabelled.

ObjectStore::writeVersion also called purgeChunks() on that path, deleting the chunks it had just written. The purge is now reached only by a genuine conflict — after an ambiguous transport failure the meta publish may well have landed, so deleting its chunks is the worst available option.

P2 — Fetch inbox subscription leak

Consumer::fetch() called $sub->unsubscribe() outside a finally, leaking the inbox subscription on any throw — and attemptReconnect() re-subscribes every leaked sid, so the leak compounded across a reconnect storm. Moved into a finally, with the collection loop extracted to collectBatch().

The pull request's expires also equalled the client deadline exactly, so a message the server dispatched at the boundary was dropped client-side after the server had already counted the delivery, burning a maxDeliver attempt for nothing. The server expiry is now pulled in ahead of the client deadline.

---

Verification

Check	Result
Unit	117 pass (80 baseline + 37 new)
E2E (nats)	68 pass, unchanged — real NATS, covers KeyValue, ObjectStore, pull consumers
Unit (pools)	+8 (4 tests x 2 adapters); Stack pass locally, Swoole needs the extension
Unit (queue)	79 pass (73 baseline + 6)
E2E (queue)	idle pooled publisher survives repeated maintenance, against real NATS
PHPStan	1 error, pre-existing (Headers.php:98) — verified by stashing and re-running on clean main
Pint	passes
bin/monorepo validate	all packages valid

The new tests were each checked against the unfixed code rather than just passing against the fix:

• reverting the KeyValue catch → 2 failures
• removing the finally → 2 failures, with 5 repeated fetches leaving 5 leaked subscriptions
• all three transports raised TypeError on a closed resource before the guard

Incidental, not fixed here

The reconnect jitter calls lcg_value(), deprecated in PHP 8.4, while the package declares php: >=8.1. Pre-existing; these tests surfaced it. Out of scope — the tests set reconnectJitter: 0.0, which also cut the unit suite from 2.05s to 0.008s. Worth its own ticket.

---

Update — the keepalive finding is now closed

The last commit closes finding 2, which this PR originally left open pending a design decision. The decision was adapter-driven tick: the queue Swoole adapter owns the interval, pools gains a maintain() sweep, and nats gains the PONG collection that makes ticking safe. Rationale and the two rejected alternatives are in the section above.

The e2e for it asserts on reconnect count, not on the publish succeeding. A keepalive that pings without ever reading the replies still ends with a working publish — checkPings() quietly rebuilds the connection once the budget is spent — so "the publish worked" cannot tell the two apart. What separates them is that the broken version gets there by churning through reconnects of a connection that was healthy all along: 4 reconnects before the fix, 0 after.

CLO-4749 ("CDN purge dies on an idle-closed pooled connection") is the same failure class in a different package, and Pool::maintain() is the mechanism that covers it too.

Stacked follow-ups

• #185 — Epic 2 (CLO-4772), based on this branch
• #186 — Epic 3 (CLO-4773), based on feat(queue): give every NATS enqueue a deduplication id (CLO-4772) #185

[levivannoort]

CI note: test (nats) failed twice before going green, for two unrelated reasons.

1. My miss — the new tests were written against none of the rector rules the check step enforces (YieldDataProviderRector, SortCallLikeNamedArgsRector, ThrowWithPreviousExceptionRector). Fixed in c86a925 with bin/monorepo check nats --fix. I had run phpstan and pint by hand but not bin/monorepo check, which is what CI actually runs.

2. Pre-existing flake, not from this branch — RequestManyTest::testStallStopsBeforeTimeout asserts a wall-clock bound of < 1.5s on a stall window of 0.3s inside a 3.0s timeout. It failed here at 2.13s, and the identical failure happened on main in run 33130472771 at 2026-08-28T00:40 (2.03s vs 1.5s), before this branch existed. Green on re-run.

That flake is worth its own ticket — three responder round-trips plus a stall window is not reliably under 1.5s on a loaded runner, and the assertion has only ~1.2s of headroom over the stall it is actually testing. Deliberately not touched here.

Everything green now, including test (queue, linked), which exercises the queue package against these nats changes resolved from the local checkout.

[levivannoort]

@greptile review