プロンプトとコンテキスト
ある Node.js サービスは大容量のアップロードを受信し、展開、ウイルススキャン、フォーマット検証、およびオブジェクトストレージへの書き込みを順次実行する必要があります。以前の実装では data イベントを手動でリッスンしていたため、メモリ消費が増加することがあり、クライアント切断後も実行を継続したり、中間ステップが失敗した際に一時ファイルが残留したりしていました。
stream.compose() または同等のパイプライン構成手法を使用してください。readable、writable、および Transform の各ステージがどのように接続され、バックプレッシャー、AbortSignal、エラー伝播、および最終クリーンアップによって単一のジョブがどのように制御可能に保たれるかを説明してください。
面接官がテストするポイント
pipe、pipeline、およびcomposeの境界とライフサイクルを区別できているか。- 単にキューの上限を引き上げるのではなく、バックプレッシャーによってどのように生成を制限するかを説明できるか。
- キャンセル、例外、およびクライアントの切断がパイプライン全体に伝播するか。
- 非同期ジェネレータ（async generator）、リソース解放、冪等な書き込み、および可観測性を適切に扱えるか。
確認すべき質問
- 入力は HTTP リクエスト、ファイル、オブジェクトストレージ SDK のどれから来ますか？マルチパート出力やリトライはサポートされていますか？
- 各ステージは Node ストリーム、Web Stream、AsyncIterable、通常の関数のいずれですか？
- スキャンと検証によって子プロセス、一時ファイル、またはデータベースの状態が作成されますか？
- オブジェクトストレージはマルチパートアップロードを中止できますか？また、キャンセルされたジョブは再開する必要がありますか？
30秒の回答
各ステージを readable、writable、Transform、または AsyncIterable として定義し、stream.compose() を使って Duplex に合成した上で、pipeline で最終宛先へパイプします。下流（downstream）がデータを受け入れ可能な場合のみプロデューサーが処理を進めるため、バックプレッシャーによってメモリが制限されます。リクエスト切断、デッドライン、ビジネス要件によるキャンセルは、対応するステージに渡される単一の AbortSignal を共有し、いずれかのエラーが発生するとチェーン全体が失敗します。一時ファイル、子プロセス、マルチパートアップロードは finally または abort ハンドラでクリーンアップされ、書き込みには冪等性キーを使用します。メトリクスには、スループット、キューの深さ、ピーク RSS、キャンセル理由、およびクリーンアップ結果を含めます。
ステップバイステップの詳細解説
各ステージのコントラクトを定義する
入力と出力の型、チャンクサイズ、null が許可されるか、ブロッキング動作、および完了時のリソースの所有権を指定します。非同期ジェネレータはソースを消費して要求に応じて yield する必要があり、通常の関数はファイル全体を暗黙的にメモリに読み込んではいけません。
再利用可能なステージを合成する
compose は、ストリーム、AsyncIterable、または関数を新しい Duplex に接続し、隣接するステージをパイプラインセマンティクスで処理します。例：
import { compose } from 'node:stream';
async function* validate(source) {
for await (const chunk of source) {
checkChunk(chunk);
yield chunk;
}
}
const processing = compose(decompress(), validate, scan());実際の実装では、処理を出力先 writable に接続し、ステージごとにエラーを握りつぶすのではなく、完了とエラーを一元管理する必要があります。
バックプレッシャーをデフォルトのコントロールプレーンにする
下流の準備ができていない場合、Readable および Transform ステージは生成を停止する必要があります。data コールバック内で無制限に push したり、high-water mark を際限なく引き上げて低速なコンシューマを隠蔽したりしてはいけません。最も低速なステージが全体の速度を決定するように、負荷テストで各キュー、スループット、RSS を記録する必要があります。
キャンセルと切断を伝播する
リクエスト切断、デッドライン、手動キャンセルを1つの AbortController に統合します。そのシグナルをサポートされている compose ステージおよび外部 SDK に渡します。中止後は読み取りを停止し、下流を破棄（destroy）して close を待機します。ログでは通常の abort、ビジネスエラー、ネットワークエラーを明確に区別する必要があります。
エラーとクリーンアップを一元管理する
単一のオーナーが pipeline/compose を呼び出し、最初のエラーを受け取ってチェーンを破棄する必要があります。一時ファイル、スキャナー、ソケット、マルチパートアップロードは、成功、失敗、キャンセルのいずれの場合でも解放されなければなりません。クリーンアップの失敗は、元のエラーを上書きすることなくジョブ ID とともにアラートを発行します。
冪等性とメトリクスを設計する
アップロード ID とステージバージョンから冪等性キーを生成し、オブジェクトの書き込みが完了した後にのみビジネス状態をコミットします。入力・出力バイト数、処理時間、ピーク RSS、キャンセル理由、失敗したステージ、クリーンアップ時間を記録します。重複書き込みを防ぐため、回復可能な境界からのみリトライします。
模範解答
アップロード、展開、スキャン、検証、ストレージを、明確な入力・出力・所有権を持つステージとして定義し、それらを readable、writable、または async-iterable パイプラインに合成して、1つのパイプラインオーナーに実行させます。下流の消費速度がバックプレッシャーを制御し、data ハンドラでの無制限なバッファリングは禁止します。切断、タイムアウト、手動キャンセルは、サポート対象ステージとストレージ SDK に渡される AbortSignal を共有します。オーナーが最初のエラーを処理してチェーンを破棄し、一時ファイル、スキャナー、マルチパートアップロードは成功、失敗、キャンセルのいずれでもクリーンアップされます。冪等性キーで書き込みを保護し、メトリクスでスループット、キュー、RSS、キャンセル、クリーンアップを監視し、リトライは安全な境界からのみ開始します。
よくある間違い
- ストリーム全体を Buffer に収集しているにもかかわらず、compose を使用していると主張する。
- 最後の writable の
errorのみをリッスンし、非同期ジェネレータや子プロセスの障害を見落とす。 - クライアントが切断された後も読み取りと書き込みを継続する。
- high-water mark を際限なく増やすことでバックプレッシャーに対処しようとする。
- 成功時のみ一時ファイルを削除し、中止や例外発生時を無視する。
- 冪等性キーなしでリトライを行い、オブジェクトやビジネス状態を重複させる。
フォローアップ質問
compose と pipeline はどう違いますか？
Compose はステージ群から再利用可能な Duplex を作成します。pipeline はエンドツーエンドの接続を駆動し、エラーを伝播させ、クローズを待機します。両者を組み合わせることも可能ですが、所有権とエラー処理は一元化する必要があります。
非同期ジェネレータが例外をスローするとどうなりますか？
合成されたストリームは失敗し、隣接するステージを破棄（destroy）します。呼び出し元はプロセスの終了のみに依存するのではなく、パイプラインのクリーンアップを待機し、失敗したステージを記録します。
バックプレッシャーが機能していることをどう検証しますか？
制御された低速なシンクに対してそれより高速なプロデューサーを用いて負荷テストを行い、制限されたキュー深度と RSS を観測して、スループットが無制限のバッファリングではなく最も低速なステージに追従することを確認します。
キャンセル直後にリトライできますか？
まずすべてのリソースが閉じられ、一時的な状態が識別可能であることを確認してください。冪等で回復可能な境界からリトライし、中断できない外部書き込みについては補償処理を行うか未知の状態としてマークします。