perf: parse incoming messages with a push based token parser

[arthurschreiber]

Summary

Incoming message data used to flow through a Message PassThrough into an async generator that awaited the message's async iterator for every chunk and, inside the token readers, whenever a token was incomplete. A single-row response cost ~80 promises and a 3-packet response ~60; delivery lagged the socket by several event loop turns. This replaces that with a synchronous, push based parser.

• IncomingMessage (new) replaces Message for incoming data. It is still a Readable for consumers that want raw bytes (TLS handshake, prelogin, tests), but a MessageSink can be attached that receives each chunk synchronously as IncomingMessageStream frames it.
• TokenStreamParser attaches itself as that sink and parses whatever complete tokens are buffered right away. Pausing a request makes push return false, which stops framing further packets until the request is resumed, so backpressure to the socket is preserved. Data that arrives while the sink is stalled is queued behind data buffered before it attached.
• StreamParser is push based: push(chunk) then parseNext() per token. Readers are synchronous and throw NotEnoughDataError for an incomplete token; the parser rewinds to the last committed position and waits for the byte count the reader asked for before retrying. Rows, NBC rows and return values keep their progress on the parser (RowState, ReturnValueState) and commit after every column and every PLP chunk, so a value spanning many packets is parsed linearly. All other tokens are retried from their start.
• The async column metadata, row, NBC row, return value and PLP readers are removed, as is the unused async metadataParse. Parser.parseTokens remains as an adapter for the token parser tests.

Measurements

Local SQL Server 2025, loopback, Node 22, default 4 KB packets, warm connection (n iterations). Ops/s for master and for this branch, measured in the same session.

benchmark	master	this branch
select 100 rows (int, nvarchar(100), nvarchar(max)), n=1000	871	1077
select 1000 rows, n=300	298	422
select 10000 rows, n=100	43	77
select 1 row, 10 KB varbinary, n=2000	1163	1384
select 1 row, 10000 char nvarchar(max), n=2000	1149	1258
select 1 MB varbinary, n=200	181	199
1 KB varbinary parameter, n=2000	1511	1643
bulk load 1000 rows, n=200	20	20

Async resources per request (async_hooks): a 3-packet response goes from 61 promises to 9, and per-request cost now rises smoothly with response size instead of stepping up at the packet boundary.

Testing

• npm test: 538 passing.
• npm run test-integration against a local SQL Server 2025: all passing except two pre-existing environment specific tests (a connect timeout test whose target address this sandbox rejects immediately, and the TDS 8 strict TLS test).
• npm run lint clean.
• Additionally exercised directly: pause/resume mid response (single and multi packet), pause and resume from within a row handler, cancel while paused, cancel immediately after execSql, error tokens, PLP and NULL (NBC) values, output parameters, prepared statements and multi result set responses.
• New unit tests for IncomingMessage cover sink attach ordering, stalling and end delivery.

🤖 Generated with Claude Code

https://claude.ai/code/session_01DZCoWULmkdZ9hXnVoBAsj5

[claude]

Review

This is a well-engineered change with a clear rationale and solid benchmarking. The core idea — a push()/parseNext() state machine instead of an async generator per token, plus a synchronous single-packet fast path for small requests — is sound, and the elimination of metadataParse, the async column/row/return-value/PLP readers, and the Message PassThrough for incoming data is a nice simplification alongside the perf win. Below are a few observations, nothing I'd consider blocking.

Correctness / robustness

• src/incoming-message.ts sink/backpressure state machine: I traced through write/end/attach/flushToSink/continueSink against the new tests plus the IncomingMessageStream call site, including the "buffered before attach, then stalls, then resumes" case and the _read/flushing reentrancy guard. It's subtle but looks internally consistent — nice work on test/unit/incoming-message-test.ts covering the tricky orderings.
• stream-parser.ts push() rebasing math (needed -= committed, the three buffer-rebuild branches): checked this against the NotEnoughDataError byte-count semantics (an absolute offset into the pre-trim buffer) and it comes out consistent, including the committed === buffer.length fast path. Worth a second pair of eyes given how load-bearing this is, but I didn't find a case where it under- or over-counts.
• Handler exceptions now surface deeper in the call stack. Previously a token handler that threw would propagate out of a Readable's internal 'data' emission (via Readable.from); now handler[...](token) is called synchronously from TokenStreamParser.drain(), which is reached directly from IncomingMessage.write() → IncomingMessageStream.processBufferedData() → the socket's 'data'/_transform path. Functionally this was always somewhat fragile, but the exact place an uncaught handler exception surfaces has moved. Worth confirming there's a test (or at least manual verification) for a handler throwing mid-stream, since that failure mode is now closer to the socket-read code.
• RpcRequestPayload.buildSync(): for a request with parameters before a streamed (AsyncIterable) one, writeTypeInfo/writeValue run once in the discarded sync attempt and then run again for real via the async iterator fallback. That's safe as long as those functions are pure w.r.t. parameter.data (true for all synchronous data types today, and streamed types return their generator without executing any body until iterated, so nothing is double-consumed) — but it's a non-obvious invariant for future data-type authors to preserve. Might be worth a short comment on writeValue's contract (or in data-type.ts) noting that it must not have observable side effects before being iterated/awaited.
• writeSinglePacketMessage writes straight to the socket/TLS cleartext stream and ignores .write()'s boolean return value, unlike OutgoingMessageStream which respects backpressure via push()/pause(). Given it's bounded to a single packet's worth of data this is unlikely to matter in practice (request/response is inherently serialized here), but it's a slightly different backpressure contract than the rest of the write path — worth a one-line comment if intentional.

Test coverage

• The new IncomingMessage unit tests are thorough. The existing row/NBC-row "one byte at a time" tests continue to exercise the resumable parsing through the Parser.parseTokens adapter, which is good — the resumability logic isn't going untested.
• I didn't see a dedicated unit test for the new fast path itself (Connection.buildPayloadSync / MessageIO.writeSinglePacketMessage), or for the specific behavior change called out in the PR description — canceling immediately after a single-packet request now always takes the attention path rather than the IGNORE path. The existing cancel test was adjusted to avoid hitting the new fast path (by forcing a multi-packet request) rather than adding a case that exercises the new "cancel after a fast-written request" behavior. Since this is a real, user-visible behavior change, it'd be good to have an automated regression test for it rather than relying on the manual testing mentioned in the PR description.

Nits

• src/value-parser.ts has a stray extra blank line introduced right before the final export block — purely cosmetic.

Overall

Solid, carefully-reasoned perf work with good measurements and a design that keeps the "retry from scratch" vs. "resumable with committed state" tradeoff explicit per token type. My comments above are mostly "please double check" rather than "this is broken" — I did not find a concrete correctness bug in the row/NBC-row/PLP/return-value resumption logic or in the new backpressure plumbing.

---

This review was generated by Claude Code (Sonnet 5) via automated PR review.

[claude]

Review

This is a well-structured perf change — the IncomingMessage/MessageSink push model and the NotEnoughDataError/rewind-and-retry parser design are a clean way to eliminate the async-generator overhead, and the commit() discipline in row-token-parser.ts, nbcrow-token-parser.ts, returnvalue-token-parser.ts, and readPLPStream (value-parser.ts) is exactly right for making large multi-packet values resumable without re-parsing.

Performance: colMetadataParser never commits progress

src/token/colmetadata-token-parser.ts:83-95 iterates all columns in a single call without ever calling parser.commit(). The docstring (line 80-81) states this is intentional ("Not resumable: if the buffered data runs out, parsing is retried from the start of the token"), so this isn't a correctness bug, but given the PR's stated goal, it's a real regression versus the pre-refactor code, which advanced parser.position after each column and only retried the one that failed.

For a wide result set (hundreds of columns, more with per-column Always Encrypted crypto metadata) whose COLMETADATA token spans multiple TDS packets, every new packet causes the entire column list to be re-parsed from column 0 before failing again on the next missing byte — quadratic cost in column count × packet count instead of linear. Since readColumn is already structured as a per-column step (metadata → table name → column name), it looks straightforward to commit after each iteration the same way row-token-parser.ts does per column, unless there's a reason this token was deliberately left out (e.g. it's assumed to almost always fit in one packet in practice — worth confirming that assumption holds for the AE/wide-schema case).

PR description vs. diff mismatch

The PR description opens with a whole section on "Write single-packet requests directly to the socket" (bullet 1, with its own behavior change around cancel() and IGNORE vs. attention path) and the title mentions "direct single-packet request writes," but the actual diff only touches the push-based token parser (IncomingMessage, IncomingMessageStream, StreamParser, the token parsers, value-parser.ts). There's no change to message.ts/request writing/message-io.ts beyond a 3-2 line tweak. Worth double-checking the PR body/title reflect what's actually in this diff — as written it'll confuse reviewers looking for the socket-write change, and the measurements table mixes numbers from both.

Test coverage: incoming-message-stream-test.ts pause/resume test no longer tests what it says

IncomingMessageStream.pause()/.resume() overrides that used to propagate to this.currentMessage were removed (correctly — messages are now sinks, not always paused/resumed this way), but the existing test "correctly handles the last packet coming in after the stream was paused" (test/unit/incoming-message-stream-test.ts:110-176) still calls incoming.pause()/.resume() and consumes via message.pipe(result), never attaching a MessageSink. Since no production code calls pause()/resume() on IncomingMessageStream anymore (message-io.ts consumes it via async iteration), this test now passes due to process.nextTick ordering of push(null)/the write callback, independent of the pause/resume calls in its own name. It's not actively wrong, but it's giving false confidence about backpressure-during-pause coverage — consider either removing it or rewriting it against the MessageSink/attach() path (which the new test/unit/incoming-message-test.ts does cover well for sink stalling/end delivery).

Nits

• src/incoming-message.ts is a solid piece of state machine code (write/end/attach/flushToSink/continueSink) — the flushing re-entrancy guard in _read() is a nice touch that's easy to miss the need for.
• No security concerns spotted — this is a parsing/plumbing refactor, not a change to how untrusted server data is validated (the underlying readXxx helpers and bounds-checking are untouched).

Nice work on the async_hooks measurements in the description — that's a good way to make the "fewer promises per response" claim concrete.

[claude]

Review

I traced through the buffer/backpressure state machine in detail (stream-parser.ts's commit/rewind/needed bookkeeping, IncomingMessage's sink attach/stall/drain protocol, and the resumable RowState/ReturnValueState/PLP readers) rather than just reading the diff at face value, since this is exactly the kind of change where a subtle off-by-one in position bookkeeping would only show up under specific packet-boundary conditions. Overall this is a well-designed and carefully implemented change.

What I verified as correct

• Parser.push()/needed rebasing (stream-parser.ts): needed is only ever set from NotEnoughDataError.byteCount, which is consistently an absolute required buffer length across every call site in helpers.ts, value-parser.ts, and the individual token parsers (verified via grep -rn "new NotEnoughDataError("). Since push() rebases needed -= committed in lockstep with how the buffer itself is rebased (buffer.subarray(committed)), the "needed" fast-path stays correct across multiple push() calls, including when a token's committed position is itself advancing mid-token (rows/NBC rows/return values/PLP chunks). I couldn't construct a case where this drifts.
• End-of-message vs. mid-token detection: isEmpty() (pendingType === undefined && committed === buffer.length) is checked in TokenStreamParser.finish(), so even if a needed threshold were ever wrong, a message ending before a token completes is still caught as 'unexpected end of message' rather than hanging — there's no path to an infinite wait.
• NBC row bitmap bit ordering: the new bitmap[index >> 3] & (1 << (index & 7)) against a Buffer matches the old bit-by-bit array construction (byte & 0b1, 0b10, …) exactly.
• IncomingMessage sink protocol: the fast path (sink.push() called directly, bypassing the Readable's own buffer) vs. the fallback path (buffered via this.push()/this.read() while stalled or before a sink attaches) is internally consistent for every ordering I traced: attach-before-end, end-before-attach, stall-on-last-packet, and pause/resume via continueSink(). The flushing flag correctly prevents _read()'s drain notification from racing with flushToSink()'s own buffered-data delivery.
• Connection-level pause/resume/cancel wiring in connection.ts is untouched by this PR (only the Message → IncomingMessage type changed there) and its public-facing pause()/resume() API on the token parser is preserved, so that integration is low-risk.

Suggestion: add unit tests for multi-packet resumable state

This is the one gap I'd actually want closed before/after merging. The whole point of this PR is that rows, NBC rows, return values, and PLP (varchar(max)/varbinary(max)) values now carry state across push() calls instead of being re-parsed from an async iterator. But today:

• row-token-parser-test.ts does have a "parsing a row delivered one byte at a time" test, which is good coverage for RowState.
• nbcrow-token-parser-test.ts has only a single "many columns" test — no test where the bitmap or a column value is split across two push()/chunk calls.
• There's no returnvalue-token-parser-test.ts at all, so ReturnValueState's resumability (e.g. an output parameter's PLP value split across packets) has no unit coverage.
• No unit test drives a PLP value (readPLPStream/readPLPValue) across a chunk boundary — e.g. the 8-byte length header split from its first chunk, or a chunk's 4-byte length prefix split from its body.

The PR description says these were exercised manually against a real SQL Server, which is good, but that's not repeatable in CI. Given this is precisely the new logic (vs. the mechanical async→sync conversion elsewhere), a regression here (e.g. a PLP chunk boundary miscalculation) could silently corrupt large nvarchar(max)/varbinary(max) values or output parameters without any test catching it. I'd suggest adding a handful of "byte at a time" / "split at the PLP header/chunk boundary" tests analogous to the existing row test, for NBC rows, return values, and PLP specifically.

Minor / non-blocking observations

• IncomingMessage.end() never calls this.push(null) when a sink is attached, so the underlying Readable never formally reaches its own 'end' state in that mode (only sink.end() fires). Functionally fine since nothing consumes it as a stream when a sink is attached, but worth a one-line comment noting it's intentional, since it's a bit surprising for a class that otherwise presents as a normal Readable.
• colMetadataParser (and the other "simple" readers) are explicitly non-resumable and re-parse from the token's start on every retry, same as before. For most responses this is irrelevant, but a very wide COLMETADATA token that happens to straddle several packet boundaries would re-walk all previously-parsed columns each time (pre-existing characteristic of the old async-retry-loop model too, not a regression — just flagging it's still O(n²)-ish in the worst case).
• IncomingMessage.resetConnection is now always false and appears vestigial for incoming messages (it's a meaningful flag only for outgoing Messages); could be dropped, but it's low value to touch given a test already asserts on it.

Performance & security

The benchmark numbers in the description (removing dozens of promises per response, and the async_hooks counts) are consistent with the structural change (synchronous push-based parsing replacing an async generator awaited per chunk/per incomplete token), and match what I'd expect from this refactor. I don't see any new security-relevant surface here — this is purely internal parsing of already-trusted-transport (TDS/TLS) data, with no new external input trust boundary, and bounds-checking is still centralized in the same Result/NotEnoughDataError helpers as before.

Nice work — this is a substantial, well-reasoned rewrite of a hot path, and the design choices (committing progress per-column/per-PLP-chunk rather than per-token, bypassing the Readable buffer on the fast path) are sound.