データを一気に取り出すのをやめて、DB から HTTP レスポンスまで Stream で繋いだらメモリ使用量が激減した

この記事は、採用管理サービス「HERP Hire」の CSV エクスポート機能を、DB のカーソルから HTTP レスポンスまで一本のストリームで処理するように書き換えた話です。実装例は TypeORM ですが、問題そのものは特定の ORM に固有のものではありません。コードは説明のために簡略化・抽象化しています。

TL;DR

• 大量の行を返す CSV エクスポートが、 「まず全件をメモリに取り出してから加工する」 実装だった（TypeORM の getMany()）
• リリース当初は問題なく動いていたが、データが日々増えるにつれて少しずつ悪化し、最終的に OOM でプロセスが再起動を繰り返すところまで来ていた
• 結果として Node.js プロセスのヒープが膨らみ、GC が CPU を占有して、同じプロセスで動いている無関係なリクエストのレイテンシまで悪化していた
• DB のカーソルから HTTP レスポンスまで、一度も配列に戻さずストリームで繋ぐように変更（実装は SelectQueryBuilder.stream() + Web Streams の TransformStream）
• メモリ使用量は劇的に下がり、可用性が SLO を下回っていた状態から回復。SLO 割れによって発生していたデプロイフリーズも解除できた
• ストリーム化で新たに踏んだ罠（ダウンロードのキャンセルで DB コネクションが返ってこない）と、その対処も書きます

背景：CSV エクスポートがサーバ全体を巻き込んで重くなる

HERP Hire には「選考データを CSV でダウンロードする」機能があります。企業ごとに大量の選考（candidacy）データがあり、1 件あたり求人・媒体・担当者など複数のテーブルを JOIN して 1 行の CSV にします。

元の実装は、ざっくりこうなっていました。

// Before: 全件をメモリに載せてから CSV にする
const rows = await repository.list(); // Entity[] が一気に返る
const csv = await csvSerializer.serialize(header, rows.map(toCsvRow)); // 巨大な文字列
ctx.body = csv;

素直で読みやすいのですが、この 1 リクエストの中で以下のピークが同時に立ちます。

1. MySQL から返ってきた生の行（RowDataPacket）の配列
2. ORM が生成したエンティティの配列
3. CSV 行オブジェクトの配列
4. 直列化された CSV 文字列（UTF-16LE 変換前・変換後）

しかも JOIN しているので、エンティティ 1 件に対して生の行は複数行になります。件数が増えると、素直に線形にメモリを食っていきます。

厄介なのは、これがそのリクエストだけの問題では終わらないことです。Node.js は基本的にシングルスレッドなので、

• 大量のオブジェクト生成 → GC が頻繁かつ長く走る
• 変換処理が同期的に回り続けている間はイベントループが返ってこない

という形で、同じプロセスに来ている他のリクエストのレイテンシが揃って悪化する。1 つのエンドポイントが、無関係な機能の可用性まで道連れにしてしまう構造です。

これは ORM の問題ではなく、「一気に取り出す」形の問題

この記事では TypeORM を例にしますが、根っこにあるのは ORM 固有の話ではありません。

• 生 SQL でも、結果セットを全部配列に読み込めば同じことが起きる
• 外部 API を全ページ取得して結合してから処理する実装も同じ
• ファイルやオブジェクトストレージを丸ごと読んでからパースする実装も同じ

共通しているのは、入力サイズが増える一方なのに、処理のピークメモリが入力サイズに比例する形になっていることです。「一気に取り出して、全部持ったまま加工して、最後に返す」という形そのものが、いずれ限界を迎えます。逆に言えば、対策も個別の ORM のテクニックではなく「入力を最初から最後まで少しずつ流す」という一般的な形になります。

方針：どこか 1 か所ではなく、端から端までストリームにする

「メモリが厳しいならページングすればいい」とも考えましたが、この処理では最終的に 1 本の CSV を返す必要があり、途中でどこか 1 か所でも配列に戻すと、そこがピークになってメリットが消えます。

そこで、DB のカーソルから HTTP レスポンスまで、配列に戻さず一本のパイプで繋ぐ方針にしました。

MySQL cursor
  → 生の行 (RowDataPacket)
  → エンティティ単位に chunk 分割
  → Entity
  → CSV 行オブジェクト
  → CSV 文字列
  → UTF-16LE + BOM
  → HTTP response

実装 1：DB のカーソルからエンティティを組み立てる

以降は TypeORM での具体例です。

TypeORM の SelectQueryBuilder には stream() があります。ただしこれが返すのはドライバの生の行で、getMany() のようにエンティティへは変換してくれません。リレーションを解決したエンティティが欲しいので、getMany() が内部でやっている変換処理を自分で呼ぶことにしました（RawSqlResultsToEntityTransformer、RelationIdLoader、RelationCountLoader あたり）。

ここで JOIN 特有の問題があります。エンティティ 1 件が生の行の複数行に散っているため、行を適当な個数で区切って変換すると、同じエンティティが chunk の境界で分断されて重複・欠損する。

対策として、メインエイリアスの主キーが変わった位置で chunk を切る Transform を挟みました。

// 主キーが変わったところで chunk を区切る（JOIN で 1 entity が複数行になるため）
function chunkByPrimaryKey<Row extends object>(pkColumns: string[]): TransformStream<Row, Row[]> {
    let buffer: Row[] = [];
    let currentKey: string | null = null;

    const keyOf = (row: Row) => pkColumns.map(c => String(row[c])).join(';');

    return new TransformStream({
        transform(row, controller) {
            const key = keyOf(row);
            if (currentKey !== null && currentKey !== key) {
                controller.enqueue(buffer);
                buffer = [];
            }
            currentKey = key;
            buffer.push(row);
        },
        flush(controller) {
            if (buffer.length > 0) controller.enqueue(buffer);
        },
    });
}

そして、行 chunk → エンティティの変換。

function rowsToEntities<Row, Entity>(...): TransformStream<Row[], Entity> {
    return new TransformStream({
        async transform(rows, controller) {
            // リレーション解決は非同期。TransformStream なら async transform がそのまま書ける
            const relationIds = await relationIdLoader.load(rows);
            const relationCounts = await relationCountLoader.load(rows);
            for (const entity of buildEntities(rows, relationIds, relationCounts)) {
                controller.enqueue(entity);
            }
        },
    });
}

ここは ORM の内部実装（エンティティ生成）を自前で呼んでいる部分です。複雑なクエリで期待通りに動く保証はないので、この経路を使うクエリには必ずテストを書く方針にしました。コードにもその旨のコメントを残しています。

Node Streams ではなく Web Streams を選んだ理由

既存のコードは stream.Transform を継承したクラスで書かれていましたが、今回は Web Streams（node:stream/web の TransformStream）に寄せました。

• transform を async 関数としてそのまま書ける（リレーション解決の await が自然に書ける）
• pipeThrough で繋げるので、パイプラインの形がコードの見た目と一致する
• Readable.toWeb() / Readable.fromWeb() / Duplex.toWeb() で Node Streams 側と相互変換できるので、境界だけ変換すれば既存資産と共存できる

型定義の都合で @types/node を上げる必要があったり、Duplex.toWeb() の返り値の型が実態と合っていなくてキャストが必要だったりと、細かい摩擦はありました。

実装 2：CSV 直列化とレスポンスまで繋ぐ

CSV の直列化には csv-stringify を使っていて、これは Node Streams の Duplex を返します。Duplex.toWeb() で TransformStream に変換すれば、そのままパイプラインに挟めます。

const csvStream = entityStream
    .pipeThrough(new TransformStream({
        transform(entity, controller) {
            controller.enqueue(toCsvRow(entity));
        },
    }))
    .pipeThrough(csvSerializer.serialisationTransform(header));

Koa のレスポンスは Node Streams を期待するので、最後に戻します。Excel 向けに UTF-16LE + BOM に変換する処理も、そのままパイプで繋がります。

ctx.body = Readable.fromWeb(csvStream)
    .pipe(iconv.decodeStream('utf8'))
    .pipe(iconv.encodeStream('utf16le', { addBOM: true }));

これで、レスポンスの最初のバイトが流れ始めるまでの時間（TTFB）も短くなりました。ユーザから見ると「押してから無言で待たされる時間」が減ります。

段階的に移行するための工夫

この手の書き換えで怖いのは、既存の呼び出し箇所を全部道連れにしてしまうことです。今回は 2 つの逃げ道を用意しました。

1. collect() を用意して配列 API を維持する

内部実装をストリームに置き換えつつ、Stream<Entity> → Entity[] に畳む関数を用意して、既存の「全件取得」インターフェースはそのまま残しました。メモリのメリットは出ませんが、既存の呼び出し箇所は無変更で動きます。ストリーム化する価値があるところから順に移していけます。

export async function collect<T>(stream: ReadableStream<T>): Promise<T[]> {
    const reader = stream.getReader();
    const result: T[] = [];
    for (;;) {
        const { done, value } = await reader.read();
        if (done) break;
        if (value !== undefined) result.push(value);
    }
    return result;
}

2. フィーチャーフラグで新旧を切り替える

ユースケース層でフラグを見て、新（ストリーム）と旧（配列）を出し分けました。エンティティ生成を自前実装している以上、複雑なクエリで壊れる可能性はゼロにできません。「まず一部の企業だけ有効化して様子を見る」「壊れたらフラグを落とす」ができる状態を先に作るのが、この変更では一番重要な設計判断だったと思っています。

罠：ダウンロードがキャンセルされると DB コネクションが返ってこない

リリース後に見つかった、ストリーム化で新しく生まれた問題です。

ユーザがダウンロードを途中でキャンセルしたり、ブラウザを閉じたりすると、HTTP のコネクションが切れます。すると下流のストリームが閉じ、上流の DB カーソルは「まだ全部読まれていない」状態で放置されます。queryRunner.release() に到達しないので、コネクションプールが枯渇していく。全件をメモリに読み切っていた頃には起き得なかった種類の障害です。

対処は「途中で切られたら、残りの結果を読み捨ててからコネクションを解放する」。

function makeDrainable(source: Readable): { stream: Readable; drain: () => Promise<void> } {
    const passThrough = new PassThrough({ objectMode: true });
    let ended = false;
    source.on('end', () => { ended = true; });
    source.pipe(passThrough);

    function drain(): Promise<void> {
        if (ended) return Promise.resolve();
        return new Promise(resolve => {
            source.unpipe();
            source.on('data', () => { /* 読み捨てる */ });
            source.on('end', resolve);
            source.resume();
        });
    }
    return { stream: passThrough, drain };
}

そして、パイプが中断されたら drain() してから release() します。

readStream
    .pipeTo(through.writable, { preventCancel: true }) // 上流を勝手に cancel させない
    .catch(error => {
        // HTTP コネクション切断時はここに undefined が来る（実装依存の挙動なので、
        // 「エラーがある場合だけログする」形にした）
        if (error !== undefined) {
            logger?.warn('Stream cancelled with error', { error });
        }
        return drain();
    })
    .finally(async () => {
        await queryRunner.release();
    });

ポイントは 2 つです。

• preventCancel: true：下流が閉じたときに上流を即キャンセルさせず、こちら側で後片付けの主導権を握る
• この pipeTo() の Promise は await してはいけません。ストリーム完了時に解決されるので、await するとレスポンスを返す前に全件読み切ることになり、ストリーム化の意味が消えます。

「途中で切られたときにリソースをどう返すか」は、ストリーム設計では機能そのものと同じくらい考える価値があります。

CPU 占有への対策も忘れずに

メモリが解決しても、変換処理が延々と同期的に回ればイベントループは詰まります。一定件数ごとに setImmediate で明示的に yield する Transform をパイプラインに挟み、他のリクエストが割り込めるようにしました。

// N 件ごとにイベントループを他へ譲る
function createYieldTransform<T>(threshold: number): TransformStream<T, T> {
    let counter = 0;
    return new TransformStream<T, T>({
        async transform(chunk, controller) {
            controller.enqueue(chunk);
            counter++;

            if (threshold <= counter) {
                counter = 0;
                await new Promise<void>(resolve => {
                    setImmediate(resolve);
                });
            }
        },
    });
}

ストリーム化は「メモリのピークを潰す」施策であって、「CPU を譲る」施策とは別物です。両方セットで初めて、他のリクエストへの影響が消えます。

結果

• ヒープ使用量が大幅に低下（グラフは上記スクショ）
• CSV ダウンロードによる他リクエストへの影響が解消し、可用性が SLO を下回る状態から回復
• SLO 割れで止まっていたデプロイフリーズも解除

一番効いたのは、特定の重いエンドポイントが、無関係な機能の可用性を人質に取る構造をなくせたことだと思っています。

まとめ

• 「全部取り出してから加工して返す」実装は、入力が増えると単体の遅さ以上の問題（GC・イベントループ占有）を引き起こす。ORM でも生 SQL でも外部 API でも同じ
• そしてその限界は、データの自然な増加によって静かに、連続的にやってくる。閾値を決めて監視し、OOM や SLO 割れになる前に着手できるかが分かれ目
• ストリーム化は端から端まで繋がって初めて効く。途中で配列に戻すと台無しになる
• Web Streams は async な変換とパイプラインの表現力が高く、toWeb / fromWeb で Node Streams と共存できる
• ストリーム化すると「途中で切られる」という新しい状態が増える。リソース解放の設計まで含めて完成
• ライブラリの内部実装に踏み込む変更では、フィーチャーフラグと逃げ道（collect()）を先に作ると精神的に楽

おわりに：「最初は問題なかった」実装が、静かに限界を超えていく

ここで強調しておきたいのは、この実装は書かれた当時、何も間違っていなかったということです。データ量がまだ小さかった頃は、全件をメモリに載せても一瞬で終わり、可読性の高い素直なコードとして正しく機能していました。

問題は、データが増えたことです。しかもそれは、サービスとして望ましい増え方でした。ユーザが使い続ければ選考データは積み上がっていく。歓迎すべき成長そのものが、そのまま負荷の増加になります。

厄介なのは、その壊れ方が「ある日突然」ではないことでした。

• サーバの負荷が、日ごとに少しずつ上がっていく
• 「何件までなら大丈夫」という閾値が、どこにも定義されていない
• どのリリースが原因でもないので、犯人探しが効かない
• 昨日まで動いていたので、今日も大丈夫に見える

そうして、少しずつ悪化した末に、OOM でプロセスが繰り返し再起動する状態にまで到達しました。この時点で兆候は十分に観測できていて、原因の当たりもついていたのですが、「まあ再起動してるし大丈夫でしょう」「今この瞬間は一応サービスが動いている」と判断されたため、修正の優先度はなかなか上がりませんでした。優先度が上がったのは、可用性が SLO を下回り、社内ルールでデプロイフリーズがかかった後です。1 つのエンドポイントが原因で、チーム全体のデリバリーが止まりました。

この経験から言えることは、たぶんこの記事の技術的な内容よりも汎用的です。

• 「初期実装として妥当」は「永遠に妥当」を意味しない。 コードは変わっていなくても、前提条件（データ量）のほうが動く
• 連続的に悪化する問題には、明示的な閾値と監視が要る。 「何件でどれだけメモリを使うか」を測って可視化しない限り、誰も危険な水位に気づけない
• 障害になる前の劣化は、緊急に見えない。 だからこそ、「まだ落ちていない」段階で手を打てるかどうかがチームの体力を決める

障害が起きてから直すと、修正のコストに加えて、フリーズ期間・調査・信頼の回復ぶんのコストを払うことになります。落ちる前に直せば、払うのは修正コストだけです。

なお、この改修自体は SLO 割れが起きる 2 ヶ月前には手元で動いていました。技術を成果と評価に繋げるためには、どう見せて伝えるかがとっても大事らしい、というのが今回の一番大きな学びでした。