Fix ResultSet.json() race condition on JSONEachRow streams

[Onyx2406]

Summary

Fix ResultSet.json() throwing "Stream has been already consumed" on the first call for JSONEachRow format due to a race condition between two readableEnded checks.

Fixes #575

Root Cause

For streamable JSON formats (JSONEachRow, etc.), json() calls stream() internally:

1. json() checks this._stream.readableEnded → false ✓
2. json() calls this.stream()
3. Race window: Stream receives all data and fires 'end' event → readableEnded becomes true
4. stream() checks this._stream.readableEnded → true ✗
5. Throws: "Stream has been already consumed" — on the first call!

This is triggered by fast/small responses (e.g. SELECT number FROM numbers(1)) where the response arrives fully before json() reaches the stream() call.

Fix (44 insertions, 2 files)

packages/client-node/src/result_set.ts:

• Add _consumed boolean flag to ResultSet class
• Add markAsConsumed() private method that checks _consumed || readableEnded and sets _consumed = true synchronously
• text(), json(), stream() all call markAsConsumed() as their first operation
• Split stream() into public stream() (with check) and private _streamImpl() (without check)
• json() calls _streamImpl() after already calling markAsConsumed(), eliminating the double-check race

This matches the pattern already used by the web client's ResultSet which uses isAlreadyConsumed boolean.

packages/client-node/__tests__/unit/node_result_set.test.ts:

• Add regression test that yields to the event loop (setImmediate) before calling json(), simulating the race condition

Test Plan

• New unit test for the race condition scenario
• CI (full test suite)
• Existing consumption tests (text(), json(), stream() only-once) continue to pass since markAsConsumed() covers all the same cases

[Copilot]

Pull request overview

Fixes a race condition in the Node.js client ResultSet.json() for streamable JSON formats (e.g. JSONEachRow) where the first call could incorrectly throw "Stream has been already consumed" due to separate readableEnded checks.

Changes:

• Add synchronous consumption tracking (_consumed + markAsConsumed()) to ResultSet and apply it to text(), json(), and stream().
• Refactor stream() to delegate to a private _streamImpl() so json() can stream rows without re-checking consumption.
• Add a unit regression test intended to simulate the fast-end/race scenario for JSONEachRow.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.

File	Description
packages/client-node/src/result_set.ts	Introduces a boolean consumption flag and refactors stream()/json() to avoid double-check races on fast-ending streams.
packages/client-node/tests/unit/node_result_set.test.ts	Adds a regression test targeting the reported "already consumed" first-call failure.

Comments suppressed due to low confidence (1)

packages/client-node/src/result_set.ts:136

• json() marks the ResultSet as consumed before verifying that the current format can actually be decoded as JSON. In the non-JSON case (Cannot decode ${this.format} as JSON), this is a behavior change: callers can no longer fall back to text()/stream() even though the underlying stream was never read. Consider checking the format first (and throwing) before calling markAsConsumed(), or only calling markAsConsumed() inside the JSON-capable branches right before consuming the stream.

    this.markAsConsumed()
    return (await getAsText(this._stream)).toString()
  }

  /** See {@link BaseResultSet.json}. */
  async json<T>(): Promise<ResultJSONType<T, Format>> {
    this.markAsConsumed()
    // JSONEachRow, etc.
    if (isStreamableJSONFamily(this.format as DataFormat)) {
      const result: T[] = []
      const stream = this._streamImpl<T>()
      for await (const rows of stream) {
        for (const row of rows) {
          result.push(row.json() as T)
        }
      }
      return result as any
    }
    // JSON, JSONObjectEachRow, etc.
    if (isNotStreamableJSONFamily(this.format as DataFormat)) {
      const text = await getAsText(this._stream)

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[peter-leonov-ch]

Thanks a bunch for looking into this tricky issue and bringing a fresh look at the problem. Let's see if we can avoid making "dark" API changes here while addressing the issue in question.

[Copilot]

Pull request overview

Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Onyx2406]

@peter-leonov-ch can you review this please?