{"id":9727,"date":"2026-05-26T08:30:30","date_gmt":"2026-05-26T13:30:30","guid":{"rendered":"https:\/\/frontendmasters.com\/blog\/?p=9727"},"modified":"2026-05-26T08:30:30","modified_gmt":"2026-05-26T13:30:30","slug":"the-production-playbook-for-node-js-stream-leaks","status":"publish","type":"post","link":"https:\/\/frontendmasters.com\/blog\/the-production-playbook-for-node-js-stream-leaks\/","title":{"rendered":"The Production Playbook for Node.js Stream Leaks"},"content":{"rendered":"\n

This is Part 2 of a two-part series. Part 1<\/a> covered the core mental model of backpressure, why highWaterMark<\/code>isn’t a safety net, the .pipe()<\/code>to pipeline()<\/code> migration, and why async\/await<\/code>doesn’t solve data volume problems. We also fixed the immediate crisis: check the .write()<\/code> return value, await the drain<\/code> event, and use pipeline()<\/code> instead of .pipe()<\/code>. Memory flatlined. The pods stopped dying.<\/p>\n\n\n

\n
\n

Article Series<\/h3>\n <\/header>\n
\n
    \n
  1. \n Your Node.js Streams Aren’t Backpressuring. They’re Silently Eating Your Memory.<\/a>\n <\/li>\n
  2. \n The Production Playbook for Node.js Stream Leaks<\/a>\n <\/li>\n <\/ol>\n <\/div>\n<\/div>\n\n\n\n

    But there are failure modes that survive even correct backpressure handling. They don’t show up in tests because tests run on fast local machines with small datasets and clients that never disconnect. Production is not that polite. Here are the five leak patterns that only surface under real traffic, and the five-rule playbook for catching them before your users do.<\/p>\n\n\n\n

    The Five Ways Your Streams Are Leaking Right Now<\/h2>\n\n\n\n

    1. The client leaves, but your server doesn’t notice<\/h3>\n\n\n\n

    A user kicks off a large CSV export, waits ten seconds, then closes the browser tab. The HTTP response emits close<\/a><\/code>. If you’re using the legacy .pipe()<\/code> method, your upstream database query keeps running. The transform keeps processing rows. The memory keeps climbing. Nobody is listening on the other end, but your server doesn’t know that because .pipe()<\/code> only triggers teardown on the finish<\/code> event, which never fires on a dropped connection.<\/p>\n\n\n

    \/\/ Broken: legacy pipe() leaks when the client drops the connection<\/span>\ndb.cursor()\n  .pipe(csvTransform)\n  .pipe(res);<\/code><\/span>Code language:<\/span> JavaScript<\/span> (<\/span>javascript<\/span>)<\/span><\/small><\/pre>\n\n\n
    import<\/span> { pipeline } from<\/span> \"node:stream\/promises\"<\/span>;\n\n\/\/ Fixed: pipeline() automatically tears down upstream when the client drops<\/span>\ntry<\/span> {\n  await<\/span> pipeline(db.cursor(), csvTransform, res);\n} catch<\/span> (err) {\n  if<\/span> (err.code === \"ERR_STREAM_PREMATURE_CLOSE\"<\/span>) {\n    console<\/span>.log(\"Client disconnected early. Cleanup handled automatically.\"<\/span>);\n  } else<\/span> {\n    throw<\/span> err;\n  }\n}<\/code><\/span>Code language:<\/span> JavaScript<\/span> (<\/span>javascript<\/span>)<\/span><\/small><\/pre>\n\n\n
    \/\/ Defensive: bail before starting expensive work if the client is already gone<\/span>\nif<\/span> (req.destroyed) {\n  return<\/span> res.end();\n}\nawait<\/span> pipeline(db.cursor(), csvTransform, res);<\/code><\/span>Code language:<\/span> JavaScript<\/span> (<\/span>javascript<\/span>)<\/span><\/small><\/pre>\n\n\n
    2. Manual listener teardown is a nightmare<\/h3>\n\n\n\n
    Before async iterators, scanning a stream for a specific value and then stopping was surprisingly difficult. You had to wire up data<\/code> events, and when you found what you needed, you couldn’t just return<\/code>. You had to manually detach listeners and destroy the stream, or else it would keep reading in the background, consuming CPU and memory.<\/p>\n\n\n
    \/\/ Broken: legacy event listeners make early exits dangerous and leaky<\/span>\nfileStream.on(\"data\"<\/span>, (line) => {\n  if<\/span> (line.includes(\"ERROR\"<\/span>)) {\n    console<\/span>.log(\"Found error!\"<\/span>);\n    \/\/ The stream is still reading! We have to manually:<\/span>\n    \/\/ fileStream.removeAllListeners(\"data\");<\/span>\n    \/\/ fileStream.destroy();<\/span>\n  }\n});<\/code><\/span>Code language:<\/span> JavaScript<\/span> (<\/span>javascript<\/span>)<\/span><\/small><\/pre>\n\n\n
    \/\/ Fixed: async iterators automatically destroy the stream on break<\/span>\nfor<\/span> await<\/span> (const<\/span> line of<\/span> fileStream) {\n  if<\/span> (line.includes(\"ERROR\"<\/span>)) {\n    console<\/span>.log(\"Found error!\"<\/span>);\n    break<\/span>; \/\/ fileStream.destroy() is called automatically behind the scenes!<\/span>\n  }\n}<\/code><\/span>Code language:<\/span> JavaScript<\/span> (<\/span>javascript<\/span>)<\/span><\/small><\/pre>\n\n\n
    One caveat: while break<\/code> triggers teardown in modern Node.js, older versions (pre-v14) had bugs where the destroy signal didn’t propagate correctly. If you’re maintaining code that must run on legacy runtimes, wrap your async iterator in a try\/finally<\/code> with an explicit .destroy()<\/code> call as insurance.<\/p>\n\n\n\n
    3. Your timeout kills the response but nothing else<\/h3>\n\n\n\n
    Your HTTP framework has a 30-second timeout. A slow client triggers it. The framework calls res.end()<\/code> and moves on. But res.end()<\/code> is a graceful close \u2014 it signals the end of the writable side without triggering an error. That means pipeline()<\/code> doesn’t see it as a failure, so it doesn’t tear down the upstream chain. The fetch that’s pulling data from a third-party API? Still running. The database cursor? Still open.<\/p>\n\n\n
    \/\/ Broken: timeout only closes the response, upstream keeps running<\/span>\nsetTimeout(()<\/span> =><\/span> res.end(), 30000<\/span>);\nawait<\/span> pipeline(fetchStream, res);<\/code><\/span>Code language:<\/span> JavaScript<\/span> (<\/span>javascript<\/span>)<\/span><\/small><\/pre>\n\n\n
    \/\/ Fixed: timeout aborts the entire pipeline, not just the response<\/span>\nconst<\/span> signal = AbortSignal.timeout(30000<\/span>);\nawait<\/span> pipeline(fetchStream, res, { signal });<\/code><\/span>Code language:<\/span> JavaScript<\/span> (<\/span>javascript<\/span>)<\/span><\/small><\/pre>\n\n\n
    When the 30 seconds expire, the signal fires, and pipeline()<\/code> destroys every stream in the chain \u2014 upstream and downstream. The rejected promise’s error will have the code ABORT_ERR<\/code>, which you can catch and handle distinctly from stream errors.<\/p>\n\n\n\n
    4. You’re tying database teardown to network speed<\/h3>\n\n\n\n
    This is a major architectural flaw that causes connection starvation. Developers often release their database connections when the downstream HTTP response finishes. But tying your database connection lifecycle to the HTTP client’s network speed is incredibly dangerous. If a mobile user on a slow 3G connection takes five minutes to download the export, you are holding a database worker open for five minutes just waiting for the TCP socket to close.<\/p>\n\n\n
    \/\/ Broken: ties database connection to downstream network latency<\/span>\nres.on(\"close\"<\/span>, () => db.releaseConnection());<\/code><\/span>Code language:<\/span> JavaScript<\/span> (<\/span>javascript<\/span>)<\/span><\/small><\/pre>\n\n\n
    The fix: decouple your upstream resources from your downstream delivery. Bind the cleanup logic directly to the database cursor’s own close<\/code> or end<\/code> event. The moment the database has yielded its last row, release the connection back to the pool immediately. Let Node.js buffer the remaining rows into memory or flush them to the slow network socket, but don’t starve your database.<\/p>\n\n\n
    \/\/ Fixed: releases connection the moment the cursor finishes, regardless of network speed<\/span>\nconst<\/span> cursor = db.cursor();\ncursor.on(\"close\"<\/span>, () => db.releaseConnection());\n\nawait<\/span> pipeline(cursor, csvTransform, res);<\/code><\/span>Code language:<\/span> JavaScript<\/span> (<\/span>javascript<\/span>)<\/span><\/small><\/pre>\n\n\n
    import<\/span> { finished } from<\/span> \"node:stream\/promises\"<\/span>;\n\nconst<\/span> cursor = db.cursor();\n\n\/\/ Release connection when cursor is done for ANY reason (success, error, or destroy)<\/span>\nfinished(cursor).then(()<\/span> =><\/span> db.releaseConnection());\n\nawait<\/span> pipeline(cursor, csvTransform, res);<\/code><\/span>Code language:<\/span> JavaScript<\/span> (<\/span>javascript<\/span>)<\/span><\/small><\/pre>\n\n\n
    5. pipeline()<\/code> worked, but the source kept going anyway<\/h3>\n\n\n\n
    This is the one that catches people who did everything right. You used pipeline()<\/code>. A downstream transform throws. pipeline()<\/code> tears down the chain and rejects the promise. But between the moment the error fires and the moment the source receives the destroy signal, the source can push a few more chunks. The destroy()<\/code> call is asynchronous \u2014 it schedules the internal _destroy<\/code> callback on the next tick. In that window, an active database cursor can push chunks that allocate memory, which never gets cleaned up.<\/p>\n\n\n
    \/\/ Broken: relies entirely on pipeline() for source teardown<\/span>\ntry<\/span> {\n  await<\/span> pipeline(source, transform, res);\n} catch<\/span> (err) {\n  log.error(\"Pipeline failed\"<\/span>);\n}<\/code><\/span>Code language:<\/span> JavaScript<\/span> (<\/span>javascript<\/span>)<\/span><\/small><\/pre>\n\n\n
    \/\/ Fixed: explicit fallback teardown<\/span>\ntry<\/span> {\n  await<\/span> pipeline(source, transform, res);\n} catch<\/span> (err) {\n  source.destroy(err);\n  log.error(\"Pipeline failed\"<\/span>);\n}<\/code><\/span>Code language:<\/span> JavaScript<\/span> (<\/span>javascript<\/span>)<\/span><\/small><\/pre>\n\n\n
    The Modern Playbook: How to Stream Without Leaking<\/h2>\n\n\n\n
    Here’s what I do now, after learning most of this the hard way. Five rules. None of them is complicated. All of them would have saved me hours of heap snapshot archaeology.<\/p>\n\n\n\n
    Rule 1: pipeline()<\/code> over .pipe(),<\/code> always<\/h3>\n\n\n\n
    Every stream chain goes through pipeline()<\/a><\/code>. No exceptions. It handles teardown on error, teardown on completion, and backpressure propagation across the entire chain. One import, one function call, one try\/catch<\/code>.<\/p>\n\n\n
    import<\/span> { pipeline } from<\/span> \"node:stream\/promises\"<\/span>;\nimport<\/span> { createReadStream, createWriteStream } from<\/span> \"node:fs\"<\/span>;\nimport<\/span> { createGzip } from<\/span> \"node:zlib\"<\/span>;\n\ntry<\/span> {\n  await<\/span> pipeline(\n    createReadStream(\".\/input.csv\"<\/span>),\n    createGzip(),\n    createWriteStream(\".\/output.csv.gz\"<\/span>)\n  );\n} catch<\/span> (err) {\n  console<\/span>.error(\"Pipeline failed:\"<\/span>, err);\n}<\/code><\/span>Code language:<\/span> JavaScript<\/span> (<\/span>javascript<\/span>)<\/span><\/small><\/pre>\n\n\n
    If the gzip transform fails, the read stream closes, and the write stream closes. If the write stream’s disk fills up, everything upstream stops. If the read stream hits a permission error, everything downstream gets destroyed. You don’t wire any of that yourself.<\/p>\n\n\n\n
    The “Drain Hang” Trap<\/h4>\n\n\n\n
    There is a dangerous edge case you must be aware of when waiting for drain. If you write:<\/p>\n\n\n
    \/\/ Dangerous: can hang forever if the stream errors or closes<\/span>\nif<\/span> (!ok) {\n  await<\/span> once(writable, \"drain\"<\/span>);\n}<\/code><\/span>Code language:<\/span> JavaScript<\/span> (<\/span>javascript<\/span>)<\/span><\/small><\/pre>\n\n\n
    If the writable stream throws an error or closes before<\/em> it drains, the drain<\/code> event will never fire. Your async function will hang in memory forever, suspended in an unresolved await<\/code>.<\/p>\n\n\n\n
    The truly bulletproof, modern fix is to wrap the race in a try\/finally<\/code> block and pass an AbortSignal<\/a><\/code> to explicitly clean up the dangling listener:<\/p>\n\n\n
    import<\/span> { once } from<\/span> \"node:events\"<\/span>;\n\n\/\/ Bulletproof: races drain against errors, cleans up the losing listener<\/span>\nif<\/span> (!ok) {\n  const<\/span> ac = new<\/span> AbortController();\n\n  \/\/ When ac.abort() fires in the finally block, it causes the losing<\/span>\n  \/\/ once() to reject with AbortError. We must catch and discard that<\/span>\n  \/\/ specific error, while re-throwing any real stream errors.<\/span>\n  const<\/span> swallowAbort = (err<\/span>) =><\/span> { if<\/span> (err.name !== \"AbortError\"<\/span>) throw<\/span> err; };\n\n  try<\/span> {\n    await<\/span> Promise<\/span>.race([\n      once(writable, \"drain\"<\/span>, { signal<\/span>: ac.signal })\n        .catch(swallowAbort),\n      once(writable, \"error\"<\/span>, { signal<\/span>: ac.signal })\n        .then((args<\/span>) =><\/span> Promise<\/span>.reject(args[0<\/span>]))\n        .catch(swallowAbort)\n    ]);\n  } finally<\/span> {\n    ac.abort(); \/\/ Destroys the dangling listener from the losing event<\/span>\n  }\n}<\/code><\/span>Code language:<\/span> JavaScript<\/span> (<\/span>javascript<\/span>)<\/span><\/small><\/pre>\n\n\n
    The part that trips people up is swallowAbort<\/code>. Why does the losing once()<\/code> throw at all? Because events.once<\/code> returns a Promise, and a Promise must either resolve or reject \u2014 it can’t silently detach and disappear. When ac.abort()<\/code> fires in the finally<\/code> block, it forces every once()<\/code> still waiting on an event to reject with an AbortError<\/code>. That’s the cleanup mechanism. Without swallowAbort<\/code>, that rejection would propagate as an unhandled error and crash your process. With it, the AbortError<\/code> gets caught and discarded, while any real stream error from the error<\/code> listener still propagates normally.<\/p>\n\n\n\n
    The result: if the downstream socket crashes, your loop immediately throws with the real error. And the losing listener leaves zero garbage behind on the stream’s event emitter.<\/p>\n\n\n\n
    Rule 3: Destroy what you create<\/h3>\n\n\n\n
    If you open a stream, you own its lifecycle. Don’t rely on garbage collection to close file descriptors. It won’t do it fast enough in a server that handles thousands of requests.<\/p>\n\n\n\n
    Use try\/finally<\/code> blocks around async iterators. Pass AbortController<\/a><\/code> signals into pipeline()<\/code> so you can tear down entire chains from the outside. When a client disconnects, when a timeout fires, when your health check fails, you need a way to kill every stream that request touched.<\/p>\n\n\n
    const<\/span> ac = new<\/span> AbortController();\nres.on(\"close\"<\/span>, () => ac.abort());\n\ntry<\/span> {\n  await<\/span> pipeline(source, transform, res, { signal<\/span>: ac.signal });\n} catch<\/span> (err) {\n  if<\/span> (err.name === \"AbortError\"<\/span>) {\n    console<\/span>.log(\"Client disconnected, pipeline aborted cleanly.\"<\/span>);\n  } else<\/span> {\n    throw<\/span> err;\n  }\n}<\/code><\/span>Code language:<\/span> JavaScript<\/span> (<\/span>javascript<\/span>)<\/span><\/small><\/pre>\n\n\n
    Client closes the tab? ac.abort()<\/code> fires. Every stream in the chain gets destroyed. No orphaned cursors, no dangling sockets.<\/p>\n\n\n\n
    For streams you consume via async iterators outside of pipeline()<\/code>, enforce the same discipline:<\/p>\n\n\n
    const<\/span> fileStream = createReadStream(\".\/large-file.log\"<\/span>);\n\ntry<\/span> {\n  for<\/span> await<\/span> (const<\/span> chunk of<\/span> fileStream) {\n    \/\/ process chunk<\/span>\n  }\n} finally<\/span> {\n  \/\/ Belt and suspenders: ensure the stream is destroyed even if<\/span>\n  \/\/ the iterator's internal .return() didn't fire for some reason<\/span>\n  if<\/span> (!fileStream.destroyed) fileStream.destroy();\n}<\/code><\/span>Code language:<\/span> JavaScript<\/span> (<\/span>javascript<\/span>)<\/span><\/small><\/pre>\n\n\n
    Rule 4: Profile before production<\/h3>\n\n\n\n
    Don’t wait for Kubernetes to tell you something is wrong. Catch it locally.<\/p>\n\n\n\n
    Run your service with --max-old-space-size=128<\/a><\/code> during load testing. This artificially constrains the V8 heap so memory leaks crash fast instead of simmering for hours. Take a heap snapshot right before the crash and load it in Chrome DevTools<\/a>.<\/p>\n\n\n\n
    Here is what you are actually looking for in that snapshot. Filter the “Summary” tab for WritableState<\/code>. Sort by “Retained Size”. You will likely see a single instance holding 90% of your entire heap.<\/p>\n\n\n\n
    Expand that WritableState<\/code> object and look for its internal queue. In modern Node, this is the buffered<\/code> array (or a bufferedRequest<\/code> linked list in older versions). Expand that, and you’ll find the exact archaeology you’re looking for: <\/p>\n\n\n\n
    \"Chrome
    Chrome DevTools heap snapshot filtered for WritableState<\/code>. The single instance retaining 92% of the heap is the writable stream’s internal buffer queue \u2014 1,834 unprocessed chunks that the producer pushed while ignoring the false<\/code>return value.<\/figcaption><\/figure>\n\n\n\n
    Every single object in that array is a chunk your producer pushed, and the writable blindly accepted, while returning false<\/code>. That is your smoking gun.<\/p>\n\n\n\n
    You don’t even need a heap snapshot to monitor this in real-time. Node exposes the exact size of this queue via stream.writableLength<\/a><\/code> (or stream._writableState.length<\/code> in older versions). If you are investigating a production incident, add a quick diagnostic log:<\/p>\n\n\n
    setInterval(()<\/span> =><\/span> {\n  console<\/span>.log(`[Diagnostic] writableLength: ${writable.writableLength}<\/span> bytes`<\/span>);\n}, 1000<\/span>);<\/code><\/span>Code language:<\/span> JavaScript<\/span> (<\/span>javascript<\/span>)<\/span><\/small><\/pre>\n\n\n
    If backpressure is working, you’ll see that the number is bound strictly to your highWaterMark<\/code>:<\/p>\n\n\n\n
    [Diagnostic] writableLength: 65536 bytes\n[Diagnostic] writableLength: 65536 bytes\n[Diagnostic] writableLength: 65536 bytes<\/pre>\n\n\n\n
    But if you have a leak, you’ll see the exact moment the producer runs away from the consumer. The queue size simply detaches from reality and climbs until the process dies:<\/p>\n\n\n\n
    [Diagnostic] writableLength: 65536 bytes\n[Diagnostic] writableLength: 1425890 bytes\n[Diagnostic] writableLength: 5892304 bytes\n[Diagnostic] writableLength: 12059382 bytes\n...\n[Diagnostic] writableLength: 3840192300 bytes\n\/\/ FATAL ERROR: Reached heap limit Allocation failed - JavaScript heap out of memory<\/pre>\n\n\n\n
    It’s that simple.<\/p>\n\n\n\n
    If you need to capture a heap snapshot from a running production process without restarting it, Node.js supports the --heapsnapshot-signal<\/a><\/code> flag. Start your service with --heapsnapshot-signal=SIGUSR2<\/code>, and when memory starts climbing, send kill -USR2 <pid><\/code> from the host. Node writes a .heapsnapshot<\/code> file to the working directory without stopping the process. No code changes, no restarts, no attaching a debugger.<\/p>\n\n\n\n
    For a broader view of your whole pipeline, run your service through clinic.js Bubbleprof<\/a>. It traces async operations and visualizes where your event loop is stalling. Backpressure stalls show up as long idle gaps between stream operations. You can see exactly which stream in the chain is the bottleneck.<\/p>\n\n\n\n
    Rule 5: Test your backpressure handling<\/h3>\n\n\n\n
    Don’t rely solely on production profiles. You can write a unit test to catch this. Create a mock writable that explicitly verifies whether the producer waited when the buffer filled up.<\/p>\n\n\n\n
    The strategy: set an absurdly low highWaterMark<\/code> (2 objects), simulate a slow consumer with setTimeout<\/code>, then feed it 100 chunks. If the producer respects backpressure, the internal queue will never grow past highWaterMark + 1<\/code> (the one chunk being actively processed). If it doesn’t, all 100 chunks land in the queue synchronously.<\/p>\n\n\n
    import<\/span> { Writable } from<\/span> \"node:stream\"<\/span>;\nimport<\/span> { pipeline } from<\/span> \"node:stream\/promises\"<\/span>;\n\nit(\"should respect backpressure\"<\/span>, async<\/span> () => {\n  let<\/span> maxBufferLength = 0<\/span>;\n  let<\/span> drainEmitted = false<\/span>;\n\n  const<\/span> slowWritable = new<\/span> Writable({\n    highWaterMark<\/span>: 2<\/span>,\n    objectMode<\/span>: true<\/span>,\n    write(chunk, enc, cb) {\n      \/\/ Track the maximum size the internal queue reached<\/span>\n      if<\/span> (this<\/span>.writableLength > maxBufferLength) {\n        maxBufferLength = this<\/span>.writableLength;\n      }\n      setTimeout(cb, 10<\/span>); \/\/ Simulate a slow consumer<\/span>\n    }\n  });\n\n  slowWritable.on(\"drain\"<\/span>, () => {\n    drainEmitted = true<\/span>;\n  });\n\n  await<\/span> pipeline(yourProducer(100<\/span>), slowWritable);\n\n  \/\/ If the producer ignored backpressure, Node queues all chunks<\/span>\n  \/\/ synchronously, and maxBufferLength will spike to 100.<\/span>\n  expect(maxBufferLength).toBeLessThanOrEqual(3<\/span>);\n  \n  \/\/ Confirm the stream actually paused and resumed<\/span>\n  expect(drainEmitted).toBe(true<\/span>);\n});<\/code><\/span>Code language:<\/span> JavaScript<\/span> (<\/span>javascript<\/span>)<\/span><\/small><\/pre>\n\n\n
    This test is your canary. Run it in CI. If someone refactors your producer and accidentally drops the drain check, this test catches it before the PR merges.<\/p>\n\n\n\n
    Where this is heading<\/h3>\n\n\n\n
    Node.js TSC member Matteo Collina<\/a> has been pushing the ecosystem toward what he calls a “stream-less future.” The idea is to replace the legacy Stream API with pure async generators<\/a> piped through pipeline()<\/code>. He’s advocated this approach across multiple NodeCongress<\/a> and JSNation<\/a> keynotes, and it’s increasingly reflected in how the Node.js core team designs new streaming APIs.<\/p>\n\n\n\n
    Why is this the future? Because it fundamentally flips the control flow. The legacy Stream API is push-based: producers throw data downstream until the downstream yells “stop.” Generators are pull-based: the downstream requests the next chunk, and the upstream calculates it on demand.<\/p>\n\n\n\n
    When you pass a generator to pipeline()<\/code>, Node’s internal machinery handles the translation. You don’t manage events. You don’t wire listeners. You write linear code that yields chunks, and pipeline()<\/code> only pulls the next iteration when the writable side’s buffer has space.<\/p>\n\n\n\n
    Here’s what a fully async generator pipeline looks like in practice, seamlessly blending a database cursor, a transformation, and a downstream HTTP response:<\/p>\n\n\n
    import<\/span> { pipeline } from<\/span> \"node:stream\/promises\"<\/span>;\n\n\/\/ Generator 1: The producer (fetches data on demand)<\/span>\nasync<\/span> function<\/span>* fetchRows<\/span>(dbCursor<\/span>) <\/span>{\n  for<\/span> await<\/span> (const<\/span> batch of<\/span> dbCursor) {\n    for<\/span> (const<\/span> row of<\/span> batch) {\n      yield<\/span> row;\n    }\n  }\n}\n\n\/\/ Generator 2: The transform (processes only when pulled)<\/span>\nasync<\/span> function<\/span>* formatCSV<\/span>(source<\/span>) <\/span>{\n  for<\/span> await<\/span> (const<\/span> row of<\/span> source) {\n    yield<\/span> `${row.id}<\/span>,${row.name}<\/span>,${row.total}<\/span>\\n`<\/span>;\n  }\n}\n\n\/\/ Pipeline handles the backpressure bridging automatically<\/span>\nawait<\/span> pipeline(\n  fetchRows(cursor),\n  formatCSV,\n  res\n);<\/code><\/span>Code language:<\/span> JavaScript<\/span> (<\/span>javascript<\/span>)<\/span><\/small><\/pre>\n\n\n
    No stream classes. No Transform<\/code> instances with confusing object mode settings. No manual drain checks. The memory profile stays entirely flat because the generator yields exactly one chunk at a time, and pipeline()<\/code> only pulls the next iteration when res<\/code> is ready.<\/p>\n\n\n\n
    That last point is the key. With legacy streams, backpressure is something your code must actively cooperate with \u2014 checking the .write()<\/code> return value, awaiting drain<\/code>, wiring up the whole dance. With generators, the backpressure cooperation is structural. Your generator physically cannot produce the next chunk until pipeline()<\/code> asks for it. You don’t check a boolean because there’s no boolean to check. The pull-based model eliminates the entire category of bug.<\/p>\n\n\n\n
    There’s one more problem this solves. You’ve built a great generator pipeline for your CSV export endpoint. Then the reports endpoint needs the same CSV-plus-gzip transformation. Then the batch job needs it too. You copy-paste the pipeline()<\/code> call three times. Someone fixes a bug in one copy but not the others. Now you have three pipelines that should be identical but aren’t.<\/p>\n\n\n\n
    import<\/span> { compose } from<\/span> \"node:stream\"<\/span>;\n\n\/\/ Define the transformation once<\/span>\nconst<\/span> csvExporter = compose(formatCSV, compressGzip);\n\n\/\/ Use it anywhere \u2014 same behavior, single source of truth<\/span>\nawait<\/span> pipeline(fetchRows(cursor), csvExporter, res);<\/code><\/span>Code language:<\/span> JavaScript<\/span> (<\/span>javascript<\/span>)<\/span><\/small><\/pre>\n\n\n
    Where pipeline()<\/code> is for consuming<\/em> a chain, compose()<\/code> is for packaging<\/em> a chain into something other code can consume. The distinction matters when your streaming logic starts getting shared across endpoints.<\/p>\n\n\n\n
    Async generators do introduce a slight CPU and microtask overhead compared to raw streams. This is rarely a bottleneck for IO-bound tasks like CSV exports or network requests, but if you’re writing ultra-hot paths, profile the abstraction cost against your specific workload.<\/p>\n\n\n\n
    The harder question is teardown. This isn’t a generator-specific problem \u2014 it’s the same Rule 3 (Destroy what you create) that applies to every pipeline in this article. The internal bridging between async generators and legacy streams has a fragile history across Node.js versions: patched memory leaks in one release, premature-close regressions in the next. The defense is the same one you’d use for any pipeline()<\/code> call \u2014 pass an AbortSignal<\/code> so the chain tears down cleanly when the downstream socket closes:<\/p>\n\n\n
    const<\/span> ac = new<\/span> AbortController();\nres.on(\"close\"<\/span>, () => ac.abort());\n\nawait<\/span> pipeline(\n  fetchRows(cursor),\n  formatCSV,\n  res,\n  { signal<\/span>: ac.signal }\n);<\/code><\/span>Code language:<\/span> JavaScript<\/span> (<\/span>javascript<\/span>)<\/span><\/small><\/pre>\n\n\n
    This isn’t a weakness of generators. It’s the standard production hygiene you apply to every stream chain, regardless of how it’s built.<\/p>\n\n\n\n
    The Fix is Simple, The System is Not<\/h2>\n\n\n\n
    The CSV export service from Part 1? The immediate fix was four lines. Check the boolean return from .write()<\/code>. If false<\/code>, await the drain<\/code> event. Resume. Memory flatlined at 180MB instead of climbing to 3.8GB. The pods stopped dying. The Slack threads went quiet.<\/p>\n\n\n\n
    But that four-line fix was the beginning, not the end. Every section of this article exists because someone shipped correct-looking code that passed tests, survived code review, and ran fine for months \u2014 until a client disconnected at the wrong moment, a timeout closed the wrong end of a pipeline, or a database connection sat idle for five minutes while a slow socket drained. Node.js streams are built on a trust system, and when you break that trust, nothing fails loudly. The process just quietly accumulates damage until something gives.<\/p>\n\n\n\n
    pipeline()<\/code> over .pipe()<\/code>. Check the boolean. Destroy what you create. Profile with a constrained heap before your orchestrator does it for you. Write the backpressure test that catches the regression before the PR merges. None of it is complicated. The hard part was always knowing it mattered.<\/p>\n\n\n
    \n  
    \n    
    Article Series<\/h3>\n  <\/header>\n  
    \n            
      \n                      
    1. \n              Your Node.js Streams Aren’t Backpressuring. They’re Silently Eating Your Memory.<\/a>\n            <\/li>\n                      
    2. \n              The Production Playbook for Node.js Stream Leaks<\/a>\n            <\/li>\n                  <\/ol>\n        <\/div>\n<\/div>\n","protected":false},"excerpt":{"rendered":"
      Short story: `pipeline()` over `.pipe()` and destroy what you create.<\/p>\n","protected":false},"author":43,"featured_media":9686,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"_acf_changed":false,"sig_custom_text":"","sig_image_type":"featured-image","sig_custom_image":0,"sig_is_disabled":false,"inline_featured_image":false,"_jetpack_memberships_contains_paid_content":false,"footnotes":""},"categories":[1],"tags":[482,442],"class_list":["post-9727","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-blog-post","tag-node-js","tag-streaming"],"acf":[],"jetpack_featured_media_url":"https:\/\/i0.wp.com\/frontendmasters.com\/blog\/wp-content\/uploads\/2026\/05\/streams.jpg?fit=2000%2C1200&ssl=1","jetpack_sharing_enabled":true,"_links":{"self":[{"href":"https:\/\/frontendmasters.com\/blog\/wp-json\/wp\/v2\/posts\/9727","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/frontendmasters.com\/blog\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/frontendmasters.com\/blog\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/frontendmasters.com\/blog\/wp-json\/wp\/v2\/users\/43"}],"replies":[{"embeddable":true,"href":"https:\/\/frontendmasters.com\/blog\/wp-json\/wp\/v2\/comments?post=9727"}],"version-history":[{"count":6,"href":"https:\/\/frontendmasters.com\/blog\/wp-json\/wp\/v2\/posts\/9727\/revisions"}],"predecessor-version":[{"id":9823,"href":"https:\/\/frontendmasters.com\/blog\/wp-json\/wp\/v2\/posts\/9727\/revisions\/9823"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/frontendmasters.com\/blog\/wp-json\/wp\/v2\/media\/9686"}],"wp:attachment":[{"href":"https:\/\/frontendmasters.com\/blog\/wp-json\/wp\/v2\/media?parent=9727"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/frontendmasters.com\/blog\/wp-json\/wp\/v2\/categories?post=9727"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/frontendmasters.com\/blog\/wp-json\/wp\/v2\/tags?post=9727"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}
    stream.compose()<\/a><\/code> (stabilized in Node.js 22) exists for exactly this. It stitches multiple streams or async generators into a single Duplex<\/a><\/code> stream \u2014 a reusable unit that you define once and plug into any pipeline:<\/p>\n\n\n
    The intuitive fix is to race the drain<\/code> event against the error<\/code> event using Promise.race()<\/code>. But doing so introduces a subtle event-loop memory leak: whichever event “loses” the race leaves a dangling listener attached to the stream via events.once<\/a><\/code>. Under heavy load, this will eventually trigger a MaxListenersExceededWarning<\/code> and leak memory.<\/p>\n\n\n\n
    Rule 2: Respect the boolean<\/h3>\n\n\n\n
    Any time you call .write()<\/a><\/code> manually, you check what it returns. If it returns false<\/code>, you stop and wait for the drain<\/a><\/code> event. This is the fundamental pattern underlying everything else in Node.js streams. Memorize it. If you’re writing a custom transform or piping data between two streams by hand, this check goes in every single write path.<\/p>\n\n\n\n
    The fix: in your catch<\/code> block, explicitly call source.destroy(err)<\/code> even though pipeline()<\/code> should have handled it. Belt and suspenders. The redundant destroy call is a no-op<\/a> if the source is already destroyed, and it saves you if it isn’t.<\/p>\n\n\n
    For even more precise lifecycle tracking, stream.finished()<\/a><\/code> lets you listen for when a specific stream is “done” \u2014 whether it completed normally, errored, or was destroyed. It’s the surgical version of binding to individual events:<\/p>\n\n\n
    The fix: on timeout, you need to destroy the stream itself, or use an AbortSignal<\/a><\/code> with a timeout, which kills the entire chain. AbortSignal.timeout()<\/a><\/code> was purpose-built for this:<\/p>\n\n\n
    The fix: async iterators<\/a> (for await...of<\/code>) are actually the safest way to consume streams. Why? Because the JavaScript iterator protocol natively hooks into stream teardown. When you break<\/code>, return<\/code>, or throw<\/code> out of an async iterator loop, JavaScript automatically calls the iterator’s .return()<\/a><\/code> method. Node.js maps this directly to stream.destroy()<\/a><\/code>.<\/p>\n\n\n
    You can also check req.destroyed<\/a><\/code> proactively inside long-running handlers. If the client has already disconnected before you even start the stream, there’s no point opening the database cursor at all:<\/p>\n\n\n
    The fix: modern Node.js (v14+) natively monitors the destination stream if you use pipeline()<\/a><\/code>. If the user closes the tab before the stream formally ends, pipeline()<\/code> detects the premature closure of the socket, throws an ERR_STREAM_PREMATURE_CLOSE<\/a><\/code> error, and automatically destroys the upstream database cursor.<\/p>\n\n\n