node/test/parallel/test-stream-iter-transform-buffering.js

// Flags: --experimental-stream-iter --expose-gc
'use strict';

const common = require('../common');
const assert = require('assert');
const { brotliDecompressSync, gunzipSync, gzipSync } = require('zlib');
const { compressBrotli, compressGzip, decompressGzip } = require('zlib/iter');
const { bytes, from, pull } = require('stream/iter');

async function testDecompressionOutputIsBounded() {
  let input = Buffer.alloc(32 * 1024 * 1024, 0x61);
  const compressed = gzipSync(input);
  input = null;
  for (let i = 0; i < 3; i++) globalThis.gc();

  const before = process.memoryUsage().arrayBuffers;
  const controller = new AbortController();
  const iterator = decompressGzip().transform(from(compressed), {
    signal: controller.signal,
  })[Symbol.asyncIterator]();
  const first = await iterator.next();
  for (let i = 0; i < 3; i++) globalThis.gc();
  const retained = process.memoryUsage().arrayBuffers - before;

  assert.strictEqual(first.done, false);
  assert.ok(retained < 8 * 1024 * 1024,
            `decompression retained ${retained} bytes before backpressure`);
  await iterator.return();
}

// An output buffer smaller than the batch high water mark fills repeatedly
// before a batch is ready, so the engine is re-entered from the write
// callback without yielding.
async function testSmallChunkSizeDecompression() {
  const input = Buffer.alloc(256 * 1024, 0x61);
  const output = await bytes(pull(from(gzipSync(input)),
                                  decompressGzip({ chunkSize: 1024 })));
  assert.deepStrictEqual(Buffer.from(output), input);
}

// Deterministic incompressible data. Brotli's buffering decisions depend on
// content, so random data would make the tests below flaky.
function incompressible(size) {
  const buffer = Buffer.allocUnsafe(size);
  let x = 0x9e3779b9;
  for (let i = 0; i < size; i++) {
    x ^= x << 13;
    x ^= x >>> 17;
    x ^= x << 5;
    buffer[i] = x & 0xff;
  }
  return buffer;
}

// Brotli buffers 1MB of incompressible input internally, so finalizing the
// stream produces more than one batch of output that must be yielded while
// the finish operation is still in progress.
async function testLargeFinishOutput() {
  const input = incompressible(1024 * 1024);
  const output = await bytes(pull(from(input), compressBrotli()));
  assert.deepStrictEqual(brotliDecompressSync(output), input);
}

// Unwrapping the transform function drops the validated fast path, so pull()
// delivers an explicit null flush signal to the compression transform.
async function testExplicitFlushSignal() {
  const input = incompressible(1024 * 1024);
  const { transform } = compressBrotli();
  const output = await bytes(pull(from(input), { transform }));
  assert.deepStrictEqual(brotliDecompressSync(output), input);
}

// Failures accessing or calling the source iterator's return() method must
// not replace the transform's successful completion.
async function testSourceReturnFailuresAreIgnored() {
  const input = Buffer.from('hello world');
  function makeSource(returnDescriptor) {
    let done = false;
    const iterator = {
      next() {
        if (done) return Promise.resolve({ done: true, value: undefined });
        done = true;
        return Promise.resolve({ done: false, value: [input] });
      },
    };
    Object.defineProperty(iterator, 'return', returnDescriptor);
    return { [Symbol.asyncIterator]() { return iterator; } };
  }

  const sources = [
    makeSource({ get: common.mustCall(() => { throw new Error('getter'); }) }),
    makeSource({
      value: common.mustCall(() => { throw new Error('return'); }),
    }),
  ];
  for (const source of sources) {
    const { signal } = new AbortController();
    const batches = [];
    for await (const batch of compressGzip().transform(source, { signal })) {
      batches.push(...batch);
    }
    assert.deepStrictEqual(gunzipSync(Buffer.concat(batches)), input);
  }
}

(async () => {
  // Run the memory measurement on its own so other tests don't skew it.
  await testDecompressionOutputIsBounded();
  await Promise.all([
    testSmallChunkSizeDecompression(),
    testLargeFinishOutput(),
    testExplicitFlushSignal(),
    testSourceReturnFailuresAreIgnored(),
  ]);
})().then(common.mustCall());