Destroy the protocol parser on disconnect when an event loop is attached

[groeneai]

Backport of nats-io#1008 by @ arnaudhe, which fixes nats-io#1007. Both are open
upstream, where the only discussion is how to test it, so I am carrying the patch onto the branch
ClickHouse pins and adding the test upstream lacks.

The parser holds half-decoded bytes belonging to one socket, so it must be discarded before any byte
from the next socket is parsed. In src/conn.c that happens in _freeConn and at the exit of
_readLoop, the blocking-IO reader that every reconnect recreates, so there the parser is per
socket. With an event loop attached, reads arrive instead through natsConnection_ProcessReadEvent,
which creates the parser lazily and never destroys it, and the disconnect funnel _processOpError
leaves nc->ps alone. A drop while the parser is inside MSG_PAYLOAD therefore survives the
reconnect, and the next bytes read on the new socket are appended to the stale pending payload. The
handshake cannot clear it: _processExpectedInfo and _sendConnect read through _readProto, one
raw byte at a time.

Measured on the unpatched tree: after a header announcing 12 payload bytes followed by 4 of them and
a close, the reconnected socket's MSG foo 1 5\r\nhello\r\n reaches the subscriber as
natsConn_processMsg(buf="AAAAMSG foo ", bufLen=12); the leftover hello is not a valid op, so the
connection is then torn down, after the wrong message was handed out. ClickHouse is always on this
path and sees it as a whole-message shortfall in test_storage_nats.

The flag is deferred rather than a destroy at the disconnect because both direct alternatives are
unavailable: natsConnection_ProcessCloseEvent takes (natsSock *socket), so handing it the
connection breaks the adapter ABI; and destroying nc->ps in _processOpError would be a
use-after-free, since ProcessReadEvent unlocks before natsParser_Parse while _processOpError
also runs on the ping timer and write paths. Consuming the flag at the top of the next read event
keeps the destroy on the event-loop thread.

Three dispositions, yours to overturn. The code is upstream's unchanged; both of its comments are
reworded to say "external event loop" instead of "libuv", because the guard is nc->el.attached and
the test below drives the suite's generic loop. The test is a new registered case, not an extension
of test_EventLoop or test_EventLoopRetryOnFailedConnect, because both drive a real nats-server
through _startServer, which never announces 12 payload bytes and then sends 4; the new case brings
its own two-accept _startMockupServer thread and needs no broker and no libuv. And I am not sending
the same cherry-pick onto ClickHouse/v3.13.0, which is diverged and has its own open bump PR; one
cherry-pick onto 137843892fbd on request. The ClickHouse-side pin then advances through
ClickHouse/ClickHouse#119867.

Requested on ClickHouse/ClickHouse#76867. Root cause:
ClickHouse/ClickHouse#116417 (comment)

Validation (click to expand)

clang-22, CMAKE_BUILD_TYPE=Debug, BUILD_TESTING=ON, NATS_BUILD_STREAMING=OFF,
NATS_BUILD_EXAMPLES=OFF, NATS_BUILD_LIB_STATIC=ON, configured out of tree. Both arms are the
committed form of the test.

Arm	src/conn.c + src/natsp.h	testsuite sha256	EventLoopParserResetOnDisconnect
base	reverted, test present	6daa0163cb384612fff6...	FAIL 5 of 5, always at the content check
fix	patched	c0ef8098240390188a08...	PASS 7 of 7

Base arm output:

#07 A message is delivered after the reconnect: PASSED
#08 It is not corrupted by the payload pending when the socket went away: FAILED
#09 No stale message follows and the connection survived: FAILED
*** TEST FAILED ***

Two earlier forms of the same test, before its teardown and comments were cleaned up, measured the
same direction on their own binaries: 13 base failures, 0 base passes, 0 fix failures.

The 114 tests in this suite that need no nats-server binary go 113 of 114 on the base tree, the
single failure being this new case, and 114 of 114 with the fix, so nothing else moves. The
nats-server-dependent tests (EventLoop, EventLoopRetryOnFailedConnect, EventLoopTLS) were
excluded rather than measured, since this machine has no broker binary. The two source files also
build clean in ClickHouse's own configuration, where this branch is consumed.

[groeneai]

Internal second-model review: adjudication log (click to expand)

Pre-publication review by an independent model (engine: codex; 5 findings) plus my own cold read of
the resulting code. Nothing was raised against any of this PR's four hunks. The gate's change map
had to be assembled from the ClickHouse superproject gitlink, which sits at master's pin
3e3a3f10, so it spanned 3e3a3f10..HEAD and included the already merged content of #5 and #6.
All five findings landed there and are attributed by hunk below;
git diff cb4edba96601..HEAD is this PR: src/conn.c, src/natsp.h, test/test.c,
test/list_test.txt.

#	Sev	Finding	Verdict	Evidence / action
1	❌	js_maybeFetchMore publishes outside the subscription locks, so a callback destroying its subscription could free state mid-publish (src/js.c)	DISAGREE	Not this PR: src/js.c is #6's merged content. #6 disclosed this window as adopted from upstream main, and its precondition is a handler destroying its own subscription, which INATSConsumer::onMsg does not do.
2	❌	requestedMsgs is updated after publishing, so with MaxMessages=2 and FetchSize=2 a fast response can trigger an oversized second pull (src/js.c)	DISAGREE	Not this PR, same merged hunk. ClickHouse's jsSubOptions are the jsSubOptions_Init defaults plus Stream, Consumer, ManualAck and optionally Queue, so MaxMessages and MaxBytes are never set.
3	⚠️	Custom inbox prefixes stay broken for synchronous _fetch and for 404/408 status messages (src/js.c)	DISAGREE	Not this PR, and pre-existing to #6, which states it verbatim in its own description. Nothing here touches or depends on inbox-prefix parsing.
4	⚠️	test_JetStream_GH823 does not deterministically force the lock inversion	DISAGREE	Not this PR: that test is #6's. This PR's only test/test.c addition is one hunk at @@ -20443, EventLoopParserResetOnDisconnect, which reddens deterministically with the two src/ hunks reverted.
5	⚠️	The libuv attach/FIFO change has no focused libuv regression test (src/adapters/libuv.h)	DISAGREE	Not this PR: that file is #5's merged content, and #5 answered the ask with a standalone ThreadSanitizer harness (25/25 before, 0 after) plus the reason libuv is not in this repo's test build.
6	💡	The commit message overstated its own evidence: it called the field comment the only divergence from upstream when both comments are reworded, and quoted a base-arm run count belonging to a superseded binary	AGREE, corrected before publication	Both sentences now match the measured committed form and the table above.
7	💡	The validation table carried a mistyped testsuite digest	AGREE, corrected before publication	Re-measured with sha256sum and copied, not retyped.
8	💡	The new test destroys its thread-arg mutex without ordering the asynchronous closed callback that locks it	AGREE, not changed	test_EventLoop's teardown is identical, its _waitForConnClosed barrier is satisfiable by the event-loop read(REMOVE) writing the same field, _close posts the disconnected callback the same way, and _destroyDefaultThreadArgs already sleeps under valgrind for this class. Suite-wide and pre-existing, so I did not touch test.c for it.

Severity: ❌ blocker / ⚠️ major / 💡 nit. DISAGREE verdicts carry recorded evidence and are terminal
per finding.

Findings 1 to 5 are recorded here for completeness, not reopened: they concern code that is already
merged on this branch, they were disclosed by the pull requests that introduced it, and I am not
raising them as new work, since none of them has a reachable observed consequence in ClickHouse.

[groeneai]

Pre-PR validation gate (click to expand)

#	Question	Answer
a	Deterministic repro?	Yes. On the unpatched tree with the new test present, ctest -R '^EventLoopParserResetOnDisconnect$' fails 5 of 5 runs of the committed form, always at the same check, in under 5 seconds, with no broker binary, no libuv and no randomization. Two earlier forms of the test, before its teardown and comments were cleaned up, measured the same direction on their own binaries: 13 base failures, 0 base passes, 0 fix failures.
b	Root cause explained?	Yes. natsParser_Destroy runs per socket only on the blocking-IO path, at the exit of _readLoop, which every reconnect recreates. With an event loop attached, reads arrive through natsConnection_ProcessReadEvent, which creates the parser lazily and never destroys it, and the disconnect funnel _processOpError leaves nc->ps alone. A drop while the parser is inside MSG_PAYLOAD therefore survives the reconnect, and the next bytes read on the new socket are appended to the stale pending payload. The handshake cannot clear it: _processExpectedInfo and _sendConnect read the greeting and the PONG through _readProto, one raw byte at a time.
c	Fix matches root cause?	Yes. It restores the per-socket destroy that the external event-loop path lacks, making it symmetric with _readLoop. Nothing is widened, no test is tagged, no guard is placed at the symptom, and nothing changes in ClickHouse. It is upstream's own patch, nats-io#1008.
d	Test intent preserved / new tests added?	Yes. No existing test is modified. One new registered case is added, and it is the only thing that reddens on the unpatched tree; the other 113 broker-free tests of this suite are untouched and green on both arms.
e	Both directions demonstrated?	Yes, on the committed form. Base arm (the two src/ hunks reverted, test present, testsuite sha256 6daa0163cb384612fff6...) fails 5 of 5; fixed arm (c0ef8098240390188a08...) passes 7 of 7. Distinct sha256s, so neither arm ran a stale binary. Two earlier binary pairs, built before the test's teardown and comments were cleaned up, measured the same directions: 13 base failures, 0 base passes, 0 fix failures.
f	Fix is general across code paths?	Yes. _processOpError is the sole disconnect funnel for the attached-loop path, and all six of its entry points (initial connect, _readLoop, the two stale-connection timer sites, the read event, the write event) reach the one block the flag is set in; natsConnection_Reconnect is covered transitively, since its whole body is natsSock_Shutdown. natsConnection_ProcessReadEvent is the only natsParser_Parse caller on that path. _readLoop already honours the invariant and is unchanged. The close-path _evStopPolling is deliberately not flagged, because _freeConn destroys the parser there and the flag would be dead. STAN wraps the same natsConnection and inherits the fix.
g	Fix generalizes across inputs (params/datatypes/wrappers)?	Yes; the input space here is protocol state rather than data types. The flag is unconditional on the attached-loop disconnect, so it fires whatever state the parser was in (OP_START, mid-header, mid-payload, mid-MSG_END) and whatever the next socket's first op is. The test pins the one state that is observably wrong, MSG_PAYLOAD with a partial message buffer; the others were already harmless and stay so, which is what the 114-test regression arm measures.
h	Backward compatible?	Yes. One bool is added to an internal struct in src/natsp.h, which is not a public header. No public API, no change to any adapter callback signature, no wire or file format, no option and no default. A connection that never disconnects never sees the flag.
i	Invariants and contracts preserved?	Yes. The invariant restored is that parser state belongs to one socket. Both writes happen under natsConn_Lock(nc), with no new lock and no new lock order. The destroy runs only on the event-loop thread, at the top of ProcessReadEvent, so no other thread can be inside natsParser_Parse on the freed parser; that is why the flag is deferred instead of destroying in _processOpError, which is entered from the ping timer and the write path on other threads. The existing lazy create below is byte-identical and rebuilds the parser before the next parse, so nc->ps == NULL is never observed by a parse. A straggler read event for the abandoned socket cannot consume the flag early, because _evStopPolling sets nc->sockCtx.fd = NATS_SOCK_INVALID and ProcessReadEvent returns on that check first. The flag is idempotent and cleared exactly where it is consumed.

Session id: cron:clickhouse-impl-slot-5:20260915-061300