sql-job-queue

sql-job-queue — SKILL.md

变体： 标准 · 使用时机： 调用此技能后，它会在 SQLAlchemy 2.x 上生成一个可运行的、由数据库支持的就绪集调度器（或其相关部分），然后将控制权交还给调用方。

概述

此技能旨在教 Agent 在 SQLAlchemy 2.x 上构建一个由数据库支持的就绪集作业调度器——一个由你掌控的轻量级轮询循环，其中关系数据库是唯一持久化基础（不使用消息代理，也没有第二个事实来源），作业会在其依赖项完成后变为可运行状态。它适用于长时间运行、有状态的作业，这些作业的并发量受外部限制（资源 / 速率 / 预算），运行在单台机器上，作为可嵌入的库使用，并可移植到 SQLite、PostgreSQL 和 MySQL。核心内容是针对不同方言的原子租约——PostgreSQL 和 MySQL 8+ 上的 SELECT ... FOR UPDATE SKIP LOCKED，以及 SQLite 上通过 BEGIN IMMEDIATE 获取写锁的认领方式——以及在此基础上实现的崩溃恢复 / 公平份额 / 幂等模式。它以同步优先：演示流程为同步实现，异步仅作简短补充。它基于 sqlalchemy 同级技能的数据层和锁定原语构建，不会重新讲解它们。

何时启用

• ✅ 构建可嵌入、运行于单台机器的工作队列，其就绪状态由依赖关系驱动（作业构成 DAG，每个作业的上游依赖都处于 done 状态后即可运行），持久化在关系数据库中，并可移植到 SQLite/PostgreSQL/MySQL。
• ✅ 实现原子认领，防止单台机器上的两个工作进程（进程 / 线程 / 异步任务）调度同一个作业，并需要针对不同方言进行分支处理。
• ✅ 构建无需消息代理且可抵御崩溃的队列——包括心跳、租约过期、回收过期租约，以及限制卡住的作业。
• ✅ 在不同作业组之间按加权公平份额分配槽位，或构建扫描→排序→租约→调度→持久化→回收的轮询循环。

不要在以下情况启用：

• 你需要 SQLAlchemy 数据层本身——引擎 / URL、DeclarativeBase/Mapped、会话、JSON 列，或 with_for_update(skip_locked=...) 原语及其方言支持矩阵 → sqlalchemy。此技能会组合使用该原语，不会讲解它。
• 你需要架构迁移 → alembic。
• 使用消息代理更合适——短暂的即发即弃任务、静态入队、多机器扇出 → Celery / Dramatiq / RabbitMQ（有关它们为何不适合此处，请参阅“反模式”）。
• 你需要按时间触发的调度器（类似 cron）→ APScheduler；或者功能繁重的编排平台 → Airflow / Prefect / Dagster。

工作流

步骤 1：确认需求形态——这真的是合适的工具吗？

仅当需求形态匹配时，才构建由数据库支持的就绪集循环：长时间运行且有状态的作业 · 由动态依赖关系驱动的就绪状态 · 并发量受外部限制（而非固定工作进程池）· 单台机器 · 可嵌入 · 数据库是唯一持久化状态。如果任务短暂、即发即弃、静态入队，而且你可以接受使用第二个数据存储，那么消息代理更简单——不要构建此方案。错误工具的分析见反模式；借鉴的核心机制（SKIP LOCKED + 租约）来自 PostgreSQL 原生队列（Procrastinate、PGQueuer）——你要做的是将其移植到轻量且可移植的循环中。

Step 2：定义数据模型 — jobs + job_deps

使用一条通用的 jobs 记录，加上一张用于表示无环 DAG 的 job_deps 边表。保持与项目无关——不包含领域列。

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）。只有叶子任务会运行。汇总是事件驱动的——在子任务完成时计算，而不是在就绪集查询中计算（查询只会筛除容器，因此每次扫描都不会遍历整个 DAG）。
• job_deps 是就绪状态 DAG：只有当所有指向某个任务的记录（depends_on_id）都为 done 时，该任务的依赖才算就绪。
• 部分索引可确保扫描复杂度为 O(就绪集)，而不是 O(表)。在 MySQL 上——没有部分索引——可以接受普通的复合索引；state 谓词仍能有效缩小扫描范围。

Step 3：就绪集查询 — 当前可运行的任务

当任务满足以下条件时，它会进入就绪集：状态可执行（ready）、依赖已完成（没有 job_deps 记录指向尚未 done 的上游任务）、仅限叶子任务（没有子任务）、已到运行时间（run_after IS NULL OR run_after <= now），且没有有效租约（不存在未过期的租约）。使用可移植的方式表达；将加锁操作推迟到 Step 4（查询选出候选任务；租约则以原子方式认领其中一个）。

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。方言之间唯一不同的地方是 Step 4 中的认领操作。

第 4 步：原子租约——三个完整的方言分支（关键所在）

这是可移植性中最关键的一点。这个声明必须是原子的：在任何其他工作进程执行操作之前，一个事务读取一条就绪记录，并写入 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——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 分支保持一致。）
• MySQL 不接受 SQLAlchemy 发出的共享锁模式。 with_for_update(read=True, skip_locked=True) 会生成旧式的 LOCK IN SHARE MODE SKIP LOCKED，MySQL 会拒绝该语句（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 在这里会被静默忽略（不会报错，只是不加锁）。因此，两个工作进程可能读取同一条就绪记录并同时声明它。解决方法是在一开始就获取 SQLite 的写锁，使用 BEGIN IMMEDIATE 在事务开始时预留数据库以供写入。这样整个声明过程就会串行执行——同一时间只有一个工作进程持有写锁；其他进程会收到 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

更完整的端到端调度器示例展示了所有三个分支和轮询循环，位于 references/sql-job-queue-extras.md。

步骤 5：崩溃后恢复——心跳、租约过期、重新认领、卡住任务超时

租约让队列在没有消息代理的情况下也能安全应对崩溃。有两种截然不同的故障模式：

• 工作器已停止（崩溃／被终止）。 任务运行期间，其工作器会续租——定期将 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；正是心跳让 TTL 可以设得较短而不致误重新认领。

步骤 6：加权公平共享——不要让一个组饿死其他组

当就绪集跨越多个 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：轮询循环——扫描 → 排序 → 租约 → 调度 → 持久化 → 重新认领

每个轮次都会重新评估整体情况；任务不会以静态方式入队。

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 的增强方式

默认方式：有界轮询。 短暂休眠后运行一个轮次，然后重复。适用于所有方言；唯一代价是新就绪任务最多会有一个轮询间隔的延迟。轮询间隔应足够短，让系统响应及时；同时也要足够长，避免对数据库造成过大压力。

PostgreSQL 优化（可选）：LISTEN/NOTIFY。 工作器会在一个通道上执行 LISTEN；生产者（或已完成的任务）会在状态变化时发送 NOTIFY，从而立即唤醒循环，而不必等待轮询间隔结束。延迟更低，但仅适用于 Postgres——SQLite 和 MySQL 没有等效机制，仍需使用轮询。应将其作为轮询基准方案之上的增强功能（仍保留轮询作为后备，以免通知丢失导致队列停滞），绝不能将其作为可移植的默认方式。

第 9 步：幂等性——至少一次，以及保证的边界

队列保证至少一次分发：重新认领（或挂起作业超时）可能会重新分发先前已部分执行并在退出前提交副作用的作业。请向使用方明确说明这一约定。

状态机通过门控来防止重新进入，因此队列绝不会重复完成：ready → leased → running → done，其中 result 写入与 done 状态转换在同一事务中原子提交。因此，重新认领的作业可以被明确识别为正在处理中（处于 leased/running，而非 done），重新运行它最多只能让它到达 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,
        ))

保证的边界：队列会让作业可重新认领，并且绝不重复完成，但不会对作业的副作用去重。如果第一次尝试发送了邮件后退出，第二次尝试会再次发送。处理程序的幂等性由使用方负责（幂等键、upsert、“我是否已经执行过？”检查）。此技能提供状态机防护，而非业务层面的去重。

第 10 步：异步补充说明（简短）

在异步模式下，锁定语义完全相同——FOR UPDATE SKIP LOCKED / BEGIN IMMEDIATE 的行为不变；只有会话 API 会改变。将 Session 换为 AsyncSession，with 换为 async with，并对 I/O 使用 await（AsyncSession / await 的机制属于 sqlalchemy 同类技能的内容，此处仅引用，不再讲解）。整个轮询循环、租约分支、重新认领和最终完成都可以一一对应地转换。同步是主要路径；仅当周边技术栈已经是异步时才使用异步。

规则

硬性规则（绝不可违反）：

• **认领必须在一个原子事务中完成。**读取就绪行，以及写入 lease_owner/lease_expires/in_flight，必须在同一事务中进行——绝不能跨两个事务先读取再更新。
• 根据方言选择租约分支。FOR UPDATE SKIP LOCKED（PostgreSQL、MySQL 8.0+ InnoDB）与 BEGIN IMMEDIATE（SQLite）二选一。通过 engine.dialect.name 检测。with_for_update 在 SQLite 上会静默失效——在此依赖它会引发竞争条件。
• 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 状态转换在同一事务中，因此重新认领的作业绝不会被重复完成。
• **容器作业只汇总，只有叶子作业会运行。**就绪集查询会过滤掉所有拥有子作业的作业；容器作业的完成由事件驱动，而非通过扫描触发。

偏好（可覆盖）：

• 默认使用轮询；只有在延迟要求较高时才添加 PostgreSQL LISTEN/NOTIFY，并保留轮询作为后备方案。
• 租约 TTL 应高于一次心跳间隔，并低于用户可感知的停滞时长。
• 就绪状态优先使用 (state, run_after) 上的部分索引；在 MySQL 上则退而使用普通复合索引。
• 优先使用同步；仅在已有异步技术栈中才使用异步。

注意事项

• with_for_update 在 SQLite 上会静默失效。 它不会报错，只是不加锁，因此两个工作进程可能领取同一个任务。在 SQLite 上使用 BEGIN IMMEDIATE；并且要在实际目标引擎上验证，不能只在 SQLite 上验证。
• MySQL 8.0 之前的版本会将 SKIP LOCKED 视为语法错误，而不是无操作。此外，MySQL 默认的 REPEATABLE READ 可能导致领取任务时的 SELECT 看不到刚提交的就绪行——请为领取事务设置 READ COMMITTED。
• MySQL 不接受 SQLAlchemy 的 with_for_update(read=True, skip_locked=True)——它会生成旧版 LOCK IN SHARE MODE SKIP LOCKED（SQLAlchemy issue #10134），而不是现代的 FOR SHARE SKIP LOCKED（这是有效的 MySQL 8 SQL）。领取任务时请使用排他性的 FOR UPDATE 形式。
• 每个 tick 都进行朴素的全表扫描无法扩展。 表中充满 done / parked 行后，未建立索引的就绪集扫描每个 tick 都会遍历所有行。针对就绪状态建立部分索引／过滤索引，才能让每个 tick 的复杂度保持为 O(就绪集)。
• 过早回收会导致仍在运行的任务被重复分派。 TTL 短于任务的实际运行时长（且没有心跳来延长 TTL）时，回收扫描会把仍在运行的任务重新放回就绪集。运行期间发送心跳；将 TTL 设为长于心跳间隔。
• 卡住但仍存活的工作进程永远不会触发回收扫描——它的租约会持续续期。只有单独设置每次分派的超时才能限制它。
• 至少一次意味着副作用可能重复发生。 状态机可以阻止重复完成，却不能阻止重复产生副作用。处理程序必须具备幂等性；队列无法替你做到这一点。
• 容器汇总应在完成时执行，而不是放进查询中。 在就绪集扫描中计算汇总会让每个 tick 都遍历 DAG。请在子任务完成时通过事件驱动的方式计算。

反模式

• 这种场景不要选择消息代理（Celery / RabbitMQ / Dramatiq / Taskiq）。 消息代理采用面向短任务的确认／可见性超时模型——长任务若运行时间超过可见性超时，就会被重新投递并重复执行。它们适用于静态入队，而不是动态依赖 DAG。它们的并发度是工作进程池大小，而不是你设置的外部上限。此外，消息代理还会成为第二个事实来源，并为单机引入繁重的基础设施。消息代理适用于多机分发短任务——不适用于此处。
• 不要选择 APScheduler。 它是时间触发型调度器（在时间 T 运行 X／每隔 N 运行一次），而不是带租约的依赖 DAG 工作队列。它支持多种方言的存储，模型也不匹配。
• 不要选择 Airflow / Prefect / Dagster。 这些是繁重的编排平台，会带来自有的元数据数据库——这会成为第二个事实来源，并引入远超单个可嵌入实例所需的基础设施。
• 不要只依赖 with_for_update 来实现任务领取，以为 SQLite 会锁定行——它不会。请检测数据库方言，并在 SQLite 上使用 BEGIN IMMEDIATE。
• 不要在两个事务中先读取再更新（“这样更清晰”）——这正是原子领取要避免的重复分派竞争条件。
• 不要因为 SQL 字符串相同，就把 MySQL 分支合并到 PostgreSQL 分支中——两者的隔离级别默认值以及版本／共享锁模式注意事项确实不同；应将其单独列为一个分支。
• 添加 LISTEN/NOTIFY 后不要移除轮询——如果通知遗漏且没有轮询兜底，队列就会悄然停滞。

输出

一个可正常运行的、基于数据库的就绪集调度器（或所需的功能切片），基于 SQLAlchemy 2.x：通用的 jobs + job_deps 架构及就绪状态索引、可移植的就绪集查询、按方言分支的原子租约（PostgreSQL / MySQL 8+ 使用 FOR UPDATE SKIP LOCKED，SQLite 使用 BEGIN IMMEDIATE）、通过心跳 + 租约过期回收实现崩溃恢复，以及挂起任务超时、基于 group_key 的加权公平份额、扫描→排序→租赁→分发→持久化→回收的 tick 循环、轮询唤醒（可选 LISTEN/NOTIFY），以及带有原子最终确认保护的至少一次幂等性契约。此产物嵌入单机应用；使用方提供处理程序（及其幂等性）和外部并发上限。

相关内容

• 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 — 更完整的端到端最简调度器示例：全部三个租约分支、dispatch/心跳/最终确认步骤，以及作为单个可运行清单的轮询运行循环。在正文片段不足以组装完整循环时加载。
• references/sources.md — 研究来源记录（Procrastinate / PGQueuer 先例 + SQLAlchemy / Postgres / MySQL 锁定文档）。在核实某项主张的来源或进行新一轮审查时加载。

正文篇幅限制

• description ≤ 1,024 个字符（agentskills.io 上限）。Claude 会在技能列表中将 description + when_to_use 合计截断至 1,536 个字符。
• 正文 ≤ 约 500 行 / 5,000 个 token。
• 每个参考文件：超过 10k 个 token 时警告，超过 25k 时报错。参考文件总计：超过 25k 个 token 时警告，超过 50k 时报错。