1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@sz_tak

キュー基盤なしで非同期ジョブを運用する— 1インスタンスのNext.jsで踏んだ5つの穴と、その塞ぎ方

1
Posted at

キュー基盤なしで非同期ジョブを運用する — 1インスタンスのNext.jsで踏んだ5つの穴と、その塞ぎ方

数分かかる処理を、キュー基盤なしで本番運用しています。

  • Next.js（App Router）1インスタンス
  • SQS / Cloud Tasks / BullMQ / Redis なし
  • cron ワーカーなし

個人開発でインスタンスを増やしたくなかったので、いわゆる fire-and-forget
（レスポンスを返してから、同じプロセスの中で処理を続ける）にしました。

これは普通に動きます。落ちるまでは。

この記事は、落ちたときに何が起きて、どう塞いだかの記録です。
コードは実際に動いているものから引いています。金銭が絡む穴が2つあって、
そこが一番痛かったです。

出発点：レスポンスを返してから処理を続ける

export async function startAssessmentJob(input: {
  userId: string; queryText: string; lang: string;
}): Promise<{ jobId: string }> {
  // 残高チェック (消費はまだしない)
  await ensureAssessmentPointAvailable(input.userId);

  const job = await createAssessmentJob({ /* status: 'pending' */ });

  // ★ await しない。レスポンスは先に返す
  void runAssessmentJob({ jobId: job.id, /* ... */ });

  return { jobId: job.id };
}

クライアントは jobId を受け取って、GET /api/.../jobs/{jobId} を
ポーリングします。ジョブの状態は DB のテーブルに持ちます。

status: pending → running → succeeded / failed

ここまでは教科書どおりです。問題はプロセスが死んだときです。

穴1：孤児ジョブ（ワーカーがいないのに running のまま残る）

コンテナが再起動したりデプロイが走ったりすると、実行中のジョブは消えます。
でもDBの行は running のまま残ります。

普通のキュー基盤なら可視性タイムアウトで別のワーカーが拾い直します。
ワーカーがいないので、誰も拾いません。ユーザーの画面は
「処理中…」のまま永久に回り続けます。

対処：ポーリングしてきた人に掃除させる

cron もワーカーも足さずに済ませました。状態を見に来る GET の中で、
古くなったジョブを failed に落とすという方法です。

// Reap orphans: jobs run as fire-and-forget tasks inside the web process,
// so a container crash leaves them stuck in pending/running with no worker.
const maxAgeMs = getAssessmentJobMaxAgeMs();      // 既定 5 分
const ageMs = Date.now() - new Date(initialJob.createdAt).getTime();

if ((initialJob.status === "pending" || initialJob.status === "running") && ageMs > maxAgeMs) {
  const reaped = await reapStaleAssessmentJob({ jobId: initialJob.id, maxAgeMs });
  if (reaped) {
    await finalizeReapedAssessmentJob(reaped);   // 後述
  }
}

「結果を知りたい人」は必ずポーリングしてくるので、掃除のトリガーが
無料で手に入るのがこの方法の良いところです。

同じ発想を期限切れデータの削除にも使っています。イベント受け口と
プレチェックの API から、頻度を絞って fire-and-forget で掃除を呼びます。
定期実行の仕組みを持たなくても、アクセスがある限り掃除は回ります。

限界も明確です。 誰もポーリングに来なければ掃除されません。
掃除が「いつ実行されるか」を保証したいなら、この方法は使えません。

穴2：回収は「条件付き UPDATE + RETURNING」で原子的にやる

上の reaper は、複数のタブから同時にポーリングされたら同時に走ります。
select して if (古い) して update と書くと、2つのリクエストが
両方「自分が回収した」と判断します。 回収のあとに返金処理が続くので、
二重返金になります。

なので1本の条件付き UPDATE に畳んで、returning で「取れたか」を見ます。

const updated = await db
  .update(assessmentJobs)
  .set({ status: "failed", errorCode: "INTERNAL_ERROR", /* ... */ })
  .where(
    and(
      eq(assessmentJobs.id, input.jobId),
      inArray(assessmentJobs.status, ["pending", "running"]),   // ★ まだ終わっていない
      sql`${assessmentJobs.createdAt} < now() - make_interval(secs => ${maxAgeSec})`,
    ),
  )
  .returning();

const row = updated[0];
return row ? mapRow(row) : null;   // null = 自分は回収者ではない
  • status in ('pending','running') が楽観ロックの役割を果たす。
    先に回収した側が failed に変えているので、後続の WHERE はマッチしない
  • 経過時間の判定もSQL側でやる。アプリのクロックを信じない
    （インスタンスが増えたときにズレる）
  • returning が空配列なら「自分は回収していない」。ここで分岐する

SELECT してから UPDATE する形を、条件を WHERE に押し込んで1文にする。
この形は、キューを持たない設計では何度も出てきます。

穴3：reaper がユーザーに嘘をついた（金銭が絡む）

これが一番まずかったものです。

回収したジョブに、最初はこういうメッセージを入れていました。

診断処理が中断されました。ポイントは消費されていません。

親切に見えます。が、嘘になる窓があります。

処理の流れはこうです。

ポイント消費 → AI処理（数分） → 成功として記録

「ポイント消費」と「成功として記録」の間でコンテナが落ちると、
消費済みなのに failed になります。 そこに「消費されていません」と
表示すると、ユーザーは当然もう一度やります。そして2ポイント目を失います。

対処：メッセージを2段階に分ける

回収した瞬間は断言しない。消費の有無を調べてから上書きします。

// 中立メッセージで一旦 failed にする。ポイント消費の有無は別途確認するため、
// ここで「消費していません」と断言してはいけない (消費後に container crash した
// 場合に嘘になる)。詳細メッセージと自動 refund は finalizeReapedAssessmentJob で行う。
errorMessage: "診断処理が中断されました。状況を確認しています…",

そのあとで台帳を引いて、消費されていたら返金します。

export async function finalizeReapedAssessmentJob(reapedJob: AssessmentJobRow) {
  let consumeTx;
  try {
    consumeTx = await findAssessmentConsumeTransactionBySessionId(reapedJob.sessionId);
  } catch (err) {
    // 消費の有無が判定できない = 返還すべきかどうかも分からない状態。手で確認が要る。
    await reportError({ scope: "payment", code: "REAP_CONSUME_LOOKUP_FAILED",
                        severity: "critical", /* ... */ });
    await safelyUpdateMessage(reapedJob.id,
      "診断処理が中断されました。ポイントの状況についてはサポートまでお問い合わせください。");
    return;
  }
  // consumeTx があれば返金してその旨を、無ければ「消費なし」を書く
}

ポイントは3つあります。

  1. 消費の有無の唯一の真実は「台帳」。ジョブの status ではありません。
    セッションIDで消費トランザクションを引いて判定します
  2. 判定できないときは、断言せずにサポート導線に出す。 そして
    severity: critical で自分に通知を飛ばす。「分からない」を
    「大丈夫です」に丸めてはいけない
  3. ジョブの状態遷移とお金の状態は別の軸。同じテーブルで管理しようとすると
    必ずズレます

非同期ジョブで課金するなら、「課金した／していない」を後から
第三者的に判定できる台帳を必ず持ってください。これが無いと、
クラッシュ後に正しいメッセージを出す方法がありません。

穴4：事前チェックと実消費の間が数分空く（プラットフォームの持ち出し）

もうひとつ金銭の穴です。migration のコメントにそのまま書いてあります。

-- Why: the synchronous balance check in POST /api/area-assessment runs minutes
-- before consumePointForAssessmentStart, so two concurrent submits from the
-- same user previously both passed the pre-flight check, both launched a full
-- AI run, and only one was charged at the end. The user was billed correctly
-- but the platform absorbed the cost of the second AI run.

残高チェックは受付時、実際の消費は処理の中。その間が数分空くので、
同じユーザーが2回submitすると両方が事前チェックを通ります。
AI処理が2回走って、課金は1回。差額はこちらの持ち出しです。

ユーザーから見れば正しく1回分しか請求されていないので、
苦情が来ません。気づけません。

対処：部分ユニークインデックスで「同時に1件」をDBに守らせる

アプリ側で「実行中か確認してから insert」と書くと、穴2と同じ競合が起きます。
DBの制約にしました。

-- This partial unique index makes the second concurrent insert fail with a
-- 23505 unique_violation, which the route handler converts into a 409
-- "ASSESSMENT_IN_FLIGHT" response. Terminal rows (succeeded/failed) are
-- excluded by the WHERE clause so the same user can still queue a new job
-- once the previous one finishes or gets reaped.

create unique index if not exists assessment_jobs_one_in_flight_per_user
  on safeland.assessment_jobs (user_id)
  where status in ('pending', 'running');

WHERE 付きのユニークインデックス（部分インデックス）なので、

  • 実行中の行は1ユーザー1件まで
  • 終わった行（succeeded / failed）は対象外なので、何件でも溜められる
  • 2件目の insert は 23505 unique_violation で落ちる → 409 に変換して返す

アプリ側は例外を専用のエラー型に変換するだけです。

export class AssessmentJobInFlightError extends Error { /* ... */ }

排他制御をアプリで書かずにDBの制約で表現できるなら、そのほうが強い。
部分ユニークインデックスは「状態がアクティブな行だけ一意」を表せるので、
ジョブ管理と非常に相性が良いです。

なお穴1の reaper があることが前提です。pending/running のまま
永久に残る行があると、そのユーザーは二度と新しいジョブを作れません。
排他と回収はセットで設計する必要があります。

穴5：キャッシュミス時に同じ外部APIを同時に叩く

最後は小さいものですが、効果が分かりやすいパターンです。

1回の処理の中で複数の外部APIを並列に叩きます。結果はメモリにキャッシュ
しています。素直に書くとこうなります。

const cached = getCachedValue(key);
if (cached) return cached;
const fresh = await fetchFromApi();   // ← 同時に来たらここが N 回走る
setCachedValue(key, fresh, ttl);

キャッシュが埋まる前に同じキーの要求が並ぶと、全部が外部APIを叩きます。
同じタイルや同じ市区町村コードを複数のレイヤーが同時に要求するので、
これは普通に起きます。

対処：進行中の Promise 自体をキャッシュする（モジュール全体で41行）

const cacheStore = new Map<string, CacheEntry<unknown>>();
const pendingStore = new Map<string, Promise<unknown>>();

export function getOrCreatePendingValue<T>(key: string, factory: () => Promise<T>): Promise<T> {
  const existing = pendingStore.get(key);
  if (existing) {
    return existing as Promise<T>;   // 進行中のものに合流する
  }

  const pending = factory().finally(() => {
    pendingStore.delete(key);        // ★ 成否にかかわらず必ず消す
  });
  pendingStore.set(key, pending);
  return pending;
}

使う側はこうです。

const cached = getCachedValue<T>(cacheKey);
if (cached) return cached;

return getOrCreatePendingValue(cacheKey, async () => {
  const payload = await fetchOnce(url);
  setCachedValue(cacheKey, payload, ttlMs);
  return payload;
});

値ではなく Promise をキャッシュするのが要点です。同じキーの2人目以降は
同じ Promise を await するので、外部APIは1回しか叩かれません。

.finally() で必ず削除するのを忘れないでください。.then() だけだと
失敗した Promise が残り続けて、以降ずっと同じエラーが返ります。

そしてこの方法はインスタンスをまたげません。 Map はプロセスローカル
なので、複数インスタンスなら台数分は叩かれます。1インスタンス前提の割り切りです
（そこが問題になる規模なら、その頃には Redis を入れるべきです）。

まとめ：キューを持たない設計の型

  1. fire-and-forget + 状態はDBに持つ。 レスポンスは先に返す
  2. 孤児は「見に来た人」に掃除させる。 cron もワーカーも足さずに済む。
    ただし「いつ実行されるか」は保証できない
  3. 回収は条件付き UPDATE + RETURNING で原子的に。
    status in (...) が楽観ロックになる。時刻判定はSQL側で
  4. クラッシュ後のメッセージで断言しない。 お金の真実は台帳。
    判定できないときは「分からない」と言ってサポート導線 + critical 通知
  5. 「同時に1件」はDBの部分ユニークインデックスで守る。
    where status in ('pending','running') で一意にして 23505 → 409
  6. 進行中の Promise をキャッシュして合流させる。
    .finally で必ず削除。プロセスローカルであることは自覚しておく

いつ本物のキューにすべきか

この構成で困らないのは、以下が全部成り立つときだけです。

  • インスタンスが1台（または状態をDBに寄せきれている）
  • 「ジョブが必ず実行される」ことを保証しなくてよい
    （落ちたら失敗として扱い、ユーザーにやり直してもらう／返金する、で許される）
  • 掃除の実行時刻を保証しなくてよい

リトライを自動でやりたくなった時点で、キューを入れたほうが安いです。
fire-and-forget にリトライを足すのは、可視性タイムアウトと冪等性を
自分で実装することになります。そこは車輪の再発明です。

逆に「失敗したら失敗として見せればいい」処理なら、
この6つを守るだけで本番に出せます。少なくとも自分の環境では、
コンテナ再起動でユーザーの画面が固まることも、
持ち出しが出ることも無くなりました。

1
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?