前提と注意事項

この記事では以下を前提としています。

• データベースには Supabase の PostgreSQL を使用しています。
• ワーカーは Next.js 16.4.2 の内部 API で、Vercel の関数として動かしています。
• キューには pgmq などの拡張機能を使わず、PostgreSQL の表と関数だけで実装します。既存のエクスポート処理と同じ形にそろえたかったためです。
• 中級のエンジニアの方を想定し、SQL と TypeScript の基本的な知識があることを前提とします。

この記事のゴール

表と関数だけで、リース（処理の持ち時間）、世代番号、バックオフ（再試行の間隔を延ばす仕組み）を備えたジョブキューを作ります。そして、複数のワーカーが同時に動いても、同じジョブを二重に処理せず安全に処理できるところまでをゴールとします。

今回の題材は、受信したメールの添付ファイルを取りに行く処理です。この処理を、Webhook の受信処理とは別のジョブとして切り出します。ワーカーは cron と、Webhook の後に実行される after() の両方から起動されます。そのため、同じジョブを2つの実行が同時に扱わないようにする必要があります。

構成概要

全体の構成は以下のとおりです。

ワーカーは claim_document_jobs 関数でジョブを取り出します。その後は、延長・完了・失敗・戻しの各関数を、ジョブの ID と世代番号つきで呼びます。

この仕組みの要点は以下の4つです。

• FOR UPDATE SKIP LOCKED でジョブを取り出します。
• リース（期限）で、ワーカーが途中で落ちたジョブを拾い直せるようにします。
• 世代番号（generation）で、古いワーカーの結果を捨てます。
• 試行回数とバックオフで、失敗したジョブを間隔を空けて再試行します。

ジョブ用の表を作成する

まず、ジョブを管理する document_jobs 表を作成します。

create table document_jobs (
  id uuid primary key default gen_random_uuid(),
  kind text not null,                 -- fetch_email など
  fax_received_email_id uuid,         -- 対象
  status text not null default 'queued' check (status in ('queued', 'running', 'completed', 'failed')),
  attempts integer not null default 0,
  max_attempts integer not null default 5,
  run_after timestamptz not null default now(),  -- この時刻まで拾わない（バックオフ）
  lease_until timestamptz,                         -- リースの期限。切れたら拾い直す
  generation integer not null default 0,           -- 取り出すたびに +1
  last_error_code text,
  last_error_message text,
  created_at timestamptz not null default now(),
  started_at timestamptz,
  finished_at timestamptz
);

主な列の意味は以下のとおりです。

• kind: ジョブの種類です。この記事では fetch_email（メールの添付を取得する処理）を使います。
• fax_received_email_id: ジョブの対象となるデータの ID です。
• status: ジョブの状態です。queued（待機中）、running（処理中）、completed（完了）、failed（失敗）のいずれかです。
• attempts と max_attempts: 試行回数と、その上限です。上限は 5 回です。
• run_after: この時刻になるまでジョブを取り出さないための列です。バックオフに使います。
• lease_until: リースの期限です。この時刻を過ぎた running のジョブは、ワーカーが落ちたものとみなして拾い直します。
• generation: 世代番号です。ジョブを取り出すたびに 1 ずつ増えます。

次に、未完了のジョブを1つに制限する一意インデックスを作成します。

-- 同じ対象・同じ種類の未完了のジョブは1つまで
create unique index on document_jobs (kind, fax_received_email_id) where status in ('queued', 'running');

これにより、同じ対象に対して、同じ種類の queued または running のジョブを2つ以上作れなくなります。

ジョブを取り出す関数を作成する

次に、ジョブを取り出す claim_document_jobs 関数を作成します。この関数は、以下の処理を行います。

• queued で run_after の時刻が来たジョブと、running でリースが切れたジョブ（ワーカーが途中で落ちたもの）を、古い順に取り出します。
• 取り出すたびに attempts と generation を 1 ずつ増やし、lease_until を「現在時刻 + 5分」にします。

create or replace function claim_document_jobs(
  p_kind  text,
  p_limit integer default 1,
  p_lease interval default interval '5 minutes'
)
returns setof document_jobs
language plpgsql
security definer
set search_path = public
as $$
begin
  return query
    with claimed as (
      update document_jobs j
         set status = 'running',
             attempts = j.attempts + 1,
             generation = j.generation + 1,
             lease_until = now() + p_lease,
             started_at = coalesce(j.started_at, now())
       where j.id in (
         select id from document_jobs
          where kind = p_kind
            and attempts < max_attempts
            and ((status = 'queued' and run_after <= now())
                 or (status = 'running' and lease_until < now()))
          order by run_after, created_at
          limit p_limit
          for update skip locked
       )
       returning j.*
    )
    select * from claimed order by run_after, created_at;
end;
$$;

実際の運用では、リースが切れて attempts が上限に達したジョブを先に failed にする処理も、この関数の冒頭に入れています。上限に達したジョブを拾い直さないためです。

SKIP LOCKED で同じ行を取らないようにする

for update skip locked を指定すると、他のトランザクションがすでにロックしている行を飛ばして、残りの行だけを取り出せます。そのため、複数のワーカーが同時にこの関数を呼んでも、同じ行を取り合いません。

limit と for update skip locked は、サブクエリの中に書きます。外側の update に直接 limit を書くことはできないためです。

service role だけが呼べるようにする

この関数は security definer で作成しており、関数の所有者の権限で実行されます。そのため、誰でも呼べる状態にならないように、実行権限を service_role だけに限定します。

revoke execute on function claim_document_jobs(text, integer, interval) from public, anon, authenticated;
grant execute on function claim_document_jobs(text, integer, interval) to service_role;

以上で、サーバー側で service_role のキーを使うワーカーだけが、この関数を呼べるようになります。

世代番号で古いワーカーの結果を捨てる

リースが切れると、別の実行がそのジョブを拾い直し、generation が進みます。このとき、元のワーカーが遅れて「完了」を書きに来ても、世代が違うため結果を捨てる必要があります。

そこで、完了・失敗・リースの延長の関数は、すべて (job_id, generation) を受け取ります。そして、世代が一致し、かつ running のときだけ更新します。完了の更新文は以下のとおりです。

update document_jobs
   set status = 'completed', finished_at = now(), lease_until = null
 where id = p_job_id and generation = p_generation and status = 'running';
-- 0行なら、リースを失っている（ワーカーはそのジョブの処理をやめる）

更新が 0 行だった場合は、そのワーカーがリースを失っていることを意味します。この場合、ワーカーはそのジョブの処理をやめます。

完了・失敗・リース延長の関数の全文は、上記の更新文を関数にしたものです。

リースを延長する

処理が長くなるときは、リースが切れないように途中で期限を延ばします。添付ファイル1件ごとに、以下のように extend_document_job_lease を呼びます。

extend_document_job_lease(job_id, generation, '5 minutes')

この関数が false を返した場合は、世代が変わっている（リースを失った）ということです。そのため、ワーカーはそのジョブの処理をやめます。

失敗時の再試行とバックオフを設定する

失敗の種類によって、以下のように扱いを分けます。

• 一時的な失敗（相手の API の 5xx や 429、通信エラー、時間切れ、Storage や DB のエラー）は、queued に戻して run_after を遅らせます。遅らせる間隔は、attempts に応じて 1分、5分、15分、60分と延ばします。
• 恒久的な失敗（404 など）は、再試行しません。
• 5回失敗したジョブは failed で止まります。

ジョブの状態遷移は以下のとおりです。

queued → running → completed
           │
           ├─ 一時的な失敗 → queued（run_after を遅らせる）
           │     バックオフ間隔: 1分 → 5分 → 15分 → 60分
           │
           └─ 恒久的な失敗、または 5 回失敗 → failed

なお、恒久的な失敗と一時的な失敗を判定する実装や、attempts からバックオフ間隔を計算するコードはここでは省略します。

試行に数えない戻し（release）を用意する

関数の時間の予算切れや、API キーの権限不足（401 や 403）は、ジョブに原因があるわけではありません。そのため、これらの場合は試行に数えずに queued へ戻す release の処理を用意します。attempts を 1 つ戻して queued にし、これも同じ世代のときだけ更新します。

update document_jobs
   set status = 'queued',
       attempts = greatest(attempts - 1, 0),
       run_after = now() + p_retry_after,
       lease_until = null,
       last_error_code = p_error_code
 where id = p_job_id and generation = p_generation and status = 'running';

failed のジョブを手動で再実行する

failed で止まったジョブは、管理画面から人が再実行できるようにしました。そのために、attempts を 0 に戻して queued にする関数を用意しています。

ワーカーを実装する

ワーカーは、Next.js の内部 API として実装します。1回の実行では、1件ずつ claim して処理し、終わってから次を claim します。1回の実行で処理するのは最大3件です。

また、時間の予算を 240 秒とし、残りが 30 秒を切ったら新しく claim しないようにします。claim したのに手を付けられず、attempts だけを消費してしまうことを避けるためです。

const deadline = Date.now() + 240_000;
for (let i = 0; i < 3; i++) {
  if (deadline - Date.now() < 30_000) break;
  const { data: jobs } = await supabase.rpc("claim_document_jobs", { p_kind: "fetch_email", p_limit: 1 });
  const job = jobs?.[0];
  if (!job) break;
  await processJob(job); // 中で extend / complete / fail / release に job.id と job.generation を渡す
}

processJob の中では、延長・完了・失敗・戻しの各関数に、job.id と job.generation を渡します。

内部 API は Vercel の関数として動かし、maxDuration = 300 を指定しています。240 秒の予算は、この 300 秒の上限に余裕を持たせた値です。

cron と after() からワーカーを呼ぶ設定は、この記事では詳しく扱いません。 pg_cron を使ったスケジュール実行については、以下の記事で解説しています。

また、Database Function から HTTP リクエストを送信する方法や、シークレットの保存方法については、以下の記事が参考になります。

トラブルシューティング

外側の update に limit を書くとエラーになる

PostgreSQL の update 文には、直接 limit を書けません。そのため、id を選ぶサブクエリを作り、その中に limit と for update skip locked を書きます。claim_document_jobs 関数は、この構成にしています。

完了や失敗の記録が DB エラーで失敗した

完了や失敗を記録する処理そのものが DB エラーで失敗すると、ジョブは running のまま残ります。ただし、リースが切れれば次の claim がそのジョブを拾い直します。そのため、この場合は特別な処理を入れず、そのままにしました。

外部 API 呼び出し後に処理が止まり、同じ処理が再実行される

外部 API を呼んだ後、DB に記録する前に処理が止まると、同じ処理がもう一度実行されます。そこで、保存先のパスを先に決めておき、保存を上書き（upsert）にします。これにより、二重に実行されても結果が増えないようにしました。

まとめ

この記事では、Supabase の PostgreSQL の表と関数だけで、リースつきのジョブキューを作る手順を解説しました。ポイントを整理すると以下のとおりです。

• for update skip locked を、limit とともにサブクエリの中に書くと、同時に呼ばれても同じ行を取りません。
• リースを設定すると、ワーカーが途中で落ちたジョブを拾い直せます。
• 世代番号を (job_id, generation) で渡し、一致するときだけ更新すると、古いワーカーの結果を捨てられます。
• 一時的な失敗はバックオフを付けて再試行し、ジョブに原因がない場合は試行に数えずに戻します。
• 保存先のパスを先に決めて upsert にすると、二重に実行されても結果が増えません。

複数のワーカーが同時に動く環境で、ジョブを二重に処理しない仕組みを作りたい方にとって参考になれば幸いです。