Streams API（バックプレッシャと内部キュー）

Streams API は「データを少しずつ流す」だけの仕組みに見えますが、その核心はむしろ流れすぎを止めることにあります。生産側が消費側より速いと、未処理データが無限に溜まってメモリを食い潰します。これを防ぐのが内部キューと高水位線（highWaterMark）に基づくバックプレッシャで、Streams 標準はこの圧力を生産側へ自動で伝える機構を厳密に定義しています。ここでは ReadableStream / WritableStream / TransformStream の内部キュー構造、desiredSize の算出、圧力の伝播経路、そしてリーダー/ライターのロック機構を原理から見ていきます。ボディが ReadableStream である Fetch API の内部 を押さえておくと、各概念が実務とつながります。

内部キューと高水位線：圧力はどう測られるか

すべてのストリームは内部キュー（チャンクの待ち行列）を 1 本以上持ちます。キューに溜まったチャンクの総「サイズ」と、許容上限である**高水位線（highWaterMark, 以下 HWM）**との差分が、ストリームの空き具合を表す desiredSize です。定義はシンプルで、desiredSize = highWaterMark - queueTotalSize です。ここで queueTotalSize は、各チャンクに size(chunk) を適用して足し合わせた値です。

チャンクのサイズをどう測るかは **queuing strategy（キュー戦略）**が決めます。標準が用意するのは 2 種類です。

たとえば new ReadableStream(source, new CountQueuingStrategy({ highWaterMark: 3 })) なら、キューに 3 チャンク溜まった時点で desiredSize が 0 になり、「これ以上は要らない」状態を示します。バイト基準なら TypedArray・ArrayBuffer のメモリレイアウト で扱う byteLength がそのまま重みになります。

ReadableStream：pull とバックプレッシャ

ReadableStream は start／pull／cancel を持つ underlying source（基盤ソース）から作ります。バックプレッシャの中核は pull がいつ呼ばれるかです。ストリームは内部キューに余裕がある（desiredSize が正）間だけ pull(controller) を呼んでソースに補充を促し、controller.enqueue(chunk) で積まれたチャンクでキューが満ちると pull の呼び出しを止めます。消費側が read() でチャンクを引き取ってキューが空けば、再び desiredSize が正に転じて pull が再開されます。

const stream = new ReadableStream({
  pull(controller) {
    // desiredSize が正の間だけブラウザが pull を呼ぶ。
    // enqueue でキューが満ちたら、次の read まで pull は呼ばれない。
    const chunk = produceOne();
    if (chunk === null) controller.close();
    else controller.enqueue(chunk);
  }
}, new CountQueuingStrategy({ highWaterMark: 3 }));

ここが重要な点で、pull を呼ぶか止めるかという制御そのものがバックプレッシャです。生産側コードは「ペースを落とせ」という命令を直接受け取るのではなく、「呼ばれないこと」によって自然に待たされます。controller.desiredSize を自分で読めば、1 回の pull でどれだけ補充すべきかを判断する材料にもできます。

WritableStream：ライター側の圧力

WritableStream は start／write／close／abort を持つ underlying sink（基盤シンク）へ書き込みます。こちらでも内部キューと HWM が効きますが、圧力の見方が逆向きです。writer.write(chunk) は Promise を返し、writer.desiredSize でシンクの空き具合が読めます。シンクの処理（write）が遅れてキューが HWM を超えると desiredSize が 0 以下になり、これが「書き込みを控えてくれ」という合図です。

正しい作法は、writer.ready の解決を待ってから次を書くことです。ready は desiredSize が正に戻るまで保留される Promise で、これを await することで生産側のペースがシンクの処理速度に同期します。

const writer = writable.getWriter();
for (const chunk of source) {
  // ready が解決＝ desiredSize が正＝書いてよい状態になるまで待つ。
  // これを省くとキューが無限に膨らみメモリを食う。
  await writer.ready;
  writer.write(chunk); // この write 自体を await する必要はない
}
await writer.close();

バックプレッシャの伝播：pipeTo と pipeThrough

個々のストリームの圧力を鎖としてつなぐのが pipeTo／pipeThrough です。readable.pipeTo(writable) は、書き込み先 writable の desiredSize が 0 以下の間、読み取り元 readable からの read() を控えます。read を控えれば readable のキューが空かず、その desiredSize も下がり、結果として readable の pull が止まる――こうして圧力が消費側から生産側へ一段ずつ遡って伝わります。

// 圧力は writable → readable の向きに自動で遡る。
// 遅い writable があると、上流の fetch のボディ取得まで自然に減速する。
await response.body
  .pipeThrough(new TextDecoderStream())
  .pipeTo(slowWritable);

pipeThrough(transform) は内部的に readable.pipeTo(transform.writable) と transform.readable の返却を組み合わせたもので、複数の TransformStream を数珠つなぎにできます。手書きの read()／write() ループでバックプレッシャを正しく実装するのは難しいため、可能なら pipe 系に任せるのが安全です。

TransformStream：2 つのキューを連結する

TransformStream は writable 側と readable 側の 2 つを 1 オブジェクトにまとめたものです。ts.writable へ書かれたチャンクが transform(chunk, controller) を通り、controller.enqueue() で ts.readable 側のキューへ流れます。内部にはキューが 2 本（writable 側と readable 側）あり、それぞれに HWM があります。

伝播はこうです。下流が遅くて ts.readable のキューが詰まると、transformer は新たな enqueue を控え、transform の呼び出しが進まなくなります。すると ts.writable 側のキューも空かず、その desiredSize が下がって上流に圧力が伝わる。TransformStream は 2 つのキューを連結することで、下流の圧力を上流へ素通しする継手として働きます。

// 1 行ごとに区切る変換器。下流が詰まれば transform が止まり、圧力が上流へ伝わる。
const lineSplitter = new TransformStream({
  transform(chunk, controller) {
    for (const line of chunk.split('\n')) controller.enqueue(line);
  }
});

リーダー/ライターのロック機構

ストリームの一貫性を守る最後の柱がロックです。readable.getReader() を呼ぶと ReadableStream はロックされ、以後その reader 経由でしか read() できません。ロック中は元の ReadableStream に対する pipeTo／pipeThrough／tee／別 reader の取得がすべて拒否されます。理由は明快で、複数の読み手が同じキューを同時に引くと、どのチャンクが誰に渡るか不定になり、ストリームの「順序保証・一度きり消費」が壊れるからです。WritableStream の getWriter() も同様にライターでロックします。

ロックは reader.releaseLock()／writer.releaseLock() で解放します。解放するまで、たとえ read を 1 つも発行していなくても元ストリームは触れません。

const reader = readable.getReader(); // ここで readable はロックされる
const { value, done } = await reader.read();
reader.releaseLock();                // 解放するまで pipe も別リーダーも不可
await readable.pipeTo(writable);     // 解放後なら OK

tee：分岐と未読バッファの代償

readable.tee() は 1 本の ReadableStream を独立に読める 2 本へ複製します（元ストリームはロックされる）。中身を 2 通りに処理したい、あるいは読みつつ別経路にも流したいときに使います。ただし設計上の代償があり、2 本の消費速度がずれると、速い側のために遅い側がまだ読んでいないチャンクを内部でバッファし続け、メモリが膨らみます。tee はバックプレッシャを「遅い方に合わせて止める」のではなく「速い方を進めるためにバッファする」挙動を既定とする点に注意してください。

ストリームをワーカ間で移送する場合は、ReadableStream/WritableStream/TransformStream 自体が **transferable（転送可能）**であり、structuredClone と転送可能オブジェクト の仕組みで postMessage の transfer リストに載せて所有権ごと渡せます。コピーではなく移送なので、元コンテキストでは使えなくなります。

まとめ