[Job] 現行jobsの永続性・重複投入・batch停止を止血する

[hmjn023]

Parent: #612

背景

現行workerには FOR UPDATE SKIP LOCKED によるatomic claim、AI/非AI concurrency、source単位の直列化がある。一方で jobs はUNLOGGEDで、dedupe制約・lease・retry情報が不足し、CCIP batch親が終端状態にならない経路もある。後続刷新の途中でも処理要求を失わないよう、まず現行系を止血する。

スコープ

• ALTER TABLE jobs SET LOGGED migration
• active job用dedupe keyとpartial unique制約
• claim token、lease/heartbeat、attempt、available_at、上限付きretry
• 親jobとdispatch job作成のtransaction化
• dispatch/child失敗時のparent reconciliation
• all-source CCIP batch、mixed success/failure、mediaIds[] status照会の既知不整合修正
• atomic claim、AI concurrency、source直列化の回帰テスト

非スコープ

• domain processing state導入
• import/download/batch/source syncの専用table化
• generic jobs の削除

受け入れ条件

• 異常終了・再起動後もjobが消えない
• 並行投入でも同一active jobが重複しない
• 期限切れclaimだけをtoken検証付き・上限付きで再取得できる
• dispatch・子jobの全失敗経路で親が終端状態になる
• all-source/mixed-result batchとstatus照会の回帰テストが通る
• 既存のatomic claimとconcurrency制御を維持する

主な参照

• apps/server/drizzle/0012_optimize_jobs_table.sql:1
• packages/db/src/schema.ts:898
• packages/db/src/repositories/job-repository.ts:70
• packages/db/src/repositories/job-repository.ts:371
• apps/server/src/infrastructure/jobs/job-worker.ts:122
• apps/server/src/infrastructure/jobs/ccip-jobs.ts:297

評価反映（2026-07-18）

既存スコープへ以下を追加要件として反映する。

dedupeの契約

• dedupe_key はpayload全体から暗黙生成せず、job typeごとに論理的一意性、merge、supersede規則を定義する。
  • media処理: 対象mediaと処理revision
  • source sync: source、sync種別、必要ならrevision
  • batch子処理: batchと対象item
• activeの定義は、実行待ちのavailable_atを含むpendingとin_progressとする。
• active rowへpartial unique制約を設定し、投入はread-then-insertではなく、同じ制約を利用したatomic insert/upsertで行う。
• sync_lancedb_deltaのpayload統合など、既存要求へmergeするtypeは、row lockまたは単一SQLで要求を失わず統合する。
• unique制約追加前に既存active duplicateを検出し、type別規則で統合・失敗・取消へreconcileする。暗黙に1件だけ残して削除しない。

現行createIfUniqueには先行SELECT後にINSERT/UPDATEする経路があるため、アプリケーション側チェックのみを重複防止の根拠にしない。

claim、lease、CAS、side effect fencing

• claimごとに新しいclaim_tokenを発行し、claim時のattempt_countとlease_expires_atを保持する。
• heartbeat、完了、失敗、retry予約、stale回収はすべてid + status + claim_tokenを条件にしたCASとし、更新件数0件はlease喪失として扱う。
• lease、heartbeat、available_atの判定にはDB時刻を使用し、worker間の時計差へ依存しない。
• workerはlease喪失を検知したら可能な範囲で処理を中断する。
• job rowの完了更新だけでなく、tag、CCIP、processing stateなどの業務出力もfenceする。DB出力は、claim tokenおよび要求revisionの検証と出力更新・完了更新を同一transaction内で行う。
• filesystemや外部AIなど同一transactionに含められないside effectは、claim/revision由来のidempotency key、revision別一時出力、atomic replaceなどにより、旧workerが新しい結果を上書きしない設計にする。
• stale workerによる完了・失敗・heartbeat・業務出力の試行は、すべて新claimへ影響せず拒否されることをテストする。

retryと不正jobの扱い

• retry可能な一時障害と、payload不正、未知type、対象消失などの恒久障害を分類する。
• retryは上限、backoff、jitter、available_atを持ち、上限到達時は終端失敗へ遷移する。
• 未知のjob typeやZod parseに失敗したpayloadをwarn後の成功扱いにしない。non-retryable failureとして記録し、親batchの失敗数・終端状態へ反映する。
• dispatch漏れを防ぐため、job typeのschemaとdispatchをdiscriminated unionで結び、網羅性検査を行う。

現状は未知typeがwarnだけでhandlerを抜け、その後workerが完了更新するため、以下を修正対象に含める。

• apps/server/src/application/services/job-dispatch-service.ts:69
• apps/server/src/infrastructure/jobs/job-worker.ts:207

SET LOGGEDの運用条件

ALTER TABLE jobs SET LOGGEDはロックとtable rewriteを伴う運用作業として扱う。

1. migration前にworkerと新規job投入をquiesceする。
2. 対象件数、table/index容量、空きdisk、想定停止時間を記録する。
3. lock_timeoutを設定し、取得できない場合は無期限に待たず中止する。
4. migration後にpg_class.relpersistence = 'p'、件数、index、claim動作を検証する。
5. worker再起動、異常終了、再取得の試験を行う。
6. 実行時間、最大lock時間、rollback境界を記録する。

2026-07-18時点の確認対象は20,774 jobs。#613の最終dump取得より前に本Issueを完了し、最終dumpへLOGGED化されたjobsを含める。

追加の検証・受け入れ条件

• 2つ以上の独立DB connection/workerを使う実PostgreSQL concurrency testがある
• 同じdedupe keyの並行投入でactive rowが1件だけ作成される
• merge対象の並行投入でもpayload要素が欠落しない
• lease失効後に新workerがclaimすると、旧workerのheartbeat・完了・失敗・業務出力が拒否される
• retryable/permanent failure、上限到達、backoff、jitterを検証している
• 未知typeと不正payloadが成功扱いにならず、親batchも終端状態になる
• parent作成とdispatch/child作成のtransaction失敗を注入しても、孤立した非終端parentが残らない
• SET LOGGED後に永続tableであること、再起動後も件数とclaim可能状態が維持されることを確認している
• atomic claim、AI concurrency、source単位直列化を実DB上で回帰確認している