一个零运行时依赖的 Python 库，为可信单机应用提供 SQLite 持久任务队列与 Transactional Outbox。它来自真实求职 Agent 项目的故障边界治理，并被提炼为不含业务代码的通用组件。
- 本地 AI Agent、桌面应用、单机 Web 服务；
- 需要“进程重启后任务仍在”、幂等入队、有限重试的项目；
- 需要发邮件、Webhook、通知等外部副作用，但不希望因崩溃盲目重复发送的项目；
- 小团队原型走向可运维单机服务的阶段。
不适合多节点消费者、高吞吐消息流、跨地域一致性或零信任公网服务。这些场景应使用 PostgreSQL 队列、Redis Streams、RabbitMQ、Kafka 或托管任务系统。
| 故障 | 本库行为 |
|---|---|
| 重复请求 | idempotency_key 唯一约束返回原记录 |
| Worker 崩溃 | lease 过期后任务可被回收 |
| 旧 Worker 恢复并误写 | 每次 claim 使用新的 fencing token |
| 长任务 | 后台 heartbeat 自动续租 |
| 临时错误 | 有上限的指数退避重试 |
| 已发送但完成写库失败 | Outbox 进入 delivery_unknown，禁止自动重发 |
| 运维确认结果 | CLI 明确标记已送达或受控重试 |
当前首版发布在 GitHub Releases，可直接从源码标签安装：
pip install "git+https://github.com/eatdrop/sqlite-durable-workflow.git@v0.2.0"也可以从 Release 下载 wheel 后离线安装：
pip install sqlite_durable_workflow-0.2.0-py3-none-any.whlfrom pathlib import Path
from sqlite_durable_workflow import DurableSQLite, DurableWorker
db = DurableSQLite(Path("./data/workflow.db"))
job = db.enqueue_job(
"analyse_resume",
{"resume_id": "r-42"},
idempotency_key="analyse-resume:r-42:v1",
)
db.enqueue_outbox(
"email",
"student@example.com",
{"subject": "分析完成", "template": "analysis-ready", "resume_id": "r-42"},
idempotency_key=f"analysis-ready:{job['id']}",
)
def analyse(payload):
return {"resume_id": payload["resume_id"], "score": 86}
def send(message):
# 调用邮件/Webhook SDK。若“可能已发送但无法确认”，抛 DeliveryUnknownError。
print(message["topic"], message["destination"], message["payload"])
worker = DurableWorker(db, {"analyse_resume": analyse}, send)
worker.start()
# 应用退出时：
worker.stop()任务消费采用 at-least-once 语义：崩溃可能使同一任务再次执行，因此 handler 必须使用业务唯一键、upsert 或外部服务的幂等键。
with db.transaction() as tx:
tx.execute(
"UPDATE orders SET status = 'paid' WHERE id = ?",
("order-42",),
)
db.enqueue_outbox(
"receipt",
"student@example.com",
{"order_id": "order-42"},
idempotency_key="receipt:order-42",
connection=tx,
)事务回滚时，业务状态与 Outbox 记录会一起回滚；调用方不得在该事务块内执行网络副作用。
from sqlite_durable_workflow import (
DeliveryUnknownError,
PermanentTaskError,
RetryWithoutAttemptError,
)
# 参数非法等永久错误：直接失败
raise PermanentTaskError("unsupported file type", code="unsupported_file")
# 已知锁竞争：稍后执行，不消耗失败预算
raise RetryWithoutAttemptError("GPU busy", retry_after_seconds=30)
# 外部服务可能已执行副作用：隔离等待人工核对，绝不盲目重试
raise DeliveryUnknownError("connection closed after provider accepted request")sqlite-durable-workflow --database ./data/workflow.db stats
sqlite-durable-workflow --database ./data/workflow.db health
sqlite-durable-workflow --database ./data/workflow.db backup ./backup/workflow.db
sqlite-durable-workflow --database ./data/workflow.db prune \
--before 2026-06-01T00:00:00Z
sqlite-durable-workflow --database ./data/workflow.db prune \
--before 2026-06-01T00:00:00Z --apply
sqlite-durable-workflow --database ./data/workflow.db outbox list-unknown
sqlite-durable-workflow --database ./data/workflow.db outbox confirm-delivered 12
sqlite-durable-workflow --database ./data/workflow.db outbox retry 12 --delay-seconds 60list-unknown 有意不输出消息 payload，避免终端日志泄露正文或个人信息。
prune 默认仅预览数量，只有显式添加 --apply 才会删除已完成/已失败的终态记录；
不会清理运行中、待重试或 delivery_unknown 记录。backup 使用 SQLite Backup API
和原子替换，默认拒绝覆盖已有文件。
- SQLite 文件默认尝试设置为
0600，父目录为0700；Windows 上需另配 ACL。 - 单进程或单主机上的多个线程/进程可共享数据库，但不支持共享网络文件系统。
- 请备份数据库文件及
-wal/-shm，或在停写后用 SQLite Backup API。 - payload 仍可能包含敏感信息；调用方负责最小化、加密磁盘和数据留存策略。
- migration 在
BEGIN IMMEDIATE内执行，并拒绝打开高于当前代码版本的数据库。 - “exactly once” 对任意外部副作用都无法凭空保证；本库采用显式歧义隔离。
python -m pip install -e ".[dev]"
ruff check .
ruff format --check .
pytest -q
python -m build公开 API 目前为 0.x，小版本可能调整接口；升级前请阅读 CHANGELOG.md。