undici 6.28.0: Stream response chunks with backpressure, opaque metadata, error propagation, and abort handling using undici.stream in undici 6.28.0

Clean-room public code sample — generation instructions

Write a brand-new, minimal, self-contained code sample in this clean-room directory.
Do not copy, paraphrase, or reference any existing project source. Work only from this spec.

A csx.json manifest scaffold already exists. Do not recreate it from memory. Preserve its case.goal, packages and symbols; fill its empty case.contract with exact assertions and correct its environment, commands and verifierAdapter for the files you generate.

Goal: Stream response chunks with backpressure, opaque metadata, error propagation, and abort handling using undici.stream in undici 6.28.0
Kind: HOW

Use EXACTLY these public packages and versions:
  - pkg:npm/undici@6.28.0
Demonstrate these symbols/APIs:
  - undici.stream

Rules:
  - One focused purpose; the smallest project that proves the goal.
  - Include a contract test (test/contract.*) that runs OFFLINE and exits 0 exactly when the goal behavior works.
  - Pin every dependency with a lockfile so resolution is reproducible.
  - No secrets, credentials, or tokens. No real URLs (only example.com or localhost). No absolute paths.
  - No personal names, emails, company names, or project identifiers of any kind.
  - No binaries and no generated output (node_modules, dist, target, venv, .git, .env).
  - Keep it under 200 files and 256KB packed.

{"case":{"caseId":"case:sha256:c79e2b0bf966fcc16f1da970d7ba92dedf7fd9cb33bf44285b361fa16ed84c50","contract":["undici.stream streams response chunks into a custom Writable stream and returns opaque metadata and response trailers","undici.stream allows inspecting status code and response headers in the factory function before accepting stream chunks","undici.stream aborts in-flight stream consumption when an AbortSignal triggers","undici.stream rejects and propagates errors when the destination Writable stream encounters a write error"],"goal":"Stream response chunks with backpressure, opaque metadata, error propagation, and abort handling using undici.stream in undici 6.28.0","kind":"HOW","packages":["pkg:npm/undici@6.28.0"],"schemaVersion":1,"symbols":["undici.stream"]},"contractCommand":["node","test/contract.js"],"environment":{"arch":"x64","distro":"ubuntu","ecosystem":"npm","language":"javascript","libc":"glibc","libcVersion":"2.39","moduleSystem":"cjs","os":"linux","osVersionBucket":"24","packageManager":"npm","packageManagerVersion":"10.9.8","runtime":"node","runtimeVersion":"22.23.2","schemaVersion":1},"license":"MIT-0","packages":["pkg:npm/undici@6.28.0"],"schemaVersion":1,"subject":"pkg:npm/undici@6.28.0","symbols":["undici.stream"],"verifierAdapter":"node-typescript@1"}

{
  "name": "undici-stream-sample",
  "version": "1.0.0",
  "lockfileVersion": 3,
  "requires": true,
  "packages": {
    "": {
      "name": "undici-stream-sample",
      "version": "1.0.0",
      "dependencies": {
        "undici": "6.28.0"
      }
    },
    "node_modules/undici": {
      "version": "6.28.0",
      "resolved": "https://registry.npmjs.org/undici/-/undici-6.28.0.tgz",
      "integrity": "sha512-LIY910g9TI13YS95lrMFrs8Rm/u/irgHeTWoKCoteeJ04CUJ92eEfj0rVn+7VKMPBpUPiUoBKfhNyLI23EE/KA==",
      "license": "MIT",
      "engines": {
        "node": ">=18.17"
      }
    }
  }
}

{
  "name": "undici-stream-sample",
  "version": "1.0.0",
  "dependencies": {
    "undici": "6.28.0"
  }
}

{
  "schemaVersion": 1,
  "goal": "Stream response chunks with backpressure, opaque metadata, error propagation, and abort handling using undici.stream in undici 6.28.0",
  "kind": "HOW",
  "packages": [
    "pkg:npm/undici@6.28.0"
  ],
  "symbols": [
    "undici.stream"
  ]
}

const { Writable } = require('node:stream');
const undici = require('undici');

/**
 * Streams an HTTP response body to a factory-provided Writable stream.
 *
 * @param {string | URL} url - Target URL.
 * @param {import('undici').StreamOptions} options - Stream request options.
 * @param {import('undici').StreamFactory} factory - Stream factory function receiving { statusCode, headers, opaque }.
 * @returns {Promise<import('undici').StreamData>}
 */
async function streamResponse(url, options, factory) {
  return undici.stream(url, options, factory);
}

/**
 * Collects a streamed HTTP response into a single Buffer using undici.stream.
 *
 * @param {string | URL} url - Target URL.
 * @param {import('undici').StreamOptions} [options] - Stream request options.
 * @returns {Promise<{ statusCode: number, headers: import('undici').IncomingHttpHeaders, body: Buffer, opaque: unknown, trailers: Record<string, string> }>}
 */
async function collectStream(url, options = {}) {
  const chunks = [];
  let responseStatus = 0;
  let responseHeaders = {};

  const writable = new Writable({
    write(chunk, encoding, callback) {
      chunks.push(Buffer.from(chunk));
      callback();
    },
  });

  const result = await undici.stream(
    url,
    options,
    ({ statusCode, headers, opaque }) => {
      responseStatus = statusCode;
      responseHeaders = headers;
      return writable;
    }
  );

  return {
    statusCode: responseStatus,
    headers: responseHeaders,
    body: Buffer.concat(chunks),
    opaque: result.opaque,
    trailers: result.trailers,
  };
}

/**
 * Creates and configures an offline MockAgent for intercepting undici streaming requests.
 *
 * @param {string | URL} origin - Base origin to intercept.
 * @returns {{ agent: import('undici').MockAgent, pool: import('undici').MockPool | import('undici').MockClient }}
 */
function createMockStreamAgent(origin) {
  const agent = new undici.MockAgent();
  agent.disableNetConnect();
  const pool = agent.get(origin);
  return { agent, pool };
}

module.exports = {
  streamResponse,
  collectStream,
  createMockStreamAgent,
  undici,
};

const assert = require('node:assert/strict');
const { Writable } = require('node:stream');
const {
  streamResponse,
  collectStream,
  createMockStreamAgent,
  undici,
} = require('../src/index.js');

async function runContracts() {
  // Contract 1: undici.stream streams response chunks into a custom Writable stream and returns opaque metadata and response trailers
  {
    const { agent, pool } = createMockStreamAgent('http://localhost:8080');

    pool.intercept({
      path: '/data/stream',
      method: 'GET',
    }).reply(200, 'chunk-alpha,chunk-beta', {
      headers: {
        'content-type': 'text/plain',
        'x-stream-id': 'stream-100',
      },
    });

    const receivedChunks = [];
    const writable = new Writable({
      write(chunk, encoding, callback) {
        receivedChunks.push(chunk);
        callback();
      },
    });

    let factoryStatus = 0;
    let factoryHeaders = null;
    const opaquePayload = { traceId: 'tr-999' };

    const streamResult = await streamResponse(
      'http://localhost:8080/data/stream',
      {
        dispatcher: agent,
        opaque: opaquePayload,
      },
      ({ statusCode, headers, opaque }) => {
        factoryStatus = statusCode;
        factoryHeaders = headers;
        assert.deepStrictEqual(opaque, opaquePayload);
        return writable;
      }
    );

    assert.strictEqual(factoryStatus, 200);
    assert.strictEqual(factoryHeaders['content-type'], 'text/plain');
    assert.strictEqual(factoryHeaders['x-stream-id'], 'stream-100');
    assert.deepStrictEqual(streamResult.opaque, opaquePayload);
    assert.ok(typeof streamResult.trailers === 'object');
    assert.strictEqual(Buffer.concat(receivedChunks).toString('utf-8'), 'chunk-alpha,chunk-beta');

    await agent.close();
  }

  // Contract 2: undici.stream allows inspecting status code and response headers in the factory function before accepting stream chunks
  {
    const { agent, pool } = createMockStreamAgent('http://localhost:8080');

    pool.intercept({
      path: '/api/resource/invalid',
      method: 'POST',
      body: JSON.stringify({ action: 'validate' }),
    }).reply(422, JSON.stringify({ error: 'Unprocessable Entity', code: 'INVALID_INPUT' }), {
      headers: { 'content-type': 'application/json' },
    });

    const collected = await collectStream('http://localhost:8080/api/resource/invalid', {
      method: 'POST',
      body: JSON.stringify({ action: 'validate' }),
      dispatcher: agent,
      opaque: { reqKey: 'k-1' },
    });

    assert.strictEqual(collected.statusCode, 422);
    assert.strictEqual(collected.headers['content-type'], 'application/json');
    assert.deepStrictEqual(collected.opaque, { reqKey: 'k-1' });

    const errorBody = JSON.parse(collected.body.toString('utf-8'));
    assert.deepStrictEqual(errorBody, { error: 'Unprocessable Entity', code: 'INVALID_INPUT' });

    await agent.close();
  }

  // Contract 3: undici.stream aborts in-flight stream consumption when an AbortSignal triggers
  {
    const { agent, pool } = createMockStreamAgent('http://localhost:8080');

    pool.intercept({
      path: '/slow-stream',
      method: 'GET',
    }).reply(200, 'slow stream data payload').delay(1000);

    const controller = new AbortController();
    setTimeout(() => {
      controller.abort();
    }, 20);

    const discardWritable = new Writable({
      write(chunk, encoding, callback) {
        callback();
      },
    });

    await assert.rejects(
      async () => {
        await streamResponse(
          'http://localhost:8080/slow-stream',
          {
            dispatcher: agent,
            signal: controller.signal,
          },
          () => discardWritable
        );
      },
      (err) => {
        assert.ok(
          err.name === 'AbortError' ||
          err.name === 'RequestAbortedError' ||
          err.code === 'UND_ERR_ABORTED' ||
          controller.signal.aborted,
          'Error must signal stream abortion'
        );
        return true;
      }
    );

    await agent.close();
  }

  // Contract 4: undici.stream rejects and propagates errors when the destination Writable stream encounters a write error
  {
    const { agent, pool } = createMockStreamAgent('http://localhost:8080');

    pool.intercept({
      path: '/streaming-target',
      method: 'GET',
    }).reply(200, 'stream content causing write failure');

    const failingWritable = new Writable({
      write(chunk, encoding, callback) {
        callback(new Error('Write pipeline failure'));
      },
    });

    await assert.rejects(
      async () => {
        await streamResponse(
          'http://localhost:8080/streaming-target',
          { dispatcher: agent },
          () => failingWritable
        );
      },
      (err) => {
        assert.strictEqual(err.message, 'Write pipeline failure');
        return true;
      }
    );

    await agent.close();
  }

  console.log('All undici.stream contract assertions passed successfully.');
}

runContracts().catch((err) => {
  console.error(err);
  process.exit(1);
});