Module A-6·23 min read

Processing gigabytes of transactional logs without V8 heap saturation — Buffer internals, Transform stream pipelines, and zero-copy ingestion.

JJS

Written by Jatin Jain Saraf · Senior Software Engineer

Module 5 — Native Streams & Off-Heap Buffer Storage

Q: Why should you use `Buffer.allocUnsafe(4096)` instead of `Buffer.alloc(4096)` when continuously allocating small buffers in a hot path?

`allocUnsafe` does not zero-fill the memory and allocates from a shared 8KB pre-allocated slab, making it significantly faster. — `Buffer.allocUnsafe()` saves CPU cycles because it doesn't zero-fill the allocated memory (meaning you must immediately overwrite it before reading to avoid stale data). Furthermore, for sizes under 8KB, it slices memory from a pre-allocated internal slab, heavily reducing system call overhead.

Q: What is the correct way to handle incoming data chunks containing incomplete messages in a custom `Transform` stream's `_transform` method?

Append the incoming chunk to an internal buffer, parse and push any complete messages, and retain the remaining incomplete bytes in the internal buffer for the next call. — Because network chunks arrive at arbitrary boundaries, a parser `Transform` stream must accumulate data in an internal buffer. It slices out and processes fully formed messages, pushing the results downstream, while holding onto the leftover incomplete tail for the next `_transform` iteration.

Q: Why is using `pipeline(readStream, gunzip, parser, dbWriter)` highly recommended over `.pipe()` chaining (e.g., `readStream.pipe(gunzip).pipe(parser).pipe(dbWriter)`) in production code?

If any stream in a `.pipe()` chain errors, sibling streams are not automatically destroyed, leading to resource leaks. `pipeline()` correctly propagates errors and destroys all streams. — The `.pipe()` method does not automatically handle stream errors or teardown. If a stream fails in the middle of a `.pipe()` chain, the source may keep reading or memory/file descriptors can leak. The `pipeline()` utility guarantees that if any stream fails, all connected streams are safely destroyed and cleaned up.

What this module covers: Processing gigabytes of blockchain transaction logs or UPI settlement files without crashing the V8 runtime requires understanding the boundary between V8-managed memory and memory that lives completely outside it. Buffer in Node.js allocates from the OS directly, bypassing V8's heap and garbage collector entirely. Custom Transform streams process that memory in chunks, keeping your heap stable regardless of input size. This module covers the precise memory model for off-heap storage, the correct implementation of Transform streams for production ingestion pipelines, and streaming pipeline composition with pipeline().

The Memory Boundary: V8 Heap vs Off-Heap

In Module 1 you learned that V8 manages its own heap — New Space for short-lived objects, Old Space for long-lived ones. All standard JavaScript objects live here: arrays, plain objects, strings, closures.

Node.js adds a second memory region: off-heap memory — memory allocated directly from the OS via malloc, completely outside V8's control. V8 has no visibility into it, no GC responsibility for it, and cannot trigger GC based on its size.

This distinction is critical for ingestion pipelines:

text

V8 Heap:
  - Managed by V8
  - GC pressure from allocations here
  - Limits: --max-old-space-size (default ~1.5GB)
  - Objects: {}, [], strings, closures

Off-Heap (Node.js Buffer):
  - Allocated via malloc/mmap
  - No GC pressure
  - Limits: available system RAM
  - Objects: raw binary data only

When you process a 500MB blockchain block file with Buffer, V8's heap stays flat. The data never touches the GC. The only heap allocation is a small Buffer JavaScript object (~100 bytes) that wraps the off-heap memory pointer.

The shipping-container analogy: Off-heap Buffers are like a shipping-container yard next to the V8 warehouse — the crane pointing at a container (the JS Buffer object) is small enough to sit on a warehouse shelf, but the container itself never enters the warehouse and therefore never triggers a warehouse-wide inventory audit (GC).

`Buffer` Internals: Three Allocation Methods

`Buffer.alloc(size)`

javascript

const buf = Buffer.alloc(65536);  // 64KB, zero-filled

Allocates size bytes of off-heap memory and zeros it. Safe for security-sensitive operations where you cannot allow previous memory contents to be readable. Slower due to the memset.

Use when: the buffer will be read before being fully written (zero prevents reading stale data), or the data is security-sensitive.

`Buffer.allocUnsafe(size)`

javascript

const buf = Buffer.allocUnsafe(65536);  // 64KB, NOT zero-filled

Allocates size bytes of off-heap memory without zeroing. The memory may contain previous data from other allocations. Faster — no memset.

The pooling threshold is half the pool size (Buffer.poolSize >>> 1) — with the default 8KB pool, that's sizes ≤ 4KB. Those allocations are sliced from a shared 8KB pooled slab that Node pre-allocates. Multiple small Buffer.allocUnsafe calls share the same underlying memory slab, reducing syscall overhead.

Use when: you will immediately write to the buffer before reading, and the data is not security-sensitive. Most ingestion pipelines fall into this category — you write parsed transaction data into the buffer immediately.

`Buffer.allocUnsafeSlow(size)`

javascript

const buf = Buffer.allocUnsafeSlow(4 * 1024 * 1024);  // 4MB, NOT zero-filled

Allocates size bytes of off-heap memory bypassing the 8KB slab pool. Each call requests memory directly from the OS (via malloc). Slower than allocUnsafe for small buffers, but avoids fragmentation for large allocations.

Use when: allocating large buffers (> 4KB) that will live for a long time. Avoids consuming the shared slab, which would be a waste for large chunks.

javascript

// Decision tree:
// - Needs zeroing or security-sensitive?  → Buffer.alloc()
// - Small (≤ 4KB), write-immediately?     → Buffer.allocUnsafe()
// - Large (> 4KB) or long-lived?          → Buffer.allocUnsafeSlow()

The Hard Ceiling: `buffer.constants.MAX_LENGTH`

Every allocation method above is still bounded by a hard maximum for a single Buffer instance:

javascript

import { constants } from 'node:buffer';
console.log(constants.MAX_LENGTH); // 4294967296 (2^32, ~4GB) on current 64-bit Node

On modern 64-bit Node, MAX_LENGTH is 2^32 bytes (~4GB) — very much reachable, not a theoretical ceiling. Older Node versions (pre-V8's larger typed-array support) capped a single Buffer at 2^31 - 1 bytes — just under 2GB — which mattered a great deal for this module's use case. A settlement file at the upper end of the "2–5GB per hour" range described below would have exceeded that older ceiling outright, and can still get uncomfortably close to today's ~4GB ceiling; Buffer.alloc(fileSize) for a file that size isn't just wasteful, it risks ERR_OUT_OF_RANGE on any runtime.

The practical takeaway hasn't changed even though the number did: MAX_LENGTH is a safety net for catching a corrupted or unexpectedly large size value passed into an allocation call, not a budget you should ever approach. If a call to Buffer.alloc() or Buffer.allocUnsafe() in your ingestion path is anywhere near this ceiling, that's a signal the architecture has reverted to loading an entire file into one buffer instead of streaming it — exactly what Transform and pipeline() below exist to avoid.

`Buffer` vs `TypedArray` vs `ArrayBuffer`

javascript

// ArrayBuffer: a fixed-size block of memory. No methods to read/write it directly.
const ab = new ArrayBuffer(1024);

// TypedArray: a typed view into an ArrayBuffer. Lives in V8 heap (the view object),
// but the underlying ArrayBuffer data may be off-heap depending on size.
const u8 = new Uint8Array(ab);       // view into ab
const f64 = new Float64Array(ab);    // another view, same memory

// Buffer: Node.js's subclass of Uint8Array. Always off-heap for large allocations.
// Has convenience methods: readUInt32BE, writeFloatLE, copy, etc.
const buf = Buffer.from(ab);         // wrap existing ArrayBuffer as Buffer

For ingestion pipelines: use Buffer. It has the convenience methods (read/write integers at offsets, copy, compare) needed for binary protocol parsing. TypedArray is better for numeric computation where V8's typed array optimization applies.

Off-Heap, Not Zero-Copy

It's worth being precise about what this module actually delivers, because "off-heap" and "zero-copy" are not the same claim. Every example in this module moves data off the V8 heap — but none of them is zero-copy. The TransactionDecoder below calls chunk.copy(this.#scratch, this.#writeOffset) on every incoming chunk; the buffer pool's processBlock calls chunk.copy(buf, offset). Both are real, userspace memcpy operations. The data is off-heap the entire time, but it is copied — at least once, usually from the kernel's socket or file buffer into a buffer your code controls.

What true zero-copy would require:

Operating on offsets into the original chunk, with no re-buffering. If a complete message is guaranteed to arrive within a single chunk (no fragmentation across the chunk boundary), you can call decodeTransaction(chunk.subarray(start, end)) directly — subarray returns a view, not a copy — and skip copying into a scratch buffer entirely. This is why fragmentation is the deciding factor: the moment a transaction can be split across two chunks (which fragmented TCP delivery guarantees will happen eventually), its bytes need to live somewhere contiguous once both chunks have arrived, and that somewhere doesn't exist until you copy them there. The scratch-buffer pattern used throughout this module is the correct response to that constraint — it isn't zero-copy, but it is off-heap and O(1) per byte.
sendfile(2)-style kernel-to-kernel transfer, for the narrower case of moving bytes from a file descriptor straight to a socket descriptor with no transformation in between. The kernel copies the bytes directly, without ever mapping them into your process's userspace. Node itself doesn't use this for fs.createReadStream().pipe(res) — that path still reads into userspace buffers via fs.read() and writes them out with a normal write() syscall, not a true zero-copy kernel transfer (a long-standing open feature request against Node core confirms this isn't implemented today). Reverse proxies like Nginx serving static files do use real sendfile(2). Either way, it stops applying the instant you need to decode, validate, reconcile, or transform the bytes, because the kernel has no way to run your JavaScript mid-transfer — which rules it out for every ingestion pipeline this module builds.

So: the correct claim for this module is off-heap, not zero-copy. The data never touches V8's GC-managed heap and never triggers GC pressure regardless of file size — that's the guarantee this module delivers. But your code still performs real copies of the bytes; what the scratch-buffer and buffer-pool patterns eliminate is redundant copying (the O(N²) blowup from repeated Buffer.concat), not copying itself.

Transform Streams: The Core of Every Ingestion Pipeline

A Transform stream is both a Readable and a Writable. It receives data chunks in, transforms them, and emits transformed chunks out. It is the correct abstraction for:

Decoding raw binary transaction data into JavaScript objects
Parsing line-delimited JSON from a log file
Decompressing compressed data streams
Normalizing inconsistent data formats

The `_transform` Method

javascript

import { Transform } from 'node:stream';

// Bytes reserved for the fixed-size length prefix that precedes every
// variable-length transaction payload on the wire (see readUInt32BE(0) below).
const TRANSACTION_HEADER_SIZE = 4;

// Scratch buffer capacity — a generous upper bound for header + the largest
// expected transaction payload, sized so the expensive compaction path
// (below) is rare in steady state rather than happening on every chunk.
const SCRATCH_CAPACITY = 1 * 1024 * 1024; // 1MB

class TransactionDecoder extends Transform {
  // Pre-allocated, fixed-capacity scratch buffer, reused across every
  // _transform call — no reallocation on the hot path.
  #scratch = Buffer.allocUnsafeSlow(SCRATCH_CAPACITY);
  #writeOffset = 0;  // end of valid (written) data in #scratch
  #readOffset = 0;   // start of unconsumed data in #scratch

  constructor() {
    super({
      readableObjectMode: true,  // emit objects downstream
      // writableObjectMode: false (default) — receives raw Buffers
    });
  }

  _transform(chunk, encoding, callback) {
    // Compact only when the incoming chunk would overflow remaining capacity —
    // shift unconsumed bytes back to offset 0. This copy is proportional to
    // the (small) unconsumed remainder, and only runs when necessary, not on
    // every chunk.
    if (this.#writeOffset + chunk.length > this.#scratch.length) {
      this.#scratch.copy(this.#scratch, 0, this.#readOffset, this.#writeOffset);
      this.#writeOffset -= this.#readOffset;
      this.#readOffset = 0;

      if (this.#writeOffset + chunk.length > this.#scratch.length) {
        return callback(new Error(
          `Transaction exceeds scratch buffer capacity (${this.#scratch.length} bytes)`
        ));
      }
    }

    // Append the incoming chunk in place — one copy of just the new bytes,
    // no reallocation of everything accumulated so far.
    chunk.copy(this.#scratch, this.#writeOffset);
    this.#writeOffset += chunk.length;

    // Parse as many complete transactions as possible
    while (this.#writeOffset - this.#readOffset >= TRANSACTION_HEADER_SIZE) {
      const txLength = this.#scratch.readUInt32BE(this.#readOffset);   // 4-byte length prefix

      if (this.#writeOffset - this.#readOffset < TRANSACTION_HEADER_SIZE + txLength) break;  // incomplete transaction

      const txStart = this.#readOffset + TRANSACTION_HEADER_SIZE;
      const txBuffer = this.#scratch.subarray(txStart, txStart + txLength);
      const transaction = decodeTransaction(txBuffer);
      this.push(transaction);  // emit decoded transaction downstream

      // Advance the read pointer past this transaction — no buffer mutation
      this.#readOffset = txStart + txLength;
    }

    callback();  // signal: ready for more input
  }

  _flush(callback) {
    // Called when the writable side ends
    // Process any remaining data in the buffer
    const remaining = this.#writeOffset - this.#readOffset;
    if (remaining > 0) {
      // Attempt to parse remaining bytes or emit as partial
      console.warn(`Flush: ${remaining} bytes remaining in buffer`);
    }
    callback();
  }
}

Why not Buffer.concat([this.#buffer, chunk]) on every chunk? That pattern — accumulating with a fresh Buffer.concat call per incoming chunk — looks correct and is easy to reach for, but it's an anti-pattern in a hot ingestion path. Each call allocates an entirely new buffer sized to the combined length and copies both the previously accumulated bytes and the new chunk into it. If a transaction is split across many small chunks (common with fragmented TCP delivery or a slow upstream), the accumulation buffer gets fully reallocated and re-copied on every single chunk — the copy cost grows with however much has accumulated so far, so total work across N chunks for one transaction is O(N²), not O(N). At tens of thousands of chunks per second, that reallocation-and-copy churn shows up directly as GC and memcpy pressure, and it directly contradicts the "off-heap, zero-allocation ingestion" goal this module (and Module 1) argue for. The fixed-capacity scratch buffer above avoids this: appending a chunk is a single O(chunk size) copy into pre-allocated memory, and the O(remaining size) compaction copy only runs when the scratch buffer would actually overflow — which, with SCRATCH_CAPACITY sized appropriately for the protocol's max transaction size, is the rare path, not the common one.

The `_flush` Method

_flush is called when the upstream has ended and there is no more data coming. It is your chance to emit any final output from accumulated state. Always call callback() — forgetting it hangs the pipeline.

javascript

// Example: batch emitter that holds items until batch is full,
// then flushes remaining items at end of stream
class BatchTransform extends Transform {
  #batch = [];
  #batchSize;

  constructor(batchSize = 100) {
    super({ readableObjectMode: true, writableObjectMode: true });
    this.#batchSize = batchSize;
  }

  _transform(item, _, callback) {
    this.#batch.push(item);
    if (this.#batch.length >= this.#batchSize) {
      this.push(this.#batch.splice(0));  // emit full batch
    }
    callback();
  }

  _flush(callback) {
    if (this.#batch.length > 0) {
      this.push(this.#batch.splice(0));  // emit partial final batch
    }
    callback();
  }
}

`pipeline()` vs `.pipe()`: Always Use `pipeline()`

.pipe() has a fatal flaw: it does not handle errors correctly. If a stream in the chain errors, other streams are not automatically destroyed. You get resource leaks.

javascript

// BAD: .pipe() — stream leaks on error
readStream
  .pipe(decompressStream)
  .pipe(parseStream)
  .pipe(writeStream);

// If parseStream throws: readStream and decompressStream keep running.
// Memory grows. File handles stay open.

javascript

// GOOD: pipeline() — proper error handling and cleanup
import { pipeline } from 'node:stream/promises';

await pipeline(
  readStream,
  decompressStream,
  parseStream,
  writeStream
);
// If ANY stream in the chain errors:
// - All other streams are automatically destroyed
// - The Promise rejects with the error
// - Resources are cleaned up

`pipeline()` with Async Generators

The most flexible form of pipeline() uses async generators as transform stages:

javascript

import { pipeline } from 'node:stream/promises';
import { createReadStream } from 'node:fs';
import { createGunzip } from 'node:zlib';

// Process a 2GB compressed blockchain transaction log
await pipeline(
  // Stage 1: Read file as chunks
  createReadStream('transactions-2026-05-17.log.gz'),

  // Stage 2: Decompress
  createGunzip(),

  // Stage 3: Split by newline using async generator
  async function* splitLines(source) {
    let partial = '';
    for await (const chunk of source) {
      const lines = (partial + chunk.toString('utf8')).split('\n');
      partial = lines.pop();  // last line may be incomplete
      for (const line of lines) {
        if (line.trim()) yield line;
      }
    }
    if (partial.trim()) yield partial;  // emit final line
  },

  // Stage 4: Parse each line as JSON
  async function* parseJSON(source) {
    for await (const line of source) {
      try {
        yield JSON.parse(line);
      } catch {
        console.warn(`Skipping malformed line: ${line.slice(0, 100)}`);
      }
    }
  },

  // Stage 5: Write to database
  async function* writeToDb(source) {
    const BATCH_SIZE = 500;
    let batch = [];

    for await (const transaction of source) {
      batch.push(transaction);
      if (batch.length >= BATCH_SIZE) {
        await db.bulkInsert(batch);
        yield batch.length;  // emit count for progress tracking
        batch = [];
      }
    }

    if (batch.length > 0) {
      await db.bulkInsert(batch);
      yield batch.length;
    }
  },

  // Stage 6: Track progress
  async function* trackProgress(source) {
    let total = 0;
    for await (const count of source) {
      total += count;
      if (total % 100_000 === 0) {
        console.log(`Processed ${total.toLocaleString()} transactions`);
      }
    }
    console.log(`Complete: ${total.toLocaleString()} transactions`);
  }
);

// Memory throughout this entire pipeline: bounded by HWM settings
// regardless of the 2GB input file size

Off-Heap Buffer Pooling for Zero-Allocation Hot Paths

When your hot path processes millions of transactions per second, even the overhead of Buffer.allocUnsafe() accumulates. Pre-allocate a pool of reusable buffers:

javascript

// Buffer pool for high-frequency binary processing
class BufferPool {
  #free = [];
  #bufferSize;
  #maxPoolSize;

  constructor({ bufferSize = 65536, maxPoolSize = 100 } = {}) {
    this.#bufferSize = bufferSize;
    this.#maxPoolSize = maxPoolSize;

    // Pre-allocate
    for (let i = 0; i < maxPoolSize; i++) {
      this.#free.push(Buffer.allocUnsafeSlow(bufferSize));
    }
  }

  acquire() {
    return this.#free.pop() ?? Buffer.allocUnsafeSlow(this.#bufferSize);
  }

  release(buf) {
    if (this.#free.length < this.#maxPoolSize) {
      this.#free.push(buf);
    }
    // If pool is full, the buffer is simply dereferenced and GC'd
    // (it's a small JS object; the underlying off-heap memory is freed by finalizer)
  }
}

const blockBufferPool = new BufferPool({ bufferSize: 5 * 1024 * 1024, maxPoolSize: 20 });

// Hot path: zero allocation for block processing buffers
function processBlock(rawBlockStream) {
  const buf = blockBufferPool.acquire();

  return new Promise((resolve, reject) => {
    let offset = 0;
    rawBlockStream.on('data', (chunk) => {
      // Buffer.copy() silently truncates instead of throwing on overflow —
      // without this check, a block at/above the pool's 5MB size would be
      // silently corrupted rather than failing loudly.
      if (offset + chunk.length > buf.length) {
        blockBufferPool.release(buf);
        rawBlockStream.destroy();
        reject(new Error(
          `Block size exceeds pooled buffer capacity (${buf.length} bytes) at offset ${offset}`
        ));
        return;
      }
      chunk.copy(buf, offset);  // copy into pooled buffer — no allocation
      offset += chunk.length;
    });
    rawBlockStream.on('end', () => {
      const result = parseBlock(buf.subarray(0, offset));
      blockBufferPool.release(buf);
      resolve(result);
    });
    rawBlockStream.on('error', (err) => {
      blockBufferPool.release(buf);
      reject(err);
    });
  });
}

The On-Call "Quick Fix" That Caused Nightly OOM Kills

Context: During a settlement-window incident, the reconciliation job was failing to produce output on time. An on-call engineer, under pressure to restore the settlement report before the bank's cutoff, traced it to a parser that was struggling to keep up — and shipped a "quick fix": swap the streaming reader for fs.readFileSync(settlementFilePath), load the entire file into a single in-memory Buffer, and parse it synchronously. It worked. The report shipped before cutoff, and the fix went out without a follow-up ticket.

The symptom: For weeks afterward, the settlement-processing pod was killed by the Linux OOM killer at almost exactly the same time every night — settlement time, when the 3GB file landed. During the day, with no settlement file to process, the pod's memory looked completely normal, which made the pattern easy to dismiss as unrelated infrastructure flakiness at first.

The diagnosis: fs.readFileSync on a 3GB file allocates a Buffer sized to the entire file, all at once, then hands back a fully-materialized in-memory copy. On a pod with a 2GB memory limit, this was an instant OOM on any settlement file over ~2GB — and even below that limit, it meant the process's memory footprint tracked the full file size, not a bounded working set, for the entire duration of parsing. The synchronous read also blocked the event loop for the full read duration, stalling health checks and every other request the pod was serving.

The fix: migrating to the pipeline()-based streaming approach shown below — createReadStream with a bounded highWaterMark, piped through SettlementParser, reconcile, and serialize stages. Memory usage during processing dropped from tracking the file size (2–5GB) to a flat, bounded footprint under 200MB, regardless of how large the settlement file grew. The lesson that stuck with the team: a synchronous, whole-file read that "just works" in a quick fix is exactly the failure mode this module's streaming patterns exist to prevent, and it will keep working right up until the file crosses whatever memory limit the container happens to have.

Processing Multi-Gigabyte Settlement Files: A Complete Example

UPI settlement files are batch files containing all payment transactions for a settlement period. A large payment processor might generate 2–5GB settlement files every hour. Processing these requires streaming — loading the full file into memory would require 2–5GB of heap.

javascript

// Process a 3GB UPI settlement file with bounded memory usage
import { pipeline } from 'node:stream/promises';
import { createReadStream, createWriteStream } from 'node:fs';
import { createGunzip, createGzip } from 'node:zlib';
import { Transform } from 'node:stream';

// Scratch buffer capacity: one read chunk at the 4MB highWaterMark used by
// createReadStream below, plus room for one leftover partial record from the
// previous call. Sized once at construction and reused for the stream's
// entire lifetime — see the note after the class for why this replaces a
// naive Buffer.concat() per chunk.
const SETTLEMENT_SCRATCH_CAPACITY = 4 * 1024 * 1024 + 256;

// Custom Transform: parses the settlement file's fixed-width record format
class SettlementParser extends Transform {
  static RECORD_SIZE = 256;  // each record is exactly 256 bytes

  // Pre-allocated, fixed-capacity scratch buffer — same pattern as
  // TransactionDecoder above, adapted for fixed-width (rather than
  // length-prefixed) records.
  #scratch = Buffer.allocUnsafeSlow(SETTLEMENT_SCRATCH_CAPACITY);
  #writeOffset = 0;  // end of valid (written) data in #scratch
  #readOffset = 0;   // start of unconsumed data in #scratch

  constructor() {
    super({ readableObjectMode: true });
  }

  _transform(chunk, _, callback) {
    // Compact only when the incoming chunk would overflow remaining capacity.
    if (this.#writeOffset + chunk.length > this.#scratch.length) {
      this.#scratch.copy(this.#scratch, 0, this.#readOffset, this.#writeOffset);
      this.#writeOffset -= this.#readOffset;
      this.#readOffset = 0;

      if (this.#writeOffset + chunk.length > this.#scratch.length) {
        return callback(new Error(
          `Settlement chunk exceeds scratch buffer capacity (${this.#scratch.length} bytes)`
        ));
      }
    }

    // Append the incoming chunk in place — one copy of just the new bytes,
    // no reallocation of everything accumulated so far.
    chunk.copy(this.#scratch, this.#writeOffset);
    this.#writeOffset += chunk.length;

    // Parse as many complete fixed-width records as are available
    while (this.#writeOffset - this.#readOffset >= SettlementParser.RECORD_SIZE) {
      const record = this.#scratch.subarray(this.#readOffset, this.#readOffset + SettlementParser.RECORD_SIZE);
      this.push(this.#parseRecord(record));
      this.#readOffset += SettlementParser.RECORD_SIZE;
    }

    callback();
  }

  _flush(callback) {
    // _transform only ever leaves a remainder strictly smaller than
    // RECORD_SIZE (it consumes exactly full RECORD_SIZE multiples), so
    // leftover bytes here can never form a complete record — they're
    // truncated or corrupt trailing data and must not be silently dropped.
    const remaining = this.#writeOffset - this.#readOffset;
    if (remaining > 0) {
      console.warn(`Flush: ${remaining} leftover bytes — truncated or corrupt trailing record discarded`);
    }
    callback();
  }

  #parseRecord(buf) {
    return {
      txId: buf.toString('ascii', 0, 32).trim(),
      amount: buf.readBigInt64BE(32),       // 8 bytes, big-endian
      senderId: buf.toString('ascii', 40, 72).trim(),
      recipientId: buf.toString('ascii', 72, 104).trim(),
      timestamp: buf.readUInt32BE(104),     // Unix seconds
      status: buf.readUInt8(108),           // 1=success, 2=failed, 3=pending
    };
  }
}

Why not Buffer.concat([this.#partial, chunk]) on every chunk here too? The same O(N²) reasoning from the TransactionDecoder example above applies, and arguably matters more here: createReadStream below reads in 4MB chunks, so a naive Buffer.concat would reallocate and copy up to several megabytes on every single _transform call for the full duration of a 2–5GB file — hundreds of megabytes of redundant copying per file, purely from re-copying bytes that were already copied in a previous call. Fixed-width records make the fix simpler than the length-prefixed case: since RECORD_SIZE never changes, the scratch buffer only ever needs to hold one read chunk plus, at most, one record's worth of leftover bytes, so SETTLEMENT_SCRATCH_CAPACITY can be sized deterministically rather than guessed at.

javascript

// Process the settlement file — memory stays bounded
async function processSettlementFile(inputPath, outputPath) {
  let processed = 0, successful = 0, failed = 0;

  await pipeline(
    createReadStream(inputPath, { highWaterMark: 4 * 1024 * 1024 }),  // 4MB read chunks
    createGunzip(),
    new SettlementParser(),

    async function* reconcile(source) {
      for await (const record of source) {
        processed++;
        const enriched = await enrichFromDatabase(record);
        if (enriched.status === 1) successful++;
        else failed++;
        yield enriched;
      }
    },

    async function* serialize(source) {
      for await (const record of source) {
        yield Buffer.from(JSON.stringify(record) + '\n', 'utf8');
      }
    },

    createGzip(),
    createWriteStream(outputPath)
  );

  return { processed, successful, failed };
}

Memory profile during this operation on a 3GB file:

V8 heap: ~40–80MB (the pipeline stream objects, parsed records in flight)
Off-heap: up to 4MB (the read HWM) + zlib decompression buffers
Total process memory: < 200MB regardless of file size

`zlib` Streaming: Decompressing Large Blockchain Archives

javascript

// Stream-decompress a compressed blockchain block archive
import { createGunzip } from 'node:zlib';
import { createReadStream } from 'node:fs';
import { pipeline } from 'node:stream/promises';

// zlib.createGunzip() is already a Transform stream
// It handles all the decompression internals off-heap

await pipeline(
  createReadStream('blocks-archive.gz', {
    highWaterMark: 2 * 1024 * 1024  // 2MB read chunks
  }),
  createGunzip(),   // Transform: compressed → decompressed bytes
  new BlockParser(),  // Transform: bytes → block objects
  new BlockWriter()   // Writable: block objects → database
);

zlib uses off-heap buffers internally. The decompression window (32KB by default) lives outside V8's heap. You can tune it:

javascript

import zlib from 'node:zlib';

const gunzip = zlib.createGunzip({
  chunkSize: 64 * 1024,    // 64KB output chunks (default 16KB)
  level: zlib.constants.Z_DEFAULT_COMPRESSION,
});

Summary

Concept	Key Takeaway
Off-heap memory	Outside V8 heap. No GC pressure. Bounded by system RAM, not `--max-old-space-size`.
`Buffer.alloc()`	Zero-filled. Slow. Use when data is security-sensitive or read before full write.
`Buffer.allocUnsafe()`	No zeroing. Fast. Uses 8KB slab for small allocations. Use for write-first buffers.
`Buffer.allocUnsafeSlow()`	No zeroing. No slab pooling. Use for large or long-lived allocations.
`Transform` stream	Readable + Writable. Process chunks as they arrive. `_transform` for chunks, `_flush` for end.
`_flush`	Always call `callback()`. Handles remaining buffered state at stream end.
`pipeline()`	Correct backpressure + error propagation + cleanup. Always use over `.pipe()`.
Async generator stages	Composable pipeline stages with natural async/await semantics. No custom stream class needed.
Buffer pooling	Pre-allocate reusable off-heap buffers. Zero allocation in the hot path.
HWM on `createReadStream`	`highWaterMark` controls chunk size. 4MB is appropriate for block archive ingestion.
`zlib` off-heap	Decompression buffers are off-heap. Tune `chunkSize` for throughput vs memory trade-off.

Off-heap Buffers and streams handle data throughput. Module 6 handles CPU throughput — how to saturate all cores on a large server using cluster and worker_threads, and how to minimize the IPC overhead that typically defeats the purpose of multi-process scaling.

Next: Module 6 — Core Scaling: Multi-Process Clustering & IPC Latency →

Knowledge Check

Why should you use Buffer.allocUnsafe(4096) instead of Buffer.alloc(4096) when continuously allocating small buffers in a hot path?

What is the correct way to handle incoming data chunks containing incomplete messages in a custom Transform stream's _transform method?

Why is using pipeline(readStream, gunzip, parser, dbWriter) highly recommended over .pipe() chaining (e.g., readStream.pipe(gunzip).pipe(parser).pipe(dbWriter)) in production code?

Test your knowledge with more question sets

Sign in to keep reading

The rest of this module is free — sign in with Google to unlock it and track your progress.

PreviousModule A-5: The HTTP/TCP Subsystem & Ingestion Backpressure Next Module A-7: Core Scaling: Multi-Process Clustering & IPC Latency

Discussion

Join the discussion

Loading comments...

Module 5 — Native Streams & Off-Heap Buffer Storage

The Memory Boundary: V8 Heap vs Off-Heap

Buffer Internals: Three Allocation Methods

Buffer.alloc(size)

Buffer.allocUnsafe(size)

Buffer.allocUnsafeSlow(size)

The Hard Ceiling: buffer.constants.MAX_LENGTH

Buffer vs TypedArray vs ArrayBuffer

Off-Heap, Not Zero-Copy

Transform Streams: The Core of Every Ingestion Pipeline

Sign in to keep reading

Discussion

`Buffer` Internals: Three Allocation Methods

`Buffer.alloc(size)`

`Buffer.allocUnsafe(size)`

`Buffer.allocUnsafeSlow(size)`

The Hard Ceiling: `buffer.constants.MAX_LENGTH`

`Buffer` vs `TypedArray` vs `ArrayBuffer`