fix: respect output stream backpressure - #122
Conversation
doistbot
left a comment
There was a problem hiding this comment.
This PR makes stdout emission backpressure-aware via a shared buffered line writer, migrating outputIds and cli-core's NDJSON emitters to it while keeping the synchronous formatting helpers intact.
Few things worth tightening:
- Attach error handling around every in-flight write, not just when
write()returnsfalse— a small write can later emitEPIPE(e.g.--ndjson | head -n 0) with no listener, raising an unhandlederrorevent. - In the accounts emitter, generate payloads lazily (e.g. pass an iterable yielding one payload at a time) instead of
accounts.map(toPayload), which materializes the full result set and defeats backpressure for large lists. - Minor: the new
outputNdjsonhelper changes the error contract thatformatNdjsonexplicitly guarantees — worth documenting the difference.
I also included a few optional follow-up notes in the details below.
Optional follow-up notes (4)
src/json.ts:43: Unlike
formatNdjson, which validates every item before any byte is written,outputNdjsonflushes chunks as it goes — so a non-serializable item late in a large stream (e.g. a function/symbol at index 50,000) leaves the earlier chunks already on stdout before the TypeError rejects. That's an inherent tradeoff of streaming, but it's a behavioral change from the all-or-nothing contract the sibling helper documents. Worth a sentence in this JSDoc noting that a mid-stream serialization failure can leave partial output behind.src/commands/update.test.ts:87: This hand-rolled
process.stdout.writespy duplicatescaptureStreamfrom../testing/console.js(or theinstallCapturedStreamwrapper from../test-support/cli-harness.js), which the sibling suites in this same PR (account.test.ts,status.test.ts) already use. Reuse it instead:import { captureStream } from '../testing/console.js'and assignstdoutSpy = captureStream(). Unlike() => true,captureStream'ssilentWritealso invokes the optional write callback on a microtask, so it stays correct if a write path ever passeswrite(chunk, cb).src/stream.test.ts:14: This test only exercises the oversized-single-line flush path (
chunkBytes >= chunkSize) — one line of exactlywritableHighWaterMarkbytes forces an immediate flush. The realistic scenario motivating this PR is many small ID/NDJSON lines whose accumulated bytes cross the buffer, which goes through the separatechunk && chunkBytes + lineBytes > chunkSizebranch and is never covered. If that accumulation branch regresses (e.g., the pre-flush check is dropped or its>comparison breaks), the test still passes while real large result sets lose backpressure. Add a case with enough small items to exceed the high water mark (asserting the first write waits ondrainbefore subsequent writes proceed) so the accumulation path is actually protected.src/json.test.ts:73: Two of these three tests re-test behavior already covered elsewhere: "writes NDJSON to stdout in one block" and "writes nothing for an empty iterable" mirror the writeLines tests in stream.test.ts (and the identical ids.test.ts outputIds cases), and the non-serializable-index test re-exercises
stringifyNdjsonItem, which formatNdjson's "throws with the bad index" test covers with the same[1, undefined, 2]input and regex. Since outputNdjson is a thin wrapper, keep a single wiring test (the happy-path one) and delete the other two.
| const output = chunk | ||
| chunk = '' | ||
| chunkBytes = 0 | ||
| if (!process.stdout.write(output)) { |
There was a problem hiding this comment.
Handle asynchronous write errors for every chunk, not only after
write() returns false. A small write can return true and later emit EPIPE when a downstream pipe closes (for example, --ndjson | head -n 0); this direct write has no error listener and Node raises an unhandled error event. Install temporary error handling around each in-flight write and propagate or explicitly handle the failure.
| // both flags are set. Empty list → no lines (EOF-as-end-of-stream). | ||
| if (view.ndjson && !view.json) { | ||
| for (const entry of accounts) console.log(formatNdjson([toPayload(entry)])) | ||
| await outputNdjson(accounts.map(toPayload)) |
There was a problem hiding this comment.
Keep payload generation lazy here.
accounts.map(toPayload) invokes renderJson and retains every generated payload before writeLines can apply backpressure, so a large account list or large custom payloads can still grow memory with the full result set. Pass a generator/iterable that yields toPayload(entry) one entry at a time.
|
Closing without merging. The backpressure concern is technically valid, but we have no reports of it causing problems in practice. Addressing it makes the output helpers asynchronous and requires broad changes across callers and tests. That cost is not justified for now. We can revisit this if we get a real-world report or start streaming result sets that are large enough to make backpressure material. |
Large IDs-only and NDJSON results can exceed stdout's writable buffer. The existing emitters could keep producing output after
writewould have returnedfalse, which allows memory usage to grow when the downstream consumer is slow.This change:
outputIdsasynchronous and backpressure-awareoutputNdjsonpublic helper