【PL/SQL】処理ステータス管理テーブル設計 | コーディングライフスタイル

バッチ処理で対象データを順番に処理するとき、単にprocessed_flagだけを持たせると運用で苦しくなります。途中で落ちたレコードはどう戻すのか、処理中のまま残ったデータをどう検出するのか、再実行してよいデータと手動確認が必要なデータをどう分けるのか。この設計が曖昧だと、障害時に手作業のSQLが増えます。

この記事では、PL/SQLで処理ステータス管理テーブルを設計する方法を整理します。未処理、処理中、完了、失敗、再実行待ちといった状態を管理し、排他制御、リトライ、滞留検知、履歴ログまでを実務で使える形にします。差分処理は差分抽出・増分処理の設計、処理対象日の固定は業務日付・営業日カレンダー設計と組み合わせると効果的です。

この記事で扱うこと

処理ステータス管理が必要になる場面
未処理・処理中・完了・失敗・再実行待ちの状態設計
処理対象テーブルと履歴テーブルのDDL
SELECT FOR UPDATE SKIP LOCKEDによる取り合い防止
成功・失敗・リトライの更新方法
処理中のまま残ったデータの検出
再実行時の戻し方と監査ログ

処理ステータス管理が必要になる場面
ステータスを定義する
処理対象テーブルを設計する
ステータス履歴を残す
処理対象を登録する
処理対象を安全に取得する
取得した対象を処理する
失敗時はFAILEDかRETRY_WAITに分ける
例外処理で状態更新を失わない
処理中のまま残ったデータを戻す
手動再実行できるようにする
ステータス別に後続処理を分ける
やってはいけないステータス管理
設計チェックリスト
まとめ

処理ステータス管理が必要になる場面

処理ステータス管理は、対象データを1件ずつ、または一定件数ずつ処理するバッチで必要になります。たとえば、注文データの集計、メール送信、外部連携、帳票作成、データ補正、エラー再処理などです。一度に全件処理できる単純なSQLなら不要な場合もありますが、失敗行だけ再実行したいなら状態管理が必要です。

対象件数が多い全件を1トランザクションで処理せず、分割して進めます。

失敗行を残したいどのデータが失敗したのか、後から確認できるようにします。

複数プロセスで処理する同じレコードを二重処理しない排他制御が必要です。

再実行がある完了済み、失敗、再実行待ちを区別します。

運用確認がある滞留、失敗、処理中放置をSQLで確認できるようにします。

外部API連携の再送キューはアウトボックスパターンと再送キュー設計で扱っています。この記事では、より汎用的なDB内バッチの処理ステータス管理を扱います。

ステータスを定義する

まず、状態を増やしすぎず、運用で判断できる粒度にします。NEW、RUNNING、SUCCESS、FAILED、RETRY_WAIT、SKIPPED程度に絞ると扱いやすくなります。

NEWまだ処理していない状態です。

RUNNINGどこかのバッチプロセスが処理中の状態です。

SUCCESS正常に処理が終わった状態です。

FAILED自動再実行ではなく、確認が必要な失敗です。

RETRY_WAIT一定時間後に再実行できる一時失敗です。

SKIPPED条件により処理しないと判断した状態です。

単純な成功・失敗だけにすると、調査中、再実行待ち、対象外が混ざります。逆に細かくしすぎると、状態遷移が複雑になり、運用者が判断しづらくなります。

処理対象テーブルを設計する

処理対象テーブルには、元データのキー、処理対象日、ステータス、リトライ回数、ロック情報、エラー内容を持たせます。対象データそのものに直接ステータス列を追加できない場合は、ワークテーブルとして切り出します。

batch-work-item-ddl.sql

CREATE TABLE batch_work_item (
  work_item_id      NUMBER PRIMARY KEY,
  job_name          VARCHAR2(100) NOT NULL,
  source_key        VARCHAR2(300) NOT NULL,
  target_date       DATE NOT NULL,
  status            VARCHAR2(20) DEFAULT 'NEW' NOT NULL,
  priority          NUMBER DEFAULT 100 NOT NULL,
  retry_count       NUMBER DEFAULT 0 NOT NULL,
  max_retry_count   NUMBER DEFAULT 3 NOT NULL,
  next_retry_at     TIMESTAMP,
  locked_by         VARCHAR2(100),
  locked_at         TIMESTAMP,
  started_at        TIMESTAMP,
  finished_at       TIMESTAMP,
  error_code        VARCHAR2(100),
  error_message     VARCHAR2(2000),
  created_at        TIMESTAMP DEFAULT SYSTIMESTAMP NOT NULL,
  updated_at        TIMESTAMP DEFAULT SYSTIMESTAMP NOT NULL,
  CONSTRAINT ck_bwi_status
    CHECK (status IN ('NEW','RUNNING','SUCCESS','FAILED','RETRY_WAIT','SKIPPED')),
  CONSTRAINT uq_bwi_job_source
    UNIQUE (job_name, source_key, target_date)
);

CREATE INDEX ix_bwi_fetch
  ON batch_work_item(job_name, status, target_date, priority, work_item_id);

CREATE INDEX ix_bwi_locked
  ON batch_work_item(status, locked_at);

source_keyには、元テーブルの主キーや複合キーを文字列化して入れます。複数種類の処理を同じテーブルで扱う場合は、job_nameで分けます。同じ対象を二重登録しないために、job_name、source_key、target_dateを一意にしています。

batch-work-item-seq.sql

CREATE SEQUENCE batch_work_item_seq
  START WITH 1
  INCREMENT BY 1
  NOCACHE;

ステータス履歴を残す

現在の状態だけでは、なぜ失敗したのか、誰が再実行したのか、いつ状態が変わったのかを追えません。監査や障害調査が必要な処理では、ステータス履歴テーブルも用意します。

batch-work-status-history-ddl.sql

CREATE TABLE batch_work_status_history (
  history_id       NUMBER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  work_item_id     NUMBER NOT NULL,
  old_status       VARCHAR2(20),
  new_status       VARCHAR2(20) NOT NULL,
  changed_by       VARCHAR2(100),
  changed_at       TIMESTAMP DEFAULT SYSTIMESTAMP NOT NULL,
  message          VARCHAR2(2000)
);

CREATE INDEX ix_bwsh_item
  ON batch_work_status_history(work_item_id, changed_at);

ジョブ全体の履歴はジョブ実行履歴テーブル設計、処理対象ごとの履歴はこのテーブル、というように役割を分けると見通しがよくなります。

処理対象を登録する

処理対象は、差分抽出や業務日付で決めた範囲から登録します。すでに登録済みの対象は一意制約で防げますが、SQL上でもNOT EXISTSを入れておくと意図が分かりやすくなります。

enqueue-work-items.sql

INSERT INTO batch_work_item (
  work_item_id,
  job_name,
  source_key,
  target_date,
  status,
  priority
)
SELECT
  batch_work_item_seq.NEXTVAL,
  'DAILY_ORDER_SUMMARY',
  TO_CHAR(o.order_id),
  :target_date,
  'NEW',
  100
FROM orders o
WHERE o.order_date >= :target_date
  AND o.order_date <  :target_date + 1
  AND NOT EXISTS (
    SELECT 1
    FROM batch_work_item w
    WHERE w.job_name = 'DAILY_ORDER_SUMMARY'
      AND w.source_key = TO_CHAR(o.order_id)
      AND w.target_date = :target_date
  );

target_dateはSQL内でSYSDATEから都度計算せず、バッチ開始時に固定した値を渡します。処理対象日の考え方は業務日付・営業日カレンダー設計と合わせると安定します。

処理対象を安全に取得する

複数プロセスで同じワークテーブルを処理する場合は、同じレコードを取り合わないようにします。OracleではSELECT FOR UPDATE SKIP LOCKEDを使うと、他セッションがロックしている行を避けて取得できます。

claim-work-items.sql

CREATE OR REPLACE PROCEDURE claim_work_items (
  p_job_name    IN VARCHAR2,
  p_target_date IN DATE,
  p_limit       IN NUMBER,
  p_worker_id   IN VARCHAR2
)
IS
  v_count NUMBER := 0;

  CURSOR c_items IS
    SELECT work_item_id, status AS old_status
    FROM batch_work_item
    WHERE job_name = p_job_name
      AND target_date = p_target_date
      AND (
        status = 'NEW'
        OR (status = 'RETRY_WAIT' AND next_retry_at <= SYSTIMESTAMP)
      )
    ORDER BY priority, work_item_id
    FOR UPDATE SKIP LOCKED;
BEGIN
  FOR r IN c_items LOOP
    EXIT WHEN v_count >= p_limit;

    UPDATE batch_work_item
    SET
      status = 'RUNNING',
      locked_by = p_worker_id,
      locked_at = SYSTIMESTAMP,
      started_at = SYSTIMESTAMP,
      updated_at = SYSTIMESTAMP,
      error_code = NULL,
      error_message = NULL
    WHERE CURRENT OF c_items;

    INSERT INTO batch_work_status_history (
      work_item_id,
      old_status,
      new_status,
      changed_by,
      message
    )
    VALUES (
      r.work_item_id,
      r.old_status,
      'RUNNING',
      p_worker_id,
      '処理対象として取得'
    );

    v_count := v_count + 1;
  END LOOP;

  COMMIT;
END;
/

全体の二重起動防止には DBMS_LOCKで二重起動防止する方法も使えます。一方、ワークアイテム単位の取り合い防止には、行ロックとSKIP LOCKEDが向いています。

取得した対象を処理する

RUNNINGにしたレコードを、ワーカー単位で処理します。このとき、取得時に渡したp_worker_idと処理時のworker_idを必ず一致させます。locked_byも条件に入れると、他のワーカーが取得した対象を誤って更新しにくくなります。

process-running-items.sql

DECLARE
  v_worker_id VARCHAR2(100) := 'worker-' || SYS_CONTEXT('USERENV','SESSIONID');
BEGIN
  claim_work_items(
    p_job_name    => 'DAILY_ORDER_SUMMARY',
    p_target_date => :target_date,
    p_limit       => 100,
    p_worker_id   => v_worker_id
  );

  FOR r IN (
    SELECT work_item_id, source_key
    FROM batch_work_item
    WHERE job_name = 'DAILY_ORDER_SUMMARY'
      AND status = 'RUNNING'
      AND locked_by = v_worker_id
    ORDER BY work_item_id
  ) LOOP
    process_order_summary(p_order_id => TO_NUMBER(r.source_key));

    UPDATE batch_work_item
    SET
      status = 'SUCCESS',
      finished_at = SYSTIMESTAMP,
      updated_at = SYSTIMESTAMP
    WHERE work_item_id = r.work_item_id
      AND status = 'RUNNING'
      AND locked_by = v_worker_id;

    INSERT INTO batch_work_status_history (
      work_item_id,
      old_status,
      new_status,
      changed_by,
      message
    )
    VALUES (
      r.work_item_id,
      'RUNNING',
      'SUCCESS',
      v_worker_id,
      '処理完了'
    );

    COMMIT;
  END LOOP;
END;
/

1件ごとにCOMMITするか、一定件数ごとにCOMMITするかは処理内容で決めます。トランザクション境界の考え方は PL/SQLのトランザクション設計パターンと合わせて確認するとよいです。

失敗時はFAILEDかRETRY_WAITに分ける

失敗したらすべてFAILEDにするのではなく、自動再実行できる失敗と、人が確認すべき失敗を分けます。一時的なロック、外部連携タイムアウト、相手先メンテナンスなどはRETRY_WAITにできます。データ不整合や必須項目不足はFAILEDとして止める方が安全です。

mark-work-item-error.sql

CREATE OR REPLACE PROCEDURE mark_work_item_error (
  p_work_item_id IN NUMBER,
  p_worker_id    IN VARCHAR2,
  p_error_code   IN VARCHAR2,
  p_error_message IN VARCHAR2
)
IS
  v_retry_count     NUMBER;
  v_max_retry_count NUMBER;
  v_new_status      VARCHAR2(20);
BEGIN
  SELECT retry_count, max_retry_count
    INTO v_retry_count, v_max_retry_count
    FROM batch_work_item
   WHERE work_item_id = p_work_item_id
     AND status = 'RUNNING'
     AND locked_by = p_worker_id
   FOR UPDATE;

  IF v_retry_count + 1 < v_max_retry_count THEN
    v_new_status := 'RETRY_WAIT';
  ELSE
    v_new_status := 'FAILED';
  END IF;

  UPDATE batch_work_item
  SET
    status = v_new_status,
    retry_count = retry_count + 1,
    next_retry_at = CASE
      WHEN v_new_status = 'RETRY_WAIT' THEN SYSTIMESTAMP + INTERVAL '10' MINUTE
    END,
    finished_at = CASE
      WHEN v_new_status = 'FAILED' THEN SYSTIMESTAMP
    END,
    error_code = p_error_code,
    error_message = SUBSTR(p_error_message, 1, 2000),
    updated_at = SYSTIMESTAMP
  WHERE work_item_id = p_work_item_id;

  INSERT INTO batch_work_status_history (
    work_item_id,
    old_status,
    new_status,
    changed_by,
    message
  )
  VALUES (
    p_work_item_id,
    'RUNNING',
    v_new_status,
    p_worker_id,
    SUBSTR(p_error_code || ': ' || p_error_message, 1, 2000)
  );

  COMMIT;
END;
/

バルクDMLの失敗行を個別に扱う場合は FORALL SAVE EXCEPTIONSで失敗行を個別ログ化する方法、DMLエラーを表に逃がす場合は DBMS_ERRLOGとLOG ERRORSの使い方も選択肢になります。

例外処理で状態更新を失わない

処理本体で例外が起きた場合、ROLLBACKするとステータス更新まで戻ることがあります。業務更新を戻したうえで、失敗ステータスだけは残すように、例外処理の流れを明確にします。

handle-error-with-status.sql

DECLARE
  v_worker_id VARCHAR2(100) := 'worker-' || SYS_CONTEXT('USERENV','SESSIONID');
BEGIN
  FOR r IN (
    SELECT work_item_id, source_key
    FROM batch_work_item
    WHERE status = 'RUNNING'
      AND locked_by = v_worker_id
  ) LOOP
    BEGIN
      SAVEPOINT before_item;

      process_order_summary(p_order_id => TO_NUMBER(r.source_key));

      UPDATE batch_work_item
      SET status = 'SUCCESS',
          finished_at = SYSTIMESTAMP,
          updated_at = SYSTIMESTAMP
      WHERE work_item_id = r.work_item_id;

      COMMIT;
    EXCEPTION
      WHEN OTHERS THEN
        ROLLBACK TO before_item;
        mark_work_item_error(
          p_work_item_id  => r.work_item_id,
          p_worker_id     => v_worker_id,
          p_error_code    => SQLCODE,
          p_error_message => SQLERRM
        );
    END;
  END LOOP;
END;
/

エラーを握りつぶして次へ進むだけでは、失敗した理由が残りません。失敗ステータス、エラーコード、エラーメッセージ、履歴をセットで残すことが重要です。

処理中のまま残ったデータを戻す

プロセスが強制終了すると、RUNNINGのまま残ることがあります。locked_atが一定時間より古いデータを滞留として検出し、再実行待ちに戻します。ただし、本当に長時間処理中のデータまで戻さないように、ジョブごとの想定処理時間を決めておきます。

reset-stale-running-items.sql

INSERT INTO batch_work_status_history (
  work_item_id,
  old_status,
  new_status,
  changed_by,
  message
)
SELECT
  work_item_id,
  'RUNNING',
  'RETRY_WAIT',
  'stale-reset-job',
  '処理中のまま一定時間を超過したため再実行待ちに戻しました。'
FROM batch_work_item
WHERE status = 'RUNNING'
  AND locked_at < SYSTIMESTAMP - INTERVAL '2' HOUR;

UPDATE batch_work_item
SET
  status = 'RETRY_WAIT',
  next_retry_at = SYSTIMESTAMP,
  locked_by = NULL,
  locked_at = NULL,
  started_at = NULL,
  error_code = 'STALE_RUNNING',
  error_message = '処理中のまま一定時間を超過したため再実行待ちに戻しました。',
  updated_at = SYSTIMESTAMP
WHERE status = 'RUNNING'
  AND locked_at < SYSTIMESTAMP - INTERVAL '2' HOUR;

滞留判定は自動で戻すだけでなく、運用監視にも使います。何件残っているか、どのワーカーで止まったか、いつから止まっているかを見られるようにします。

monitor-work-item-status.sql

SELECT
  job_name,
  status,
  COUNT(*) AS item_count,
  MIN(updated_at) AS oldest_updated_at,
  MAX(updated_at) AS newest_updated_at
FROM batch_work_item
WHERE target_date = :target_date
GROUP BY job_name, status
ORDER BY job_name, status;

手動再実行できるようにする

運用では、失敗した一部のデータだけを再実行したいことがあります。手動でUPDATE status = 'NEW'と書き換えるのではなく、再実行用の手順を決めます。履歴を残すことで、誰がいつ再実行対象に戻したかを追えます。

manual-retry-work-item.sql

DECLARE
  v_old_status batch_work_item.status%TYPE;
BEGIN
  SELECT status
    INTO v_old_status
    FROM batch_work_item
   WHERE work_item_id = :work_item_id
     AND status IN ('FAILED','RETRY_WAIT')
   FOR UPDATE;

  UPDATE batch_work_item
SET
  status = 'NEW',
  next_retry_at = NULL,
  locked_by = NULL,
  locked_at = NULL,
  started_at = NULL,
  finished_at = NULL,
  error_code = NULL,
  error_message = NULL,
  updated_at = SYSTIMESTAMP
WHERE work_item_id = :work_item_id
  AND status IN ('FAILED','RETRY_WAIT');

INSERT INTO batch_work_status_history (
  work_item_id,
  old_status,
  new_status,
  changed_by,
  message
)
VALUES (
  :work_item_id,
  v_old_status,
  'NEW',
  :operator_name,
  '手動再実行のため未処理に戻しました。'
);

  COMMIT;
END;
/

本番データ補正後に再実行する場合は本番データ補正スクリプトの作り方と同じく、対象条件と戻し方を明確にしてから実行します。

ステータス別に後続処理を分ける

ステータス管理は、単に処理済みを判定するためだけではありません。後続処理、通知、集計、品質チェックを分けるための判断材料になります。

status-based-next-action.sql

SELECT work_item_id,
       source_key,
       CASE status
         WHEN 'SUCCESS' THEN '集計対象'
         WHEN 'FAILED' THEN '運用確認'
         WHEN 'RETRY_WAIT' THEN '再実行待ち'
         WHEN 'RUNNING' THEN '処理中'
         WHEN 'SKIPPED' THEN '対象外'
         ELSE '未処理'
       END AS next_action
FROM batch_work_item
WHERE job_name = :job_name
  AND target_date = :target_date;

処理完了後の集計検算や不整合検出はデータ品質チェック・整合性検証バッチ設計とつなげると、処理結果の品質を確認できます。