inboxに置いたJSONを1件ずつ処理する｜flockとmvだけで作る最小ジョブキュー

TL;DR

• AIエージェントや定期ジョブに渡す作業を、jobs/inbox/*.json に置くだけで1件ずつ消化します。
• 排他は flock -n、取り出し・投入は同一ファイルシステム上の mv（rename）。DBもキューサービスも不要です。
• bash 5.2 / util-linux flock 2.41 / jq 1.7 で動作確認済みです。

構成

ワーカー

worker.sh

#!/usr/bin/env bash
# inbox の JSON を1件ずつ処理する最小ワーカー
# 使い方: JOBS_ROOT=jobs JOB_CMD=./process_demo.sh ./worker.sh
set -euo pipefail
ROOT="${JOBS_ROOT:-jobs}"
CMD="${JOB_CMD:-cat}"
mkdir -p "$ROOT"/{tmp,inbox,processing,done,failed}

exec 9>"$ROOT/.lock"
if ! flock -n 9; then
  echo "another worker holds the lock; exit" >&2
  exit 0
fi

shopt -s nullglob
for f in "$ROOT"/inbox/*.json; do
  base=$(basename "$f")
  # 取り出しは rename（同一FSなら原子的）
  mv "$f" "$ROOT/processing/$base"
  if "$CMD" "$ROOT/processing/$base"; then
    mv "$ROOT/processing/$base" "$ROOT/done/$base"
  else
    mv "$ROOT/processing/$base" "$ROOT/failed/$base"
  fi
done

• flock -n はロックを取れなければすぐ戻ります。cron やエージェントから重ねて起動しても、動いているのは1プロセスだけです。
• 取れなかったときの終了コードは 0 にしています。二重起動を「エラー」にせず、静かにスキップするためです。
• if "$CMD" ...; then の形にしてあるので、set -e でも失敗ジョブでワーカー全体が止まりません。失敗分は failed/ に残ります。

処理コマンド（デモ）

process_demo.sh

#!/usr/bin/env bash
set -euo pipefail
f="$1"
if jq -e '.fail == true' "$f" >/dev/null 2>&1; then
  echo "fail: $(jq -c . "$f")" >&2
  exit 1
fi
echo "ok: $(jq -c . "$f")"

本番ではここを、エージェントCLIの呼び出しや投稿スクリプトに差し替えます。

実行例（投入は tmp → mv）

投入側は いったん jobs/tmp/ に書いてから mv で inbox/ へ入れるようにします。> で直接 inbox/ に書くと、書き込み途中の半端なファイルをワーカーが拾うことがあります。mv（同一FS上の rename）なら、inbox に現れた時点で中身は揃っています。

$ mkdir -p jobs/{tmp,inbox,processing,done,failed}
$ printf '%s\n' '{"id":"a","task":"summarize"}' > jobs/tmp/a.json
$ mv jobs/tmp/a.json jobs/inbox/a.json
$ printf '%s\n' '{"id":"b","task":"review","fail":true}' > jobs/tmp/b.json
$ mv jobs/tmp/b.json jobs/inbox/b.json
$ printf '%s\n' '{"id":"c","task":"lint"}' > jobs/tmp/c.json
$ mv jobs/tmp/c.json jobs/inbox/c.json
$ JOBS_ROOT=jobs JOB_CMD=./process_demo.sh ./worker.sh
ok: {"id":"a","task":"summarize"}
fail: {"id":"b","task":"review","fail":true}
ok: {"id":"c","task":"lint"}
$ find jobs -type f ! -name '.lock' | sort
jobs/done/a.json
jobs/done/c.json
jobs/failed/b.json

• 注意: tmp と inbox は同じファイルシステム上に置いてください。別マウントだと mv が copy+delete になり、途中のファイルが見える／消えるタイミングがずれます。processing / done / failed も同様に同一FSにしてください。

ロック中に重ねて起動した場合:

$ # 別プロセスが flock を保持しているとき
$ JOBS_ROOT=jobs JOB_CMD=./process_demo.sh ./worker.sh
another worker holds the lock; exit

inbox が空のときは何もせず exit 0 です。done / failed にあるファイルは再処理しません。

クラッシュ後の復旧（processing → inbox）

ワーカーが processing/ へ移した直後に落ちると、その1件が processing に残ります。同じ .lock を取る復旧スクリプトで、inbox へ戻します。ワーカーがロックを持っているときは拒否するので、動いている処理と取り合いません。

recover.sh

#!/usr/bin/env bash
# processing/ に残ったジョブを inbox/ へ戻す（クラッシュ後の復旧）
# ワーカーが動いていないときだけ実行すること（同じ .lock を使う）
# 使い方: JOBS_ROOT=jobs ./recover.sh
set -euo pipefail
ROOT="${JOBS_ROOT:-jobs}"
mkdir -p "$ROOT"/{tmp,inbox,processing,done,failed}

exec 9>"$ROOT/.lock"
if ! flock -n 9; then
  echo "worker holds the lock; refuse to recover" >&2
  exit 1
fi

shopt -s nullglob
n=0
for f in "$ROOT"/processing/*.json; do
  base=$(basename "$f")
  mv "$f" "$ROOT/inbox/$base"
  echo "requeued: $base"
  n=$((n + 1))
done
echo "recovered $n file(s)"

$ # クラッシュで processing に残った想定
$ printf '%s\n' '{"id":"d","task":"format"}' > jobs/tmp/d.json
$ mv jobs/tmp/d.json jobs/processing/d.json
$ JOBS_ROOT=jobs ./recover.sh
requeued: d.json
recovered 1 file(s)
$ JOBS_ROOT=jobs JOB_CMD=./process_demo.sh ./worker.sh
ok: {"id":"d","task":"format"}

ロック保持中に復旧をかけた場合:

$ JOBS_ROOT=jobs ./recover.sh
worker holds the lock; refuse to recover

（このときの終了コードは 1 です。）

運用上の注意

• mv の原子性は同一ファイルシステム上で成り立ちます。投入（tmp→inbox）も消化（inbox→processing）も、ディレクトリを別マウントに分けないでください。
• 復旧（recover.sh）はワーカー停止中にだけ使います。同じ .lock で排他しているので、ロック中は exit 1 で止まります。
• ジョブの中身は小さな JSON に限定すると、エージェントへの引き渡しが単純になります。

まとめ

• 投入は tmp に書いてから inbox/ へ mv。消化は flock 付きワーカー1本。
• 成功は done/、失敗は failed/。二重起動はロックで静かに回避。
• processing に残った残骸は recover.sh で inbox に戻せる。
• キュー製品を入れる前の、エージェント連携の土台として使えます。

参考

• flock(1) (util-linux): https://man7.org/linux/man-pages/man1/flock.1.html
• jq Manual: https://jqlang.github.io/jq/manual/