refactor: add MessageIO.writeMessage/readMessage and use them for the PRELOGIN exchange

[arthurschreiber]

This is the first step of the plan from #1656: replacing the Message/IncomingMessageStream/OutgoingMessageStream machinery with two straightforward functions. It adds the new primitives plus a first low-risk consumer, without touching the existing stream classes yet:

• writeMessage(stream, packetSize, type, payload, { debug, resetConnection, cancelSignal, signal }) writes a payload (a sync or async iterable of Buffers) to a stream, wrapped in TDS packets, respecting backpressure via drain. If iterating the payload throws, the message is terminated with a final packet that has the IGNORE flag set (so the server disregards the partial message) and the error is re-thrown.
• readMessage(stream, { debug, signal }) is an async generator that yields the packet contents of the next message on the stream. Bytes arriving after the message's last packet (e.g. the start of the next message) are pushed back onto the stream with unshift, to be consumed by the next read — this is what will later keep the message stream aligned around ATTENTION acknowledgements.

Compared to the draft in #1656, two bugs are fixed: the packet-framing loop now correctly waits for more data when only a partial packet is buffered (the draft could spin forever), and stream errors/closes that occur while the consumer is processing a yielded chunk are re-checked on re-entry instead of being lost (the draft could hang).

The first consumer is the PRELOGIN exchange during connection establishment: sendPreLogin and readPreloginResponse now run directly on the raw socket, and the MessageIO instance is only constructed once the prelogin response has been fully read. This works unchanged for encrypt: 'strict', where the socket is already the TLS cleartext stream at this point.

Both functions come with a unit-test suite covering backpressure, payload errors while draining, stream errors, premature close, trailing-byte unshift, and listener cleanup.

Benchmarks

Setting up per-message stream objects is a significant part of the message handling cost, especially before v8 optimizations kick in. Comparing the new functions against the existing stream classes (messages/second, Node.js v24.18.0, median of 3 runs; benchmarks included in this PR):

n	IncomingMessageStream	readMessage
100	9,889	22,441	+127%
1,000	26,244	58,069	+121%
10,000	76,942	150,728	+96%
100,000	173,808	293,468	+69%

n	OutgoingMessageStream	writeMessage
100	9,353	16,518	+77%
1,000	29,759	48,655	+64%
10,000	91,387	127,079	+39%
100,000	156,812	191,138	+22%

(The connection-establishment change itself is once-per-connection and not performance-relevant; the wins materialize once request/response traffic moves over in follow-up PRs.)

Follow-ups will move the remaining outgoing writes (and delete OutgoingMessageStream), then the incoming reads (and delete IncomingMessageStream/Message).

🤖 Generated with Claude Code

API shape

Both are plain module-level functions (matching the precedent set by connector.ts) rather than statics on MessageIO, and everything beyond the essential inputs — debug (now optional), resetConnection, signal — lives in a trailing options object. They are also re-attached to the CommonJS export, since module.exports = MessageIO would otherwise drop them for require users.

AbortSignal support

For readMessage this is a correctness requirement rather than a convenience: calling .return() on the generator while it is suspended waiting for stream data gets queued behind the pending .next() call, so racing reads against an abort promise and cleaning up via .return() deadlocks whenever the abort fires while the stream is open but quiet — exactly the connect-timeout scenario (server accepts TCP, never responds). Aborting the signal settles the pending read from the inside, letting the generator unwind and clean up its listeners.

For writeMessage, the two ways a message can be interrupted are expressed as two separate signals:

• cancelSignal — graceful cancellation, connection stays usable. Stops consuming the payload, discards buffered data, and terminates the message with the IGNORE flag set. This is a normal protocol outcome: the promise resolves, the TDS stream stays aligned, and the server sends a short response to the ignored message (callers distinguish via cancelSignal.aborted). Pending drain waits are skipped, since nothing further will be written for this message. This replaces the earlier plan of expressing cancellation by throwing a sentinel error from the payload iterable and swallowing it at the call site; in the follow-up async makeRequest, Request.cancel() and the request timeout feed this signal.
• signal — teardown. An abort interrupts drain waits and payload waits and throws the abort reason. This can leave a partially written message on the stream, so it is documented as teardown-only. When both signals are aborted, teardown wins.

readMessage deliberately gets no cancelSignal: on the read side, cancellation requires continuing to read (drain the current response, then the attention acknowledgement) or the stream misaligns — cancel is a consumer-level behavior (discard tokens), not a framing-level one. Only teardown interrupts a read.

The PRELOGIN exchange passes the connect signal down, which collapses readPreloginResponse into a plain for await loop. Test coverage includes: already-aborted signals, abort during a drain wait, abort during a payload wait (no spurious IGNORE terminator), abort on a quiet stream, abort after a partial message, a regression test for the .return() deadlock, and abort-listener cleanup via getEventListeners.

[claude]

Review

Nice piece of work — the two new primitives are well-documented, the write/read framing logic is careful about partial packets and backpressure, and the benchmark numbers make a strong case for the direction. A few things worth a look before this pattern is rolled out to the rest of the outgoing/incoming paths.

Correctness

• writeMessage's close handling looks asymmetric with readMessage's (src/message-io.ts). In readMessage, a close event sets a closed flag and rejects the pending wait with new Error('Premature close'). In writeMessage, close is wired to the same handler as drain:

  stream.on('drain', onDrain);
  stream.on('close', onDrain);
  stream.on('error', onError);

  So if the stream closes (without an error, e.g. a plain .destroy()) while writeMessage is blocked on backpressure, the pending waitForDrain() promise resolves rather than rejects, and the loop resumes writing to an already-closed stream. In practice a subsequent stream.write() on a destroyed stream will likely raise ERR_STREAM_DESTROYED via the error listener that's still attached, so this probably self-corrects — but that's relying on Node's write-after-destroy behavior rather than an explicit, tested contract, and it means a "successful drain" is signaled for something that isn't actually a drain. It'd be more robust (and consistent with readMessage) to reject on close the same way, or at minimum add a test that exercises a clean stream.destroy() (no error) while a write is pending on backpressure — the current test suite only covers destroy(new Error(...)), which goes through onError first and never exercises this branch.

• writeMessage/sendPreLogin aren't wired into the connection's AbortSignal. Every other step of Connection.connect() (connectOnPort, readPreloginResponse, performTlsNegotiation, login, etc.) races its work against signalAborted via withAbortRace/Promise.race, so connectTimeout and close() can interrupt them promptly. sendPreLogin now does a plain await MessageIO.writeMessage(...) with no signal at all:

  try {
    await this.sendPreLogin(socket);
  } catch (err: any) {
    signal.throwIfAborted();
    throw this.wrapSocketError(err);
  }

  For the current PRELOGIN payload this is very low risk (it's tiny and essentially never blocks on drain), but the PR description notes follow-ups will move the remaining outgoing writes onto writeMessage too — at that point a stalled socket mid-write could hang past connectTimeout/close() since there's no way to cancel a writeMessage() call in progress. Worth deciding now whether writeMessage should accept an optional signal, since retrofitting it later means touching every call site again.

Test coverage

• The new unit test suite for writeMessage/readMessage is thorough (backpressure, payload errors mid-drain, stream errors, premature close, partial/misaligned packet delivery, trailing-byte unshift, listener cleanup). Good use of a real Duplex with a controllable highWaterMark to force backpressure deterministically.
• One coverage gap: assertNoDanglingEventListeners only checks error/drain/readable listener counts, but both writeMessage and readMessage also register a close listener. It happens to be correctly removed in both finally blocks today, but the helper wouldn't catch a regression that leaked a close listener. Might as well add close to that assertion for symmetry.
• No test hits the "clean close during drain wait" path discussed above for writeMessage.
• No direct unit test for the updated sendPreLogin(socket)/readPreloginResponse(socket, signal) signatures in connection.ts; presumably relying on the existing connect()-level integration tests, which seems fine given they should exercise this path end-to-end, but worth confirming those tests still pass for both encrypt: 'strict' and non-strict modes given the reordering of this.messageIo construction relative to sendPreLogin/readPreloginResponse.

Minor / style

• writeMessage's payload-iterator detection ((payload as AsyncIterable<Buffer>)[Symbol.asyncIterator]) is a reasonable duck-typing check and matches how the payload is described in the JSDoc; no issue, just flagging it's a slightly unusual pattern worth a one-line comment for future readers unfamiliar with the sync/async iterable split.
• The error-masking case in writeMessage (if writing the IGNORE-flagged termination packet itself throws inside the catch block, that new error propagates instead of the original iteration error) seems intentional/acceptable — a stream failure is arguably more actionable than the original payload error — but might be worth a one-line comment since it's a subtle behavior.

Performance & security

• No security concerns — this is transport framing code operating on already-validated packet headers, and the invalid-length check (length < HEADER_LENGTH) guards against malformed packets from the server, consistent with prior behavior.
• The benchmark numbers in the description are convincing (~2x on reads, ~1.2-1.8x on writes for small n), and avoiding per-message stream object allocation is a sensible win. No performance concerns with this PR itself.

Overall this is a solid, well-tested foundation for the migration. The two correctness notes above are edge cases rather than blockers for this PR's actual scope (the low-traffic PRELOGIN exchange), but probably worth resolving — or at least deciding on purpose — before writeMessage/readMessage become the path for larger, cancellable, backpressure-sensitive traffic in the follow-ups.

[claude]

Review

Nice piece of work — the writeMessage/readMessage design is a clean, well-documented replacement for the stream-class machinery, and the abort-signal handling is genuinely subtle and well thought through (the .return()-deadlocks-behind-pending-.next() issue is a real Node stream gotcha, and the regression test for it is a good catch). The benchmark numbers back up the stated motivation.

Correctness

• readMessage's per-message loop looks solid: the packet-header partial-read case (bl.length < length → break, wait for more data) correctly avoids the "spin forever" bug called out in the PR description, and the buffered post-terminal bytes are pushed back with stream.unshift() for the next reader.
• The "did the stream error while we yielded?" re-check (message-io.ts:472-474) only re-checks error, not closed. In practice this is harmless — if the stream closes without an error event while the consumer is processing a yielded chunk, and more complete packets are already buffered in bl, they'll still be drained on the next inner-loop iteration before the outer loop's if (closed) throw catches it. Worth a one-line comment noting this is intentional (already-buffered data is delivered even if the underlying stream closed in the meantime) so a future reader doesn't "fix" it into an inconsistency.
• writeMessage's error/drain/abort races via Promise.withResolvers() + manual listener bookkeeping are correct as far as I can trace, and the finally blocks consistently strip all listeners on every exit path (verified against the assertNoDanglingEventListeners tests).
• connection.ts: moving this.messageIo construction to after the prelogin response is fully read is a nice simplification, and the PR description's claim that this is safe for encrypt: 'strict' checks out — connectOnPort already returns the TLS cleartext socket in that case (connection.ts:2287). The only theoretical gap is _cancelAfterRequestSent (connection.ts:1810) dereferencing this.messageIo before it exists if cancel() fires during the prelogin window — but that handler is only reachable from a state that's unreachable this early, so it's a non-issue in practice, not a regression.

Style / consistency

• New code follows the existing conventions well (error messages like 'Premature close' match the pre-existing ones, bl/Packet usage mirrors incoming-message-stream.ts/outgoing-message-stream.ts, and the new benchmarks match the require('tedious/lib/...') pattern used elsewhere in benchmarks/).
• The JSDoc on both static methods is thorough and explains the trickier semantics (backpressure, IGNORE-flagged termination, abort-is-teardown-only) rather than restating the signature — good adherence to "explain the why, not the what."

Test coverage

• Excellent breadth: backpressure, payload errors mid-drain, stream errors during write/drain/waiting-for-payload, premature close, non-writable/non-readable streams, already-aborted signals, abort during each of the wait states, trailing-byte unshift, and listener-leak assertions via getEventListeners/assertNoDanglingEventListeners.
• One minor nit: 'reads packets that arrive in chunks that do not align with packet boundaries' (test/unit/message-io-test.ts:1082) schedules one setTimeout per byte with increasing delays (0, 1, 2, ...ms). Under CI load these could in principle be reordered since they're independent timers rather than chained callbacks, which would make the test flaky (delivering bytes out of order rather than the intended "one byte per tick" simulation). Chaining via setImmediate/nested callback rather than n independent setTimeouts would remove that possibility, though this is a pre-existing pattern risk rather than something introduced by significant new logic.
• No new test exercises Connection.sendPreLogin/readPreloginResponse directly (there don't appear to be any pre-existing unit tests for these either), so coverage here still relies on integration tests exercising the full connect() path. Given this is flagged as a low-risk first consumer, that seems like an acceptable tradeoff, but worth confirming CI's integration suite actually exercises the encrypt: 'strict' path since that's the one case explicitly called out as needing to keep working unchanged.

Performance

• The benchmarked wins are substantial and the reasoning (avoiding per-message stream object setup) is sound. No obvious regressions introduced — BufferList usage for reassembly avoids repeated Buffer.concat allocations in the common case.

Security

• Nothing concerning. The IGNORE-flagged terminator on payload-iteration errors is a nice correctness/security touch — it keeps the TDS message boundary aligned with the server instead of leaving a truncated message that could be misinterpreted downstream.

Overall: solid, well-tested infrastructure work with clear docs and a sound migration story. The two nits above (the error-vs-closed re-check comment, and the per-byte setTimeout test pattern) are minor and don't block anything.

[claude]

Review: MessageIO.writeMessage/readMessage + PRELOGIN refactor

Overall this is a clean, well-documented refactor. The JSDoc comments explain the tricky invariants well (backpressure, IGNORE-flagged termination, the .return()/abort deadlock), and the test suite is thorough — backpressure, payload errors while draining, stream errors, premature close, trailing-byte unshift, and listener cleanup are all covered. The connection.ts changes for PRELOGIN look correct: wrapSocketError/signal.throwIfAborted() are applied consistently with the existing pattern used elsewhere in initialiseConnection, and pushing trailing bytes back via stream.unshift() before MessageIO starts piping means no data can be lost between the raw-socket phase and the MessageIO-managed phase.

One correctness concern I'd like to flag for verification, plus a couple of minor nits.

Possible swallowed stream error in writeMessage (src/message-io.ts:398-404)

const onError = (err: Error) => {
  if (drain) {
    const { reject } = drain;
    drain = null;
    reject(err);
  }
};

This only surfaces a stream 'error' if something is currently awaiting drain (i.e. inside waitForDrain()). But stream.write() can return true (no backpressure) even when the write will later fail — e.g. a Writable._write that invokes its callback with an error synchronously still returns true from the outer .write() call, because Node defers onwriteError/errorOrDestroy (and therefore the 'error' emission) to process.nextTick when the callback fires synchronously. In that case waitForDrain() is never called, drain is null when 'error' fires, and the error is silently dropped — writeMessage would resolve normally even though the stream never actually accepted the data.

Unlike writeMessage, readMessage guards against exactly this by keeping a persistent error/closed flag that's checked at the top of the loop and again after every yield. Worth mirroring that pattern here — e.g. set a persistent streamError in onError unconditionally, and check/throw it before/after each writePacket() call, not just when a drain wait happens to be in flight.

The test 'handles errors on the stream during writing' (test/unit/message-io-test.ts, around line 844) looks like it's specifically targeting this "callback errors synchronously, no drain needed" scenario (default highWaterMark, single 3-byte payload that fits in one packet) — as opposed to the highWaterMark: 1 variant a few tests down, which explicitly forces the drain-wait path. Worth double-checking that this test actually exercises/fails without a fix — I wasn't able to execute the suite in this environment to confirm empirically, but if the Writable internals behave as described above, it's worth a second look.

Minor nits

• assertNoDanglingEventListeners (test/unit/message-io-test.ts:699) checks 'error', 'drain', and 'readable' listener counts, but both writeMessage and readMessage also register a 'close' listener that's cleaned up in their finally blocks. Adding stream.listenerCount('close') there would make the "no dangling listeners" assertion complete.
• The 'Premature close' errors thrown in message-io.ts (lines 227, 382, 415, 444) are plain Errors, whereas Node's built-in stream teardown error carries code: 'ERR_STREAM_PREMATURE_CLOSE'. Not a functional issue today since nothing branches on .code, but worth keeping in mind if callers ever want to distinguish this programmatically.

Nice touches worth calling out

• promise.catch(() => {}) on the abort promise to avoid unhandled rejections when nothing is currently racing against it.
• Re-checking error right after yield in readMessage, so a stream error arriving while the consumer is processing a chunk isn't lost.
• Re-attaching writeMessage/readMessage onto module.exports for CommonJS consumers, since module.exports = MessageIO would otherwise drop them.

The benchmark numbers in the PR description are a nice touch and make the perf motivation concrete.

[codecov]

Codecov Report

❌ Patch coverage is 95.89744% with 8 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.38%. Comparing base (650843a) to head (6831568).
⚠️ Report is 1 commits behind head on master.

Files with missing lines	Patch %	Lines
src/message-io.ts	96.70%	2 Missing and 4 partials ⚠️
src/connection.ts	84.61%	2 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #1756      +/-   ##
==========================================
+ Coverage   80.84%   81.38%   +0.53%     
==========================================
  Files          90       90              
  Lines        4887     5065     +178     
  Branches      924      958      +34     
==========================================
+ Hits         3951     4122     +171     
- Misses        640      644       +4     
- Partials      296      299       +3     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[claude]

Review

I read through src/message-io.ts (the new writeMessage/readMessage primitives), the connection.ts wiring for the PRELOGIN exchange, and the new unit tests in test/unit/message-io-test.ts.

Overview

This is a well-scoped, low-risk first step: it introduces writeMessage/readMessage as plain async functions, wires them up for the PRELOGIN exchange only, and leaves the existing MessageIO/Message/IncomingMessageStream/OutgoingMessageStream machinery untouched for everything else. The benchmark numbers are a nice touch and make the motivation concrete. Test coverage for the new functions is extensive (backpressure, drain errors, payload errors, abort/cancel interactions, listener cleanup via getEventListeners).

Potential bug: readMessage can hang if the stream closes (without an error event) while a yielded chunk is being processed

In the packet-framing loop, after yield packet.data() the code only re-checks error, not closed:

yield packet.data();

// Did the stream error while we yielded?
if (error) {
  throw error;
}

if (packet.isLast()) {
  ...
}

onClose only rejects the pending waiting promise if one exists at the moment close fires:

const onClose = () => {
  closed = true;
  if (waiting) {
    const { reject } = waiting;
    waiting = null;
    reject(new Error('Premature close'));
  }
};

If the stream closes without emitting error (e.g. socket.destroy() with no error argument, or a plain FIN close) while the generator is suspended at yield for a non-final packet, waiting is null at that moment, so the close is silently recorded (closed = true) but nothing is rejected. When the consumer calls .next() again, the post-yield check only tests error, so execution falls through, finds no more buffered/available data, and re-enters:

waiting = Promise.withResolvers();
await waiting.promise;

Since the close event has already fired once and won't fire again, this promise never settles — the generator (and its caller) hangs forever. The top-of-loop if (closed) throw ... check never gets a chance to run again because the code never returns to the top of while (true) after this point.

This is the same class of bug the PR description says was fixed for the draft ("stream errors/closes that occur while the consumer is processing a yielded chunk are re-checked on re-entry instead of being lost") — but the fix only covers error, not closed. The existing "throws when the stream is closed before the message was fully read" test doesn't cover this window, since it closes the stream while a waiting promise is already pending, not while the generator is suspended past a yield.

Suggested fix: mirror the error recheck for closed right after the yield:

yield packet.data();

if (error) {
  throw error;
}
if (closed) {
  throw new Error('Premature close');
}

and consider adding a regression test analogous to the existing abort-during-partial-message test, but closing the stream (without an error) instead of aborting a signal, after a non-final packet has been yielded.

This is a real (if narrow) race: for the PRELOGIN consumer today the window is tiny since the loop body is just a Buffer.concat, but this primitive is meant to replace the token-stream reading path too, where consumers will do real work between chunks, widening the window considerably.

Minor observations (not blocking)

• In writeMessage, stream.on('close', onDrain) treats a plain close the same as a successful drain. This looks intentional (avoids hanging on drain if the writable side closes), and appears self-healing in practice — a subsequent stream.write() on the destroyed stream will trigger an async error event that the still-attached onError listener picks up, rejecting the promise instead of hanging. Worth double-checking there isn't a Node version/stream-implementation where a post-destroy write() doesn't reliably re-emit error, but I didn't find a concrete failure case.
• The CommonJS re-attachment (module.exports.writeMessage = writeMessage) is a slightly unusual pattern; the accompanying comment explaining why is helpful and this matches precedent from other parts of the module system in this repo.
• writeMessage's cancellation/abort logic (three racing signals: drain, signal, cancelSignal) is intricate but the tests do a good job covering the interesting interleavings (already-aborted, abort during drain, abort during payload wait, both signals aborted). Worth a comment block summarizing the priority rules (signal wins over cancelSignal) directly above the function, since it's currently only in the JSDoc prose and easy to miss when skimming the implementation.

Performance & security

• Benchmarks show solid wins (up to ~127% for reads, ~77% for writes at small n), and the packet-framing logic (length validation, HEADER_LENGTH checks) mirrors the existing implementation, so no new attack surface from malformed length fields — the length < HEADER_LENGTH check is preserved.
• No new dependencies introduced (bl was already used by incoming-message-stream.ts/outgoing-message-stream.ts).

Overall this is solid, well-tested work — the hang scenario above is the one thing I'd want addressed (or explicitly ruled out) before merging, given it's a primitive intended to become the backbone of all future message I/O.