Streaming pipeline

// AsyncDisposableStack manages a 4-stage streaming pipeline.
// If any stage throws, all earlier stages are disposed in LIFO order.

async function runPipeline(scenario) {
  await using stack = new AsyncDisposableStack();

  // Acquire each stage — add to stack immediately so it's cleaned up on throw
  const reader = await openHttpReader(url);
  stack.defer(async () => { await reader.cancel(); log('Reader cancelled'); });

  const decoder = new TextDecoderStream();
  stack.defer(async () => { await decoder.readable.cancel(); log('Decoder closed'); });

  const transformer = new TransformStream({ ... });
  stack.defer(async () => { await transformer.readable.cancel(); log('Transformer closed'); });

  const writer = await openFileWriter(path);
  stack.defer(async () => { await writer.close(); log('Writer closed'); });

  // Pipeline is open; process stream
  for await (const chunk of reader.pipeThrough(decoder).pipeThrough(transformer)) {
    await writer.write(chunk);
    if (scenario === 'throw-transform') throw new Error('transformer failed');
  }

  // Happy path: stack disposes all stages in LIFO order when block exits
}

// On throw, stack unwinds: Writer → Transformer → Decoder → Reader
// All disposers run even if an earlier one throws