パイプライン処理を複数のworkerへ分散する設計｜PostgreSQLキュー・実行先プール・プロセス内状態の制約

概要

ファイルの到着・定期実行・Webhookをきっかけに動くデータパイプラインは、使われるほど実行件数が増えます。最初は1台のサーバーで十分でも、VLMで画像を読む処理や大きなExcelの変換が重なると、「待ちが長い」「別のパイプラインが詰まる」といった問題が出てきます。

このとき考えるべきことは主に3つあります。

1. workerより多いjobが来たとき、どう待たせ、どう捨てずに処理するか
2. どの処理を別プロセス・別マシンへ分けられ、どの処理は分けられないか
3. Webサーバーのプロセス数を単純に増やすと、何が壊れるか

DigitalBaseのパイプライン実行基盤は、オンプレミス・1台構成からの拡張を前提に、Redisなどの追加部品を使わずPostgreSQLだけでキューを構成しています。その設計で決めたことと、運用上の制約を整理します。

---

workerよりjobが多いと何が起きるか

パイプラインの実行（run）は、常駐するworkerプロセスが1本につき同時に1件ずつ処理します。つまり 同時実行数 = workerの本数 です。workerがすべて埋まっている間に届いたrunは、DB上で queued（待機）として積まれ、空いたworkerが順に取り出します。

以前の構成では、runごとにプロセスを起動し、同時実行の上限を超えた起動要求はその場で拒否していました。現在は常駐workerがキューから取り出す方式に変え、上限を超えた分は拒否せずに待たせています。

同じパイプラインは順番に処理する

同じパイプラインのrunは、実行中1本＋待機は複数 という制約にしています。ファイル1件ごとに台帳へ追記するような処理を並列に走らせると、書き込み順が入れ替わったり、同じ行を取り合ったりするためです。別のパイプライン同士は並列に動きます。

当初は「実行中に届いたトリガーは飛ばす」作りだったため、連続して届いたファイルの2件目以降が処理されないことがありました。そこで、実行中でも待機を複数積めるように改めています。

待ちが溢れたときの扱いは入口ごとに変える

待機できる本数には上限を設けています。上限を超えた場合の扱いは、起動の入口によって変えています。

ポイントは、「黙って消える」経路を作らないことです。cronも当初は実行中の時刻を飛ばしていましたが、「時刻が消える」ことが分かりにくいため、1本だけ待たせる形に改めています。

---

キュー設計：Redisは必要か

ジョブキューというとRedisやメッセージブローカーを思い浮かべがちですが、パイプラインのrunは1件あたりの処理時間が数秒〜数分と長く、秒間に大量の小さなジョブを捌く用途とは性質が異なります。オンプレミスで部品を増やすと、監視・バックアップ・障害時の切り分けの対象も増えます。そこで、すでに使っているPostgreSQLをそのままキューにしています。

取り出しは FOR UPDATE SKIP LOCKED

複数のworkerが同時に同じrunを取らないよう、行ロックを取りながら、他のworkerがロック中の行は飛ばして次を取ります。以下は一般化した例です。

-- 同じジョブの実行中は1本だけ（部分ユニークインデックスで保証）
CREATE UNIQUE INDEX job_runs_one_running
    ON job_runs (job_id) WHERE status = 'running';

-- 空いたworkerが、自分のプールの待機runを古い順に1件取る
UPDATE job_runs
   SET status = 'running', worker_host = :host, worker_pid = :pid
 WHERE id = (
       SELECT r.id
         FROM job_runs r
         JOIN jobs j ON j.id = r.job_id
        WHERE r.status = 'queued'
          AND j.worker_pool = :pool
          AND r.job_id NOT IN (SELECT job_id FROM job_runs WHERE status = 'running')
        ORDER BY r.created_at
        LIMIT 1
          FOR UPDATE OF r SKIP LOCKED)
RETURNING id;

キューがDBにあるため、APIプロセスやworkerが落ちても待機中のrunは失われません。また、runの記録そのもの（status が queued の行）がキューを兼ねるので、キューとrunの記録を別々に管理して食い違わせる心配もありません。

アイドル時はDBを叩かない

workerがDBを数秒おきにポーリングすると、何もしていない時間帯でもディスクとログが動き続けます。そこで、投入時と終了時に NOTIFY を打ち、待機中のworkerは LISTEN で起きる形にしています。定期的な見回りは保険として長い間隔で1回だけです。

二重起動と取りこぼしを防ぐ

無人運転では「同じイベントを2回処理する」「停止中のイベントを取りこぼす」ことが問題になります。次の仕組みで防いでいます。

• 冪等キー：イベント起動はイベントのID、WebhookはDelivery IDなどのヘッダー、cronの追いつき実行は逃した時刻をキーにし、同じキーのrunは作らず既存のrunを返します
• イベントのoutbox：ファイル到着などのイベントはいったんDBに書いてから配送し、失敗したら間隔を伸ばしながら再送、上限を超えたら隔離（dead letter）します
• 中断runの回収：workerが落ちて実行中のまま残ったrunは、設定した回数までは待機に戻し、超えたら失敗として記録します
• 連続失敗での一時停止：直近のrunが続けて失敗したパイプラインは自動起動を止め、手動実行で再開します

---

重い処理を別マシンへ：実行先プール

キューがDBにあるので、workerを本体の外に立てても仕組みは変わりません。workerがDBから仕事を取りに行く（pull）形なので、worker側でポートを開ける必要もありません。

各パイプラインには 実行先プール の名前を持たせ、外部のworkerは起動時に自分が担当するプール名を指定します。たとえば、VLMで画像を大量に読むパイプラインだけをGPUに近いマシンのプールへ回し、軽い集計は本体のworkerで動かす、といった分け方ができます。実行先の変更は資源配分にあたるため、管理者だけが行えるようにしています。

外部workerで押さえていること

前提として、外部workerは本体のPostgreSQLに接続でき、接続先の秘密情報を復号するための鍵を本体と揃える必要があります。ファイルを扱うパイプラインでは、ファイルの保存先をNFSなどで共有し、同じパスで見えるようにします。並列度は「1プロセスで同時1本、台数で増やす」という単純な考え方です。

---

Pythonサンドボックスなど、パイプライン以外の子プロセス

パイプラインのworker以外にも、本体は重い処理や利用者のコードを別プロセスに切り出しています。ただし、DBのキューで動き、別マシンに分けられるのはパイプラインのworkerだけです。それ以外は本体プロセス内のセマフォやロックで本数を制御しているため、同じマシンの中でしか動きません。

利用者のPythonをスレッドで動かすと、時間切れになっても止められません。そのためサンドボックスは必ず子プロセスで実行し、親子間はJSONだけでやり取りします。子プロセスに渡す環境変数も絞り、DB接続文字列やAPIキーなどの秘密を渡さないようにしています。

運用上の意味は、外部workerを足して増えるのはパイプラインの同時実行数だけということです。チャットからのPython実行やデータレイクの問い合わせが増えた場合の余力は、本体のマシンの性能で決まります。どの処理が律速になっているかを見極めずにworkerだけを増やしても、効果は限られます。

---

uvicornのworker数を増やすだけでは壊れるもの

「APIが重いならuvicornの --workers を増やせばよい」と考えたくなりますが、この構成ではAPIを 1プロセス で動かすことを前提にしています。ログイン失敗の回数、承認待ち、会話のメモリなど、数十の状態をプロセス内のメモリに持っているためです。プロセスを増やすと、リクエストがどのプロセスに届くかによって、次のような不整合が起きます。

これらを複数プロセスで共有するには、外部のストアに出す必要があります。現時点では 「同じDBを見るAPIは1台・1プロセス」 を前提とし、水平に広げるのはパイプラインの実行（外部worker）だけ、と割り切っています。本体を2台並べた場合、片方が落ちたときにその本体の内蔵workerが実行していたrunを誰も回収しない、という問題もあります。

---

運用で見るべき指標

キューを使う構成では、「処理が遅い」のか「処理が止まっている」のかを早く見分けることが重要です。メトリクスとしては次のような値を出しています。

• 最も古い待機runの経過時間（キューの滞留）
• workerが居ないのに実行中のまま残っているrunの数
• 配送待ちのイベント数と、最も古いイベントの経過時間
• 隔離（dead letter）されたイベントの数
• スケジューラーが動いているか

待機runの経過時間が伸び続けるなら、worker不足か、特定のパイプラインが長時間占有しているかのどちらかです。前者ならworkerやプールを足し、後者ならそのパイプラインを別のプールへ分けます。

---

まとめ

パイプラインの実行件数が増えたときの設計の要点は、次のとおりです。

1. 溢れたjobは捨てずに待たせる：同時実行数はworkerの本数で決め、超えた分はDBのキューで待機させる。入口ごとに上限超過時の扱いを決め、黙って消える経路を作らない
2. キューはPostgreSQLで足りる：FOR UPDATE SKIP LOCKED と LISTEN/NOTIFY、冪等キー、outboxで、追加部品なしに取りこぼしと二重起動を防ぐ
3. 広げられるのはパイプラインの実行だけ：外部workerを実行先プールで分け、重い処理を別マシンへ回す。Pythonサンドボックスなど他の子プロセスは本体のマシンの中で動く
4. APIのプロセス数は増やさない：承認待ち・会話のメモリ・セッションRAG・SQLジョブ・動画の完了待ちなどがプロセス内にあるため、uvicornのworker数を増やすと壊れる

「どこまでがキューで分散でき、どこからが1プロセス前提か」の境界を先に決めておくことで、負荷が増えたときにどこへ手を入れればよいかが明確になります。

---

関連記事

• CronとWebhookで構築する業務フロー自動化｜設計指針と5つの実装パターン
• LLM I/Oとデータ連携基盤の境界設計｜構造化出力・冪等性・スキーマ駆動・観測可能性