sql-job-queue

sql-job-queue — SKILL.md

バリアント： 標準 · 使用する場面： このスキルが呼び出されると、SQLAlchemy 2.x で動作する DB ベースの ready-set スケジューラー（またはその関連部分）を構築し、制御を呼び出し元に戻します。

概要

このスキルでは、SQLAlchemy 2.x による DB ベースの ready-set ジョブスケジューラーの構築方法を学びます。これは、自分で管理する薄い tick ループであり、リレーショナルデータベースだけを唯一の永続化基盤として使用します（ブローカーも、第二の信頼できる情報源もありません）。ジョブは依存関係が完了すると実行可能になります。対象は、外部の制限（リソース／レート／予算）によって同時実行数が上限設定される、長時間実行されるステートフルなジョブです。単一マシン上で組み込み可能なライブラリとして動作し、SQLite、PostgreSQL、MySQL間で移植できます。中心となるのは方言ごとのアトミックなリース処理です。PostgreSQL と MySQL 8 以降では SELECT ... FOR UPDATE SKIP LOCKED を、SQLite では BEGIN IMMEDIATE による書き込みロックでの確保を使用します。その上に、クラッシュ後の再開、公平な共有、冪等性のパターンを重ねます。同期処理を優先しており、具体的な実装は同期処理で、非同期処理は簡単な補足として扱います。sqlalchemy の関連スキルが扱うデータレイヤーとロックプリミティブを土台としており、それらを改めて説明するものではありません。

有効化する場面

• ✅ リレーショナル DB に永続化し、SQLite／PostgreSQL／MySQL 間で移植可能な、組み込み可能な単一マシン用のキューを構築する場合。ジョブの準備状態は依存関係によって決まり、各ジョブは上流の依存ジョブが done になると実行可能になります。
• ✅ 1 台のマシン上の 2 つのワーカー（プロセス／スレッド／非同期タスク）が同じジョブをディスパッチしないようにするアトミックな確保処理を組み立てる場合で、方言ごとの分岐が必要なとき。
• ✅ ブローカーなしでキューをクラッシュセーフにする場合。ハートビート、リースの有効期限切れ、古いリースの再確保、ハングしたジョブの実行時間上限を扱います。
• ✅ ジョブグループ間でスロットを重み付きの公平な共有によって割り当てる場合、またはスキャン→順位付け→リース→ディスパッチ→永続化→再確保のtick ループを構築する場合。

次の場合は有効化しないでください：

• SQLAlchemy のデータレイヤー自体が必要な場合 — エンジン／URL、DeclarativeBase／Mapped、セッション、JSON カラム、または with_for_update(skip_locked=...) プリミティブと方言の対応表が必要な場合 → sqlalchemy。このスキルはそのプリミティブを組み合わせて使用するものであり、解説するものではありません。
• スキーマのマイグレーションが必要な場合 → alembic。
• ブローカーが適している場合 — 短時間で完了する fire-and-forget タスク、静的なエンキュー、複数マシンへのファンアウト → Celery／Dramatiq／RabbitMQ（この用途で適さない理由は「アンチパターン」を参照）。
• 時間起動型のスケジューラー（cron のようなもの）が必要な場合 → APScheduler。または、大規模なオーケストレータープラットフォームが必要な場合 → Airflow／Prefect／Dagster。

ワークフロー

ステップ 1：要件の形を確認する — 本当に適切なツールですか？

要件が次の条件に合致する場合に限り、DB ベースの ready-set ループを構築してください。長時間実行されるステートフルなジョブ · 動的な依存関係に基づく準備状態 · 外部の制限によって上限が決まる同時実行数（固定ワーカープールではない） · 単一マシン · 組み込み可能 · DB が唯一の永続状態。短時間で完了する fire-and-forget タスクで、静的なエンキューが可能かつ 2 つ目のデータストアを許容できるなら、ブローカーのほうがシンプルです。これを自作しないでください。適さないツールを選ぶ場合の判断理由はアンチパターンにあります。借用する中核機能（SKIP LOCKED + リース）は Postgres ネイティブのキュー（Procrastinate、PGQueuer）に由来し、薄くて移植性の高いループとして再構築します。

ステップ 2：データモデルを定義する — jobs + job_deps

汎用の jobs 行 1 件と、非巡回 DAG を表す job_deps エッジテーブル 1 つを用意します。プロジェクトに依存しない構成にし、ドメイン固有のカラムは設けません。

from datetime import datetime
from sqlalchemy import ForeignKey, Index, JSON, func, text
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class Job(Base):
    __tablename__ = "jobs"

    id: Mapped[int] = mapped_column(primary_key=True)
    type: Mapped[str]                                   # handler key — which function runs it
    args: Mapped[dict] = mapped_column(JSON)            # input payload
    result: Mapped[dict | None] = mapped_column(JSON)   # output, written atomically with `done`
    state: Mapped[str] = mapped_column(default="ready") # ready|leased|running|done|failed
    parent_id: Mapped[int | None] = mapped_column(ForeignKey("jobs.id"))  # container; leaves run
    group_key: Mapped[str]                              # fair-share dimension
    weight: Mapped[int] = mapped_column(default=1)      # fair-share weight
    lease_owner: Mapped[str | None]                     # worker id holding the claim
    lease_expires: Mapped[datetime | None]             # claim TTL — reclaim past this
    in_flight: Mapped[bool] = mapped_column(default=False)
    heartbeat_at: Mapped[datetime | None]              # last liveness bump
    run_after: Mapped[datetime | None]                 # time gate — not ready until now >= this
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(server_default=func.now(), onupdate=func.now())

    # Partial/filtered index so the per-tick scan stays cheap when the table
    # holds many done/parked rows. SQLite + PG honor the WHERE; MySQL ignores it
    # (no partial indexes) and falls back to a plain (state, run_after) index.
    __table_args__ = (
        Index("ix_jobs_ready", "state", "run_after",
              # text() predicate, not a bare `state` ref — the mapped column isn't
              # reliably resolvable at class-body time for the partial-index WHERE.
              postgresql_where=text("state = 'ready'"),
              sqlite_where=text("state = 'ready'")),
    )

class JobDep(Base):
    __tablename__ = "job_deps"
    job_id: Mapped[int] = mapped_column(ForeignKey("jobs.id"), primary_key=True)
    depends_on_id: Mapped[int] = mapped_column(ForeignKey("jobs.id"), primary_key=True)

• parent_id = コンテナです。 子を持つジョブはコンテナであり、ディスパッチされることはありません。最後の子が完了すると、状態が done／failed に遷移するようロールアップします。実行されるのはリーフのみです。ロールアップはイベント駆動型で、子の完了時に計算されます。ready-set クエリ内では計算しません（このクエリはコンテナを除外するだけなので、ティックごとのスキャンで DAG 全体を走査することはありません）。
• job_deps は readiness DAG です。depends_on_id がそのジョブを指すすべての行が done の場合にのみ、ジョブは依存関係を満たして実行可能になります。
• 部分インデックスにより、スキャンの計算量をテーブル全体ではなく O(ready-set) に抑えられます。部分インデックスのない MySQL では、通常の複合インデックスを使用します。state の条件でも十分に絞り込めます。

ステップ 3：ready-set クエリ — 現時点で実行可能なもの

ジョブが ready-set に含まれるのは、状態が実行可能（ready）で、依存関係がすべて完了しており（job_deps に、未完了の上流ジョブを指す行がない）、リーフのみで（子がなく）、実行時刻の条件を満たし（run_after IS NULL OR run_after <= now）、有効なリースがない（期限切れではないリースがない）場合です。移植性の高い形で記述し、ロックはステップ 4 まで保留します（クエリは候補を選択し、リースが1 つをアトミックに確保します）。

from sqlalchemy import and_, exists, not_, or_, select

def ready_set(now):
    Child = Job.__table__.alias("child")
    Dep = JobDep.__table__
    Up = Job.__table__.alias("up")

    has_child = exists().where(Child.c.parent_id == Job.id)
    has_open_dep = exists().where(and_(
        Dep.c.job_id == Job.id,
        Up.c.id == Dep.c.depends_on_id,
        Up.c.state != "done",
    ))
    return (
        select(Job)
        .where(Job.state == "ready")
        .where(or_(Job.run_after.is_(None), Job.run_after <= now))
        .where(not_(has_child))           # leaf-only — containers roll up, never dispatch
        .where(not_(has_open_dep))        # all deps done
        .where(or_(Job.lease_expires.is_(None), Job.lease_expires <= now))  # no live lease
        .order_by(Job.id)
    )

これは、どの方言でも使える SQLAlchemy です。方言によって異なるのは、ステップ 4 の確保処理だけです。

ステップ 4：アトミックなリース — 方言ごとの完全な分岐を 3 つ実装する（可搬性の要）

これは、可搬性において最も重要なポイントです。主張はアトミックでなければなりません。つまり、1 つのトランザクションで処理可能な行を読み取り、他のワーカーが先に処理する前に lease_owner と lease_expires と in_flight を書き込みます。ロックのプリミティブ（with_for_update(skip_locked=...)）とそのサポート状況は、sqlalchemy の関連セクションに記載されているため、ここでは既知のものとして扱います。このスキルでは、組み合わせ、すなわち取得トランザクション、方言の検出、二重ディスパッチ防止のガードを説明します。方言を一度検出して分岐します。

def claim(engine, worker_id, ttl):
    name = engine.dialect.name          # "postgresql" | "mysql" | "sqlite"
    if name == "postgresql":
        return _claim_pg(engine, worker_id, ttl)
    if name == "mysql":
        return _claim_mysql(engine, worker_id, ttl)
    if name == "sqlite":
        return _claim_sqlite(engine, worker_id, ttl)
    raise RuntimeError(f"unsupported dialect: {name}")

分岐 1 — PostgreSQL：SELECT ... FOR UPDATE SKIP LOCKED。 処理可能な行を取得し、別の取得処理がすでにロックしている行はスキップします（ブロックしません）。その後、同じトランザクション内でその行をリース済みに変更します。PostgreSQL のデフォルトの分離レベルは READ COMMITTED であり、これは取得処理に最適です。各ステートメントが新たにコミットされた行を参照するためです。

from datetime import timedelta
from sqlalchemy import update
from sqlalchemy.orm import Session

def _claim_pg(engine, worker_id, ttl):
    now = _now()
    with Session(engine) as s, s.begin():               # one atomic claim transaction
        row = s.execute(
            ready_set(now).with_for_update(skip_locked=True).limit(1)
        ).scalars().first()
        if row is None:
            return None
        s.execute(
            update(Job).where(Job.id == row.id).values(
                state="leased", lease_owner=worker_id,
                lease_expires=now + timedelta(seconds=ttl),
                in_flight=True, heartbeat_at=now,
            )
        )
        return row.id

FOR UPDATE SKIP LOCKED が二重ディスパッチを防ぐガードになります。並行する 2 つの取得処理は重複しない行をロックするため、同じジョブを選ぶことはありません。

分岐 2 — MySQL 8.0 以降（独自の実装例）。 同じ FOR UPDATE SKIP LOCKED ステートメントを使用しますが、MySQL 固有の注意点があります。

• InnoDB と MySQL 8.0 以降が必要です。 5.7 では SKIP LOCKED は何もせずに処理されるのではなく、構文エラーになります。バージョンを確認してください。
• デフォルトの分離レベルは REPEATABLE READ です（PostgreSQL の READ COMMITTED とは異なります）。REPEATABLE READ では、通常の SELECT はトランザクション開始時点で取得した一貫性のあるスナップショットを読み取ります。そのため、自分のトランザクション開始後に別のトランザクションがコミットした行を見落とす可能性があり、処理待ちのジョブがあるのに取得結果が「空」に見えることがあります。各ステートメントで新しいコミットを参照できるよう、取得トランザクションには READ COMMITTED を設定してください。（FOR UPDATE 自体は最新のコミット済み行を読み取る「カレントリード」ですが、READ COMMITTED に固定すれば、取得トランザクション全体を PostgreSQL の分岐と一貫させられます。）
• SQLAlchemy が出力する共有モードの構文は MySQL に拒否されます。 with_for_update(read=True, skip_locked=True) は、MySQL が拒否する従来の LOCK IN SHARE MODE SKIP LOCKED を出力します（SQLAlchemy issue #10134）。これは MySQL の SQL レベルで無効だからではありません（FOR SHARE SKIP LOCKED は MySQL 8 で有効です）。取得処理には、いずれにしても排他的な FOR UPDATE 形式を使用してください。共有ロックでは排他的な取得を実現できません。

def _claim_mysql(engine, worker_id, ttl):
    now = _now()
    # READ COMMITTED for THIS txn — override MySQL's REPEATABLE READ default so
    # the claim sees rows committed after the txn began (see caveat above).
    with engine.connect().execution_options(isolation_level="READ COMMITTED") as conn:
        with conn.begin(), Session(bind=conn) as s:
            row = s.execute(
                ready_set(now).with_for_update(skip_locked=True).limit(1)  # exclusive — NOT read=True
            ).scalars().first()
            if row is None:
                return None
            s.execute(update(Job).where(Job.id == row.id).values(
                state="leased", lease_owner=worker_id,
                lease_expires=now + timedelta(seconds=ttl),
                in_flight=True, heartbeat_at=now,
            ))
            return row.id

分岐 3 — SQLite：BEGIN IMMEDIATE。 SQLite には行レベルのロックがありません。ロックはデータベース／ファイルレベルであり、with_for_update はそこで何もせずに無視されます（エラーにはなりませんが、ロックもしません）。そのため、2 つのワーカーが同じ処理可能な行を読み取り、両方が取得してしまう可能性があります。対処法は、トランザクション開始時に BEGIN IMMEDIATE を実行して、SQLite の書き込みロックを先に取得することです。これにより DB が書き込み用に予約され、取得処理全体が直列化されます。一度に書き込みロックを保持できるワーカーは 1 つだけで、他のワーカーは SQLITE_BUSY を受け取り、次のティックで再試行します。

from sqlalchemy import text

def _claim_sqlite(engine, worker_id, ttl):
    now = _now()
    with engine.connect() as conn:
        conn.exec_driver_sql("BEGIN IMMEDIATE")          # reserve the write lock NOW
        try:
            row = conn.execute(ready_set(now).limit(1)).first()  # no with_for_update — no-op on SQLite
            if row is None:
                conn.exec_driver_sql("COMMIT")
                return None
            job_id = row.id
            conn.execute(
                text("UPDATE jobs SET state='leased', lease_owner=:o, "
                     "lease_expires=:e, in_flight=1, heartbeat_at=:h WHERE id=:i"),
                {"o": worker_id, "e": now + timedelta(seconds=ttl), "h": now, "i": job_id},
            )
            conn.exec_driver_sql("COMMIT")
            return job_id
        except Exception:
            conn.exec_driver_sql("ROLLBACK")
            raise

3 つの分岐とティックループを示した、より完全なスケジューラーの実装例は references/sql-job-queue-extras.md にあります。

ステップ 5：クラッシュ後の再開 — ハートビート、リースの期限切れ、再取得、ハングしたジョブのタイムアウト

リースによって、キューはブローカーなしでもクラッシュに耐えられます。障害モードは次の 2 つです。

• ワーカーの停止（クラッシュ／強制終了）。ジョブの実行中、ワーカーはリースを更新し、定期的に lease_expires（および heartbeat_at）を先へ延ばします。ワーカーが停止すると更新が止まり、lease_expires は過去の時刻になります。期限切れリースの再取得スイープ（各ティックで実行）により、期限切れリースのジョブが再び実行待ちセットに戻されます。

def reclaim(engine, now):
    with Session(engine) as s, s.begin():
        s.execute(update(Job).where(
            Job.state.in_(("leased", "running")),
            Job.lease_expires < now,
        ).values(state="ready", lease_owner=None, in_flight=False))

• **ワーカーは稼働中だがハングしている（別の障害）。**ワーカーはリースの更新を続けていますが、ジョブ自体が停止しています（ネットワーク呼び出しの停止、無限ループなど）。リースが期限切れにならないため、再取得スイープは実行されません。ディスパッチごとのタイムアウトを設定し、ジョブが許容時間を超えて running 状態にある場合は、タイムアウトさせて失敗または再取得します。これはリースとは独立して行います。タイムアウトしたジョブをエスカレーションまたは保留にするポリシーは、このスキルではなくコンシューマーが決めます。

TTL と実行時間のトレードオフ。リース TTL が短すぎると、稼働中の長時間ジョブが早まって再取得され、実行中にもかかわらず二重にディスパッチされます。長すぎると、ワーカーの停止後、ジョブが長時間にわたって回復できない状態になります。TTL はハートビート間隔より十分に長く、ユーザーが停止に気づくまでの時間より十分に短く設定します。ハートビートがあることで、誤った再取得を起こさずに TTL を短くできます。

ステップ 6：重み付きフェアシェア — 1 つのグループがほかを飢餓状態にしないようにする

実行待ちセットに複数の group_key が含まれる場合、利用可能なスロットは厳格な優先順位ではなく、weight に比例して割り当てます。負荷の高い高ウェイトのグループが、低ウェイトのグループを飢餓状態にしてはいけません。具体的には、free 個のスロットがあり、実行待ちセットが group_key ごとにまとめられている場合、存在する各グループに round(free * weight_g / sum(weight)) 個のスロットを割り当てます（どのグループも完全に飢餓状態にならないよう、各グループに少なくとも 1 個）。その後、端数が最も大きい順に残りのスロットを割り当てます。

def fair_share(ready_rows, free):
    groups = {}
    for r in ready_rows:                                 # bucket the ready-set by group
        groups.setdefault(r.group_key, []).append(r)
    weights = {g: rows[0].weight for g, rows in groups.items()}
    total = sum(weights.values())
    chosen = []
    for g, rows in groups.items():
        n = max(1, round(free * weights[g] / total))
        chosen.extend(rows[:n])
    return chosen[:free]                                 # cap at the slot budget

DAG によって並列実行数は自然に制限されます。フェアシェアが決めるのは、限られたスロットを獲得する実行待ちジョブだけです。

ステップ 7：ティックループ — スキャン → 順位付け → リース → ディスパッチ → 永続化 → 再取得

1 回のティックで全体を再評価します。ジョブを静的にエンキューすることはありません。

def tick(engine, worker_id, handlers, ttl, max_concurrency):
    now = _now()
    reclaim(engine, now)                                          # 1. RECLAIM stale leases
    free = max_concurrency - count_in_flight(engine)             # cap = min(external limits)
    if free <= 0:
        return
    with Session(engine) as s:
        candidates = s.execute(ready_set(now)).scalars().all()  # 2. SCAN ready-set
    for job in fair_share(candidates, free):                     # 3. RANK by fair-share
        job_id = claim(engine, worker_id, ttl)                   # 4. LEASE (atomic, per-dialect)
        if job_id is None:
            break                                                # someone else took them
        dispatch(engine, job_id, handlers, ttl)                  # 5. DISPATCH → 6. PERSIST inside

• 同時実行数の上限 = min(external limits)（リソース／レート／予算）です。ワーカープールのサイズではありません。DAG と上限によって、同時に実行される量が制限されます。
• ティックごとに再評価します。前回のスキャン以降に実行可能になったジョブは、次のスキャンで取り上げられます。
• ステップ 1～6（ハンドラーを実行し、結果をアトミックに書き込む dispatch／persist ステップ）の全体的な流れは references/sql-job-queue-extras.md に記載されています。

ステップ 8：起動方法 — デフォルトはポーリング、LISTEN/NOTIFY は PostgreSQL 専用の拡張

**デフォルト：間隔を制限したポーリング。**短い間隔でスリープし、ティックを実行して繰り返します。すべての方言で動作します。コストは、新たに実行可能になったジョブに対して最大で 1 間隔分の遅延が発生することだけです。応答性を感じられる程度に短く、DB に過度な負荷をかけない程度に長い間隔を設定します。

**PostgreSQL の最適化（任意）：LISTEN/NOTIFY。**ワーカーはチャネルを LISTEN し、プロデューサー（または完了したジョブ）が状態変更時に NOTIFY してループを即座に起動します。ポーリング間隔が終わるのを待たずに済むため、遅延が短くなりますが、PostgreSQL 専用です。SQLite と MySQL には同等の機能がないため、ポーリングを継続します。これはポーリングを基本とした仕組みに重ねる拡張として扱い、通知を取り逃がしてキューが停止しないよう、バックストップとしてポーリングも必ず継続してください。ポータブルなデフォルトとして扱ってはいけません。

ステップ 9：冪等性 — 少なくとも 1 回の実行と保証の限界

キューが保証するのは少なくとも 1 回のディスパッチです。再取得（またはハングしたジョブのタイムアウト）により、以前の試行で一部が実行され、副作用が確定した後に停止したジョブが再ディスパッチされることがあります。この契約をコンシューマーに明示してください。

状態マシンが再入を制御するため、キューで二重に最終確定されることはありません。ready → leased → running → done の順に遷移し、result の書き込みは done への遷移と原子的に確定されます（1 つのトランザクション）。したがって、再取得されたジョブは処理中（leased／running であり、done ではない）だと判別でき、再実行しても done に到達できるのは 1 回だけです。状態の切り替えと done の書き込みは、全体が成功するか、全体が取り消されます。

def finalize(engine, job_id, result):                    # result + done flip = ONE transaction
    with Session(engine) as s, s.begin():
        s.execute(update(Job).where(Job.id == job_id).values(
            state="done", result=result, in_flight=False, lease_owner=None,
        ))

保証の限界： キューはジョブを再取得可能にし、二重に最終確定されることはありませんが、ジョブの副作用は重複排除しません。試行 1 でメールを送信した後に停止した場合、試行 2 でもう一度送信されます。ハンドラーの冪等性はコンシューマーの責任です（冪等性キー、アップサート、「すでに実行したか」の確認など）。このスキルが提供するのは状態マシンによるガードであり、ビジネスレベルの重複排除ではありません。

ステップ 10：非同期処理について（簡潔に）

ロックのセマンティクスは非同期処理でも同じです。FOR UPDATE SKIP LOCKED／BEGIN IMMEDIATE の動作は変わらず、変わるのはセッション API だけです。Session を AsyncSession に、with を async with に置き換え、I/O を await します（AsyncSession／await の仕組みは sqlalchemy の関連スキルで参照でき、ここでは再説明しません）。ティックループ全体、リースの分岐、再取得、最終確定は、そのまま 1 対 1 で置き換えられます。主な経路は同期処理です。非同期処理を選ぶのは、周辺のスタックがすでに非同期の場合に限ってください。

ルール

必須ルール（絶対に違反しないでください）：

• クレームは 1 つの原子的なトランザクションにします。 準備完了状態の行の読み取りと lease_owner／lease_expires／in_flight の書き込みは、単一のトランザクション内で行います。読み取りと更新を 2 つに分けてはいけません。
• 方言に応じてリース処理を分岐します。 FOR UPDATE SKIP LOCKED（PostgreSQL、MySQL 8.0+ InnoDB）と BEGIN IMMEDIATE（SQLite）を使い分けます。engine.dialect.name で判定してください。SQLite では with_for_update は何もせずに処理されるため、これに依存すると競合が発生します。
• MySQL では 8.0+／InnoDB、クレーム時の READ COMMITTED、排他 FOR UPDATE が必要です。 SKIP LOCKED は 5.7 ではエラーになります。SQLAlchemy の with_for_update(read=True, skip_locked=True) は旧式の LOCK IN SHARE MODE SKIP LOCKED を出力し、MySQL で拒否されます（issue #10134）。そのため、排他 FOR UPDATE を使い、クレームのトランザクションではデフォルトの REPEATABLE READ を上書きしてください。
• 最終確定は原子的に行います。 result の書き込みと done への遷移を 1 つのトランザクションにすることで、再取得されたジョブが二重に最終確定されないようにします。
• コンテナは集約し、実行するのはリーフだけです。 準備完了ジョブのクエリでは子を持つジョブを除外します。コンテナの完了はスキャンではなく、イベントによって処理します。

推奨事項（変更可能）：

• デフォルトではポーリングを使います。レイテンシー上必要な場合にのみ PostgreSQL の LISTEN/NOTIFY を追加し、バックストップとしてポーリングも維持します。
• リースの TTL は、ハートビートの間隔より長く、ユーザーが停止を認識するまでの時間より短く保ちます。
• 準備完了状態には (state, run_after) の部分インデックスを推奨します。MySQL では通常の複合インデックスを代わりに使います。
• まずは同期処理を使い、すでに非同期のスタック内でのみ非同期処理を選んでください。

注意点

• SQLite では with_for_update はエラーにならず、黙って何もしません。 ロックされないため、2 つのワーカーが同じジョブを取得する可能性があります。SQLite では BEGIN IMMEDIATE を使用し、SQLite だけでなく、実際の対象エンジンでも検証してください。
• MySQL では 8.0 より前のバージョンで SKIP LOCKED を使うと、何もしないのではなく構文エラーになります。 また、MySQL のデフォルトである REPEATABLE READ によって、ジョブ取得の SELECT が、コミットされたばかりの実行可能な行を見逃すことがあります。ジョブ取得トランザクションでは READ COMMITTED を設定してください。
• SQLAlchemy の with_for_update(read=True, skip_locked=True) は MySQL では拒否されます。 最新の FOR SHARE SKIP LOCKED（有効な MySQL 8 の SQL）ではなく、旧式の LOCK IN SHARE MODE SKIP LOCKED が出力されます（SQLAlchemy の issue #10134）。ジョブ取得には排他形式の FOR UPDATE を使用してください。
• ティックごとにテーブル全体を単純にスキャンしても、スケールしません。 テーブルに done／parked の行が増えると、インデックスのない実行可能行のスキャンですべての行を毎回たどることになります。実行可能状態に対する部分／フィルター付きインデックスがあれば、各ティックの処理量を O(実行可能行数) に保てます。
• 早すぎる再取得は、実行中のジョブを二重ディスパッチします。 ジョブの実際の実行時間より短い TTL を設定し（リースを延長するハートビートもない場合）、再取得の処理がまだ実行中のジョブを実行可能行に戻してしまいます。実行中はハートビートを送り、TTL はハートビート間隔より長く設定してください。
• ハングしていてもワーカーが生存している場合、再取得の処理は実行されません。 リースが更新され続けるためです。ジョブを制限するのは、ディスパッチごとに設定する別個のタイムアウトだけです。
• 少なくとも 1 回の実行では、副作用が繰り返される可能性があります。 ステートマシンが防ぐのは二重の完了処理であり、二重の副作用ではありません。ハンドラーは冪等である必要がありますが、キューによって冪等にすることはできません。
• コンテナーのロールアップ処理はクエリ内ではなく、完了時に行ってください。 実行可能行のスキャン内でロールアップを計算すると、ティックのたびに DAG 全体をたどることになります。子の完了時にイベント駆動で計算してください。

アンチパターン

• この用途で、安易にブローカー（Celery／RabbitMQ／Dramatiq／Taskiq）を使わないでください。 ブローカーは短時間のタスクを想定した確認応答／可視性タイムアウトモデルを使います。可視性タイムアウトを超えて実行される長時間のジョブは、再配信され、二重に実行されます。ブローカーが想定するのは静的なエンキューであり、動的な依存関係 DAG ではありません。並行数も外部の上限ではなく、ワーカープールのサイズで決まります。また、ブローカーを導入すると、第 2 の信頼できる情報源が生まれ、1 台のマシンには重すぎるインフラが追加されます。ブローカーが適しているのは、複数のマシンで短時間のタスクをファンアウトする場合であり、この用途ではありません。
• 安易に APScheduler を使わないでください。 APScheduler は時間トリガー型のスケジューラー（時刻 T に X を実行する、または N ごとに実行する）であり、リース機能を持つ依存関係 DAG のワークキューではありません。複数の方言に対応したストアですが、モデルが適合しません。
• Airflow／Prefect／Dagster を安易に使わないでください。 これらは独自のメタデータ DB を持ち込む、オーケストレーションのための大規模なプラットフォームです。第 2 の信頼できる情報源となり、単一の組み込み可能なマシンに必要な量をはるかに超えるインフラを伴います。
• SQLite が行をロックすると想定して、with_for_update だけでジョブ取得を実装しないでください。 SQLite では行はロックされません。方言を判定し、SQLite では BEGIN IMMEDIATE を使用してください。
• 2 つのトランザクションで読み取り後に更新する実装を、（「こちらのほうが分かりやすい」という理由で）行わないでください。 それは、アトミックなジョブ取得で防ぐ二重ディスパッチの競合状態です。
• SQL 文字列が同じという理由で、MySQL の分岐を PostgreSQL の分岐にまとめないでください。 分離レベルのデフォルト設定や、バージョン／共有ロックモードに関する注意点は実際に異なるため、独立した分岐として示してください。
• LISTEN/NOTIFY を追加した後も、ポーリングをなくさないでください。 通知を取りこぼし、ポーリングによる補完手段もないと、キューが気付かれないまま停止します。

出力

SQLAlchemy 2.x を使った、動作する DB バックエンドの ready-set スケジューラー（または指定された範囲）です。ready-state インデックスを備えた汎用の jobs ＋ job_deps スキーマ、移植性のある ready-set クエリ、方言ごとに分岐するアトミックなリース（PostgreSQL／MySQL 8 以降の FOR UPDATE SKIP LOCKED と SQLite の BEGIN IMMEDIATE）、ハートビート＋リース期限切れの回収＋ハングしたジョブのタイムアウトによるクラッシュ後の再開、group_key に基づく重み付きフェアシェア、スキャン→順位付け→リース→ディスパッチ→永続化→回収のティックループ、ポーリング（オプションで LISTEN/NOTIFY）による起動通知、そしてアトミックな最終確定ガードを備えた少なくとも 1 回の実行を保証する冪等性契約を含みます。この成果物は単一マシンのアプリケーションに組み込まれます。ハンドラー（およびその冪等性）と外部の同時実行数上限は利用側が用意します。

関連項目

• sqlalchemy — このデータレイヤーの基盤です。エンジン／URL、DeclarativeBase／Mapped モデル、セッション、JSON、および with_for_update(skip_locked=...) プリミティブと方言ごとのサポート一覧が含まれます。このスキルは、そのプリミティブを要求ループに組み込むものであり、プリミティブ自体を再解説するものではありません。
• alembic — jobs／job_deps テーブルと部分インデックスのスキーママイグレーションです（SQLite のバッチによる移行・コピーの注意点を含みます）。このスキルは、スキーマがすでに存在することを前提とします。
• pydantic-v2 — アプリケーション境界で args／result の JSON ペイロードを検証するためのものです（ORM のカラム型ではありません）。

段階的な情報開示

• references/sql-job-queue-extras.md — 3 種類のリース分岐、dispatch／ハートビート／最終確定の手順、ポーリングによる実行ループを、実行可能な 1 つのコード例にまとめた、より詳細なエンドツーエンドの最小スケジューラーです。**読み込むタイミング：**本文内のスニペットだけでは、ループ全体を組み立てられない場合。
• references/sources.md — 調査の出典情報（Procrastinate／PGQueuer の先行事例と、SQLAlchemy／Postgres／MySQL のロックに関するドキュメント）です。**読み込むタイミング：**主張の出典を確認するとき、または新たにレビューするとき。

本文の上限

• description は 1,024 文字以下（agentskills.io の上限）。Claude はスキル一覧で description と when_to_use を合わせて 1,536 文字で切り詰めます。
• 本文は約 500 行／5,000 トークン以下。
• 参照ファイルごとに、10,000 トークンを超えた場合は警告、25,000 トークンを超えた場合はエラー。参照ファイルの合計は、25,000 トークンを超えた場合は警告、50,000 トークンを超えた場合はエラー。