Piping ReadableStreams into Workers

A ReadableStream is transferable, which means the cheapest correct streaming pipeline between two threads is the one where you write no protocol at all.

This is the platform-native path described in Streaming & Backpressure Across Threads, part of High-Performance Computation Patterns. Transfer the stream, read it in the worker, and the browser propagates chunk delivery, backpressure, cancellation and errors across the boundary for you.

Minimal Working Example

A 3 GB CSV file, decoded and counted in a worker, with the page never touching a byte:

// page.ts
const worker = new Worker(new URL('./count.worker.ts', import.meta.url), { type: 'module' });

export function countRows(file: File): Promise<number> {
  return new Promise((resolve, reject) => {
    const stream = file.stream();                       // ReadableStream<Uint8Array>

    worker.addEventListener('message', function done(event: MessageEvent) {
      worker.removeEventListener('message', done);
      const data = event.data as { ok: true; rows: number } | { ok: false; error: string };
      data.ok ? resolve(data.rows) : reject(new Error(data.error));
    });

    // Transferable, not cloneable: the stream MUST appear in the transfer list.
    worker.postMessage({ kind: 'COUNT', stream }, [stream]);
  });
}

// count.worker.ts
self.addEventListener('message', async (event: MessageEvent) => {
  const { stream } = event.data as { stream: ReadableStream<Uint8Array> };

  try {
    const reader = stream.pipeThrough(new TextDecoderStream()).getReader();
    let rows = 0;
    let tail = '';

    for (;;) {
      const { value, done } = await reader.read();      // pulls one chunk when ready
      if (done) break;
      const text = tail + value;
      const lines = text.split('\n');
      tail = lines.pop() ?? '';                         // keep the partial last line
      rows += lines.length;
    }
    if (tail.length) rows++;

    self.postMessage({ ok: true, rows });
  } catch (error) {
    self.postMessage({ ok: false, error: (error as Error).message });
  }
});

Three characteristics make this different from a chunk-posting loop: the page’s memory never holds more than one chunk, the file is read lazily so a cancelled job stops reading from disk, and there is no protocol to get wrong beyond the single result message.

Walking the Critical Lines

[stream] in the transfer list. Streams are transferable and not cloneable. Omitting the list is a DataCloneError every time, which is a good failure — the alternative would be a silent copy.

The stream is unusable on the page afterwards. Transfer moves ownership exactly as it does for an ArrayBuffer. Reading file.stream() again gives a new stream over the same file, which is fine; reading the transferred object is not.

pipeThrough(new TextDecoderStream()) runs in the worker. The transform is constructed on the worker side, so the UTF-8 decoding cost — real on a multi-gigabyte file — lands on the worker’s thread, not the page’s. It also handles multi-byte characters that straddle a chunk boundary, which is a genuinely awkward thing to do by hand.

The tail variable. Chunk boundaries fall in the middle of lines. Keeping the partial last line and prepending it to the next chunk is the standard fix, and forgetting it produces a row count that is off by roughly the number of chunks — a bug that looks like a rounding error and is not.

The for(;;) loop instead of for await. Async iteration over a ReadableStream is a recent addition and not yet available everywhere. An explicit reader loop behaves identically and works in every engine that supports transferable streams at all.

Streaming Results Back

The reverse direction is symmetric, and it is how a worker returns a large result set progressively rather than in one 200 MB message. Create a TransformStream in the worker, transfer its readable end, and write into the writable end:

// worker.ts — produce a stream the page consumes
self.addEventListener('message', async (event: MessageEvent) => {
  const { stream } = event.data as { stream: ReadableStream<Uint8Array> };

  const out = new TransformStream<Float64Array, Float64Array>();
  // Hand the readable end to the page before writing anything into it.
  self.postMessage({ kind: 'RESULTS', readable: out.readable }, [out.readable]);

  const writer = out.writable.getWriter();
  const reader = stream.getReader();

  for (;;) {
    const { value, done } = await reader.read();
    if (done) break;
    await writer.ready;                 // backpressure: pauses if the page lags
    await writer.write(transform(value));
  }
  await writer.close();
});

// page.ts
worker.addEventListener('message', async (event: MessageEvent) => {
  if (event.data.kind !== 'RESULTS') return;
  const reader = (event.data.readable as ReadableStream<Float64Array>).getReader();
  for (;;) {
    const { value, done } = await reader.read();
    if (done) break;
    render(value);                       // slow rendering throttles the worker
  }
});

await writer.ready is the line that makes this a pipeline rather than a fire hose. It resolves while the internal queue has room and stalls when it does not, so a page that renders slowly automatically slows the worker’s loop — the same guarantee the credit protocol builds by hand, obtained here from the platform.

Cancellation and Errors

Both directions propagate, which is the second reason to prefer streams over a chunk protocol.

Cancellation. reader.cancel() on the consuming side propagates upstream through every transform to the original source. On a File stream that stops the disk read; on a fetch response body it aborts the download. Nothing in your code needs to notice.

const reader = readable.getReader();
abortButton.onclick = () => void reader.cancel(new Error('user cancelled'));

Errors. A throw inside a transform, or an underlying source failure, errors the stream. Every downstream read() rejects with that error — across the thread boundary — so a single try/catch around the consumer loop sees failures that originated in the other thread. The error object is structured-cloned, so a custom error class arrives as a plain Error; the preservation techniques are in Structured Error Serialization Across Threads.

One direction does not propagate: terminating the worker. worker.terminate() kills the thread without erroring the streams it held, so a page-side reader waits forever. Track worker lifetime separately and cancel readers explicitly when you terminate.

Composing Transforms Inside the Worker

Once the stream is on the worker side, everything the Streams API offers is available there — and each stage runs on the worker’s thread, which is the point.

// worker.ts — three stages, all off the main thread
const lines = new TransformStream<string, string>({
  transform(chunk, controller) {
    this.tail = (this.tail ?? '') + chunk;
    const parts = (this.tail as string).split('\n');
    this.tail = parts.pop() ?? '';
    for (const line of parts) controller.enqueue(line);
  },
  flush(controller) { if (this.tail) controller.enqueue(this.tail as string); },
} as Transformer<string, string> & { tail?: string });

const records = new TransformStream<string, Record<string, string>>({
  transform: (line, controller) => controller.enqueue(parseCsvLine(line)),
});

const reader = stream
  .pipeThrough(new TextDecoderStream())
  .pipeThrough(lines)
  .pipeThrough(records)
  .getReader();

Each pipeThrough adds a queue with its own high-water mark — one chunk by default for the transform streams above — so the pipeline’s total buffering is the sum of the stages, not unbounded. If a stage is expensive and bursty, raising its high-water mark (new TransformStream(transformer, { highWaterMark: 4 })) smooths it without changing anything upstream.

Two habits keep such pipelines debuggable. Give every stage a single responsibility, so a failure names itself in the stack. And put the flush handler on any stage that keeps state across chunks — the partial-line tail above is lost at end-of-stream without it, silently dropping the last record of every file.

Gotchas

Locked streams cannot be transferred. Calling getReader(), pipeThrough(), tee() or pipeTo() locks the stream, and a locked stream throws on transfer. Transfer first, read second. If the page genuinely needs a copy of the data too, tee() before transferring one branch — and remember that a tee’d stream buffers whatever the slower branch has not read, which reintroduces unbounded memory if one side stalls.

fetch bodies work, with one caveat. (await fetch(url)).body is a ReadableStream and transfers fine, so a worker can decompress or parse a download the page never sees. But an aborted transfer needs an AbortController on the fetch as well as a cancel on the reader if you want the connection closed promptly.

A TransformStream whose transform is slow blocks its own pipeline. Each stage runs on the worker’s single thread, so an expensive transform stalls every stage upstream of it. That is correct backpressure, but it means adding stages does not add parallelism — for that, the stages must live in different workers.

Service workers see the request, not the stream. If a service worker intercepts the fetch and constructs its own Response, the body the page transfers is the one the service worker produced. That is usually invisible and occasionally the explanation for a stream that ends early — see Service Workers for Computation.

A transferred stream cannot be shared with a second worker. Transfer moves it, so two workers cannot both read one stream. To fan out, either tee() on the producing side before transferring each branch — accepting that the slower branch buffers — or slice the source into independent ranges instead, as in Processing Large Files with Blob.slice in Workers.

Chunk sizes are not yours to choose. A File stream yields chunks the browser picks (commonly 64 kB); a fetch body yields whatever the network delivered. If the downstream work has a preferred granularity, insert a TransformStream that re-chunks, rather than assuming.

Performance Note

Counting rows in a 3 GB CSV on a 2023 laptop: the transferred-stream pipeline held under 12 MB of peak process memory beyond the baseline and completed in 21 s, disk-bound. The equivalent implementation that read the whole file with file.arrayBuffer() and posted it in one message peaked at 3.1 GB and failed outright on a machine with 8 GB of RAM. A chunk-posting loop without flow control finished in the same 21 s but peaked at 480 MB, all of it queue.