tasks-lessons.md 118 KB
异步任务开发
术核 AI · 后端工程课件
01 提交导出任务，返回 202 与可查询状态
为 notes-service 建立导出任务资源，先固定状态转换与查询协议，再接入执行器。
本节目标
- 区分已受理与已完成
- 实现 POST /exports 和 GET /exports/{job_id}
- 拒绝不合法的状态转换
必要概念
202 的含义
202 只说明服务受理了请求，不能表示文件已经产生。提交响应应提供 job_id、当前状态和查询地址；最终结果通过查询接口获取。
状态转换是业务规则
本模块使用 queued、running、succeeded、failed、cancelled。running 才能进入成功或失败；终态不再被旧执行者改写。取消请求与已取消会在后续拆开处理。
先使用受控的内存实验
本课只建立协议，不启动后台执行。记录存在当前进程内，重启会丢失。第 11 课才把状态迁入 PostgreSQL；不能把本课 202 作为可靠交付承诺。
动手实现
1. 准备隔离练习目录
在 notes-service 下创建 tasks_lab 并进入，使用 Python 3.11+ 虚拟环境。第 01–07 课每个文件独立运行。Windows 可用 py -3 -m venv .venv 和 ..venv\Scripts\Activate.ps1；下面是 Linux/WSL2 命令。没有启动任何外部生产服务。
文件：终端 / 当前课时
mkdir tasks_lab
cd tasks_lab
python3 -m venv .venv
source .venv/bin/activate
python -m pip install "fastapi>=0.115,<1" "uvicorn[standard]>=0.30,<1" "httpx>=0.27,<1"
python --version
2. 定义状态机
创建 api_states.py。transition 只用于同进程实验，先检查再修改；错误时不能留下半更新记录。保留原始 note_ids，后续任务执行只接收业务标识或可序列化快照。
文件：api_states.py
from uuid import uuid4
from fastapi import FastAPI, HTTPException, Response
from pydantic import BaseModel, Field
app = FastAPI()
jobs: dict[str, dict] = {}
allowed = {
"queued": {"running", "cancelled"},
"running": {"succeeded", "failed", "cancelled"},
"succeeded": set(), "failed": set(), "cancelled": set(),
}
class ExportIn(BaseModel):
note_ids: list[int] = Field(min_length=1, max_length=100)
def transition(job_id: str, target: str):
row = jobs[job_id]
if target not in allowed[row["status"]]:
raise ValueError(f'{row["status"]} -> {target} is not allowed')
row["status"] = target
3. 增加提交与查询
把这段追加到同一文件。返回 202 和 Location，查询不存在的编号返回 404。当前所有任务保持 queued，不能通过等待假定它会完成。运行 python -m uvicorn api_states:app --reload 后在 /docs 操作。
文件：api_states.py
@app.post("/exports", status_code=202)
async def submit_export(body: ExportIn, response: Response):
job_id = str(uuid4())
jobs[job_id] = {
"id": job_id, "status": "queued",
"note_ids": body.note_ids, "result": None, "error": None,
}
response.headers["Location"] = f"/exports/{job_id}"
return jobs[job_id]
@app.get("/exports/{job_id}")
async def get_export(job_id: str):
if job_id not in jobs:
raise HTTPException(404, "导出任务不存在")
return jobs[job_id]
4. 验证成功与非法转换
单独创建 verify_states.py，执行 python verify_states.py。TestClient 与 transition 在同一进程共享字典，这不是修改另一个已启动服务的状态。记录 202、Location、queued→running→succeeded，以及终态回退失败。
文件：verify_states.py
from fastapi.testclient import TestClient
from api_states import app, jobs, transition
with TestClient(app) as client:
response = client.post("/exports", json={"note_ids": [1, 2]})
assert response.status_code == 202
job_id = response.json()["id"]
assert response.headers["location"] == f"/exports/{job_id}"
transition(job_id, "running")
jobs[job_id]["result"] = {"rows": 2}
transition(job_id, "succeeded")
assert client.get(f"/exports/{job_id}").json()["status"] == "succeeded"
try:
transition(job_id, "running")
except ValueError:
print("terminal transition rejected")
验收结果
- POST 返回 202，Location 能查询到同一 job_id。
- 新提交任务为 queued，未知 job_id 返回 404。
- succeeded→running 被拒绝，原状态仍为 succeeded。
生产约束与故障演练
用户导出 100 条笔记，希望关闭页面后仍能查询结果。
实现决策
- job_id 是独立资源标识，不能用请求连接标识代替。
- 正式受理需要持久化输入与任务记录，本课内存接口不能承担该承诺。
- 结果保存引用和校验信息，不把大文件塞进查询响应。
隔离环境演练
只在本地提交一条 queued 任务后重启服务，记录查询变成404，说明进程内存的丢失边界。
恢复验收
- 接口能区分受理、运行与终态。
- 非法终态回退不能覆盖结果。
- 文档明确本课没有自动执行和持久化。
常见错误
客户端把 202 当成下载成功
原因：受理与执行完成混在一个判断中。
处理：仅在查询得到 succeeded 且存在结果引用后开放下载。
重启后已受理任务全部消失
原因：状态只存在内存，尚未连接可靠执行系统。
处理：将本课限定为协议实验，后续使用持久状态和 outbox。
扩展练习
增加 GET 查询中的 terminal 布尔字段，验证 queued 为 false、succeeded 为 true。
提示：根据状态计算字段，不另存一份可能不同步的布尔值。
参考答案：
返回记录时增加 terminal=row['status'] in {'succeeded','failed','cancelled'}；不要让客户端传入 terminal。
技术文档
02 用 TaskGroup、队列与超时约束 asyncio
批量读取笔记时同时限制执行中的 I/O 和等待处理的数据，取消时回收全部子任务。
本节目标
- 使用 Python 3.11 的 TaskGroup 管理子任务
- 用有限队列和 Semaphore 限制并发与排队
- 让总超时触发取消并验证 finally 执行
必要概念
并发数与任务数量是两种限制
Semaphore 控制同时进入临界区的协程数量，却不会阻止一次创建十万个 Task。有限 Queue 加固定数量消费者，同时限制等待项和任务对象数量。
TaskGroup 管理一组相关任务
退出 TaskGroup 时等待子任务；子任务出现非取消异常时会取消其余任务，并以 ExceptionGroup 传播。它不会把任务保存到磁盘，也不负责服务重启后恢复。
超时通过取消实现
asyncio.timeout() 超时后在上下文外捕获 TimeoutError。协程在 await 等位置响应取消，finally 适合清理；捕获 CancelledError 后应重新抛出。
动手实现
1. 定义可等待的读取函数
新建 bounded_async.py，使用 sleep 代替笔记 API 的 I/O。把输出保存在字典，以 note_id 建立对应关系；实际外部调用还必须设置连接和读取超时。
文件：bounded_async.py
import asyncio
async def read_note(note_id: int):
await asyncio.sleep(0.1)
return {"id": note_id, "title": f"note-{note_id}"}
2. 建立固定消费者与有限缓冲
追加函数。四个消费者最多有两个同时读取；队列最多保存四项。None 是本实验的结束哨兵，真实输入不使用 None。task_done 必须放在 finally，失败也要结清已取出的项。
文件：bounded_async.py
async def collect_notes(note_ids):
queue = asyncio.Queue(maxsize=4)
slots = asyncio.Semaphore(2)
results = {}
async def consume():
try:
while True:
note_id = await queue.get()
try:
if note_id is None:
return
async with slots:
results[note_id] = await read_note(note_id)
finally:
queue.task_done()
finally:
print("consumer closed")
async with asyncio.TaskGroup() as group:
for _ in range(4):
group.create_task(consume())
for note_id in note_ids:
await queue.put(note_id)
for _ in range(4):
await queue.put(None)
await queue.join()
return results
3. 给整个批次设置截止时间
追加入口，执行 python bounded_async.py。正常预算 3 秒足够处理 20 条；把预算改成 0.15 秒复现取消。TimeoutError 在 timeout 上下文外处理，不能返回一个看似完整的部分结果。
文件：bounded_async.py
async def main():
try:
async with asyncio.timeout(3):
rows = await collect_notes(range(20))
print("completed", len(rows))
except TimeoutError:
print("batch timed out")
if __name__ == "__main__":
asyncio.run(main())
4. 复现单条错误与协作取消
临时在 read_note 中加入下面两行，再次执行。TaskGroup 失败应停止整批，其余消费者均输出清理日志。随后恢复函数；如果产品允许部分成功，应另行定义逐条错误结果，不能默默吞掉异常。
文件：bounded_async.py
# 临时插入 read_note 的 await 之前：
# if note_id == 3:
# raise ValueError("invalid note for drill")
# 如果明确需要观察组合异常，可在入口外围写：
# try:
# asyncio.run(main())
# except* ValueError as group:
# print("batch failed", len(group.exceptions))
验收结果
- 正常分支返回20条对应编号数据，只有固定四个消费者。
- 0.15秒总预算触发 batch timed out，四个消费者执行清理。
- 单条 ValueError 导致整批失败，未返回伪造的完整结果。
生产约束与故障演练
一次导出需要读取大量笔记，不能耗尽连接池和内存。
实现决策
- 队列上限、消费者数和I/O并发数分别配置。
- 整批失败与部分成功必须由接口契约决定。
- 总截止时间与每次I/O超时同时存在，超时不代表外部请求未生效。
隔离环境演练
在本地第3条输入抛出异常，再用0.15秒预算运行，记录两种情况下消费者关闭与结果行为。
恢复验收
- 并发峰值≤2、排队容量≤4。
- 异常和超时都没有遗留未等待任务。
- 不会把部分数据标记为完整导出。
常见错误
用了Semaphore仍消耗大量内存
原因：先为全部笔记创建任务，再限制内部I/O。
处理：改为固定消费者和有界Queue，生产端await put自然等待。
超时后批次仍持续运行
原因：子协程吞掉CancelledError或长期没有await。
处理：清理后重新抛出取消，阻塞I/O移到下一课的合适执行器。
扩展练习
给 read_note 加运行中计数与峰值，验证最大同时读取数不超过2。
提示：计数增加、减少包围真实I/O，减少放finally。
参考答案：
在Semaphore内部增加计数并更新峰值；try/finally中减少。正常与超时路径峰值均≤2，结束后运行中计数为0。
技术文档
03 区分 BackgroundTasks 与 create_task 的生命周期
在响应后执行一个短任务，观察进程退出、异常回收和重启丢失。
本节目标
- 使用 FastAPI BackgroundTasks 执行响应后工作
- 保存 create_task 的强引用并获取异常
- 在 lifespan 中停止接受任务并等待有界清理
必要概念
BackgroundTasks 仍属于当前应用进程
任务在响应发出后运行，不是持久队列。执行异常无法把已发送的202改成500；重要结果必须通过任务状态和日志记录。
create_task 需要显式管理
保存强引用，读取完成异常，退出时决定等待或取消。只调用create_task不会自动得到重试、限流或跨进程恢复。
进程生命周期不是业务完成保证
正常关闭可以给任务清理时间；强制终止、崩溃和机器断电不会执行finally。会影响用户资产的任务应进入后续持久队列方案。
动手实现
1. 运行响应后的小任务
新建 api_background.py，复用第01课的app和状态。这里新增独立路径，避免覆盖原来只受理的/exports。用 python -m uvicorn api_background:app 启动；此实验不要开reload以免编辑时无意重启。
文件：api_background.py
import asyncio
from uuid import uuid4
from fastapi import BackgroundTasks, Response
from api_states import app, jobs, transition, ExportIn
async def local_export(job_id: str):
transition(job_id, "running")
try:
await asyncio.sleep(1)
jobs[job_id]["result"] = {"rows": len(jobs[job_id]["note_ids"])}
transition(job_id, "succeeded")
except asyncio.CancelledError:
transition(job_id, "cancelled")
raise
except Exception as exc:
jobs[job_id]["error"] = type(exc).__name__
transition(job_id, "failed")
@app.post("/exports/background", status_code=202)
async def submit_local(body: ExportIn, background: BackgroundTasks, response: Response):
job_id = str(uuid4())
jobs[job_id] = {"id": job_id, "status": "queued",
"note_ids": body.note_ids, "result": None, "error": None}
background.add_task(local_export, job_id)
response.headers["Location"] = f"/exports/{job_id}"
return jobs[job_id]
2. 追踪自行创建的Task
新建 task_registry.py。done回调既移除强引用，也获取异常；不要在回调里再次抛出未处理异常。这个组件尚未暴露提交路由，下一步只在独立生命周期实验里使用。
文件：task_registry.py
import asyncio
import logging
running: set[asyncio.Task] = set()
def observe(task):
running.discard(task)
if task.cancelled():
return
error = task.exception()
if error is not None:
logging.error("local task failed: %r", error)
def start(coro):
task = asyncio.create_task(coro)
running.add(task)
task.add_done_callback(observe)
return task
3. 关闭时等待，再协作取消
新建 registry_demo.py独立运行。先给3秒自然完成窗口，剩余任务发取消并等待清理。此例子仅含遵守取消的sleep；不能对任意阻塞库宣称3秒后必定退出。
文件：registry_demo.py
import asyncio
from task_registry import start, running
async def work():
try:
await asyncio.sleep(30)
finally:
print("cleanup")
async def main():
start(work())
await asyncio.sleep(0.1)
pending = set(running)
if pending:
_, pending = await asyncio.wait(pending, timeout=3)
for task in pending:
task.cancel()
await asyncio.gather(*pending, return_exceptions=True)
print("remaining", len(running))
if __name__ == "__main__":
asyncio.run(main())
4. 记录正常停止和崩溃的差异
提交 /exports/background 并轮询第01课查询路径，观察running到succeeded。再将sleep临时改为30秒，仅在本地终止该实验进程并重启；原job_id查询404。比较正常关闭可能等待与强制退出丢失，随后恢复1秒。
文件：终端 / 当前课时
POST /exports/background
Content-Type: application/json
{"note_ids":[1,2]}
# 保存202响应中的id，随后查询：
GET /exports/{job_id}
验收结果
- 短任务在响应后运行，查询最终为succeeded。
- registry_demo输出cleanup，且未保留运行中的Task。
- 重启使内存任务消失，实验记录不把它解释为可自动恢复。
生产约束与故障演练
接口响应后写一条可丢失的诊断信息，与必须完成的用户导出同时存在。
实现决策
- 短且允许丢失的工作可留在应用进程。
- 用户导出不能只依赖BackgroundTasks或create_task。
- 退出先关闭入口，再给任务有界清理时间，最终保证由持久系统承担。
隔离环境演练
本地延长任务后终止应用，重启查询旧id，验证不存在自动恢复机制。
恢复验收
- Task引用和异常都被回收。
- 失败状态与HTTP受理状态分开呈现。
- 没有将应用进程任务描述为持久队列。
常见错误
返回202后异常只出现在终端
原因：响应已发送，后台工作不再控制这次HTTP状态。
处理：将异常转为任务failed并记录关联job_id。
Task exception was never retrieved
原因：create_task任务失败但无人await或读取异常。
处理：保存引用并获取完成异常，或对相关工作使用TaskGroup。
扩展练习
让work在0.1秒后抛出ValueError，确认回调记录异常且running恢复为空。
提示：不要用空except隐藏错误。
参考答案：
将sleep后加raise ValueError('drill')；observe记录错误，回调移除任务引用，程序结束时无未获取异常提示。
技术文档
04 用 Lock 与 Queue 管理线程共享状态
两个线程处理笔记导出片段，保证每条任务结清并限制等待队列。
本节目标
- 用Lock保护复合读改写
- 用Queue传递线程间任务
- 用哨兵和join完成可预测退出
必要概念
GIL 不提供业务事务
字典读改写涉及多个步骤，不能用GIL代替业务锁。锁只保护同一进程中的共享状态，不能协调多个Celery进程或多台机器。
Queue 同步与业务状态分开
queue.Queue提供线程安全的put/get，有界容量用于限制待处理项。取出后必须对应task_done，join等待的是计数清零，不是线程自然退出。
锁内只做短操作
耗时读取、网络等待、文件写入放到锁外。不要在持锁期间等待另一个需要相同锁的任务，避免死锁和无意义串行化。
动手实现
1. 定义共享计数和有限队列
创建 threaded_queue.py。每个输入只是一条笔记编号，结果计数属于这个脚本；不要把该锁当成PostgreSQL记录锁。
文件：threaded_queue.py
from queue import Queue
from threading import Lock, Thread
import time
pending = Queue(maxsize=4)
mutex = Lock()
counts = {"done": 0, "failed": 0}
errors = []
2. 实现保证结清的消费者
追加consumer。异常时记录失败，finally中task_done；即使收到结束哨兵也要结清那一次get。模拟耗时发生在锁外，锁中只修改计数。
文件：threaded_queue.py
def consume():
while True:
note_id = pending.get()
try:
if note_id is None:
return
time.sleep(0.05)
if note_id == 7:
raise ValueError("bad note for drill")
with mutex:
counts["done"] += 1
except Exception as exc:
with mutex:
counts["failed"] += 1
errors.append((note_id, type(exc).__name__))
finally:
pending.task_done()
3. 启动、入队与退出
追加并执行脚本。非daemon线程让退出路径可见。先发送全部编号，再发送两个哨兵；队列满时put等待消费者释放容量。
文件：threaded_queue.py
if __name__ == "__main__":
threads = [Thread(target=consume) for _ in range(2)]
for thread in threads:
thread.start()
for note_id in range(20):
pending.put(note_id)
for _ in threads:
pending.put(None)
pending.join()
for thread in threads:
thread.join()
print(counts, errors)
4. 验证失败项不会卡住队列
预期done=19、failed=1、错误编号7。再把失败条件改成两个编号，确认总数仍20。不要用empty()判断所有任务完成：队列已空时消费者仍可能在处理。
文件：threaded_queue.py
# 验收断言可放在所有thread.join之后：
assert counts["done"] + counts["failed"] == 20
assert all(not thread.is_alive() for thread in threads)
assert len(errors) == counts["failed"]
验收结果
- 完成数与失败数合计20，错误记录对应失败编号。
- pending.join能够返回，全部线程退出。
- 模拟耗时发生在mutex保护范围之外。
生产约束与故障演练
同步笔记读取库只能在线程中调用，希望稳定处理小批量导出。
实现决策
- 共享状态采用短临界区；每个线程自行管理不支持共享的外部连接。
- 队列明确容量，过载时上游等待或拒绝。
- 线程失败必须成为记录中的失败项，不能静默丢失。
隔离环境演练
让第7条输入抛异常并把Queue容量缩小到1，验证错误收集和任务结清。
恢复验收
- 所有入队项都有成功、失败或明确取消结论。
- join不依赖empty/qsize的瞬间值。
- 线程锁没有被当成跨进程幂等方案。
常见错误
queue.join永远不返回
原因：异常或哨兵分支漏了task_done。
处理：让每次成功get都在finally中task_done一次。
两个线程耗时与一个相同
原因：持有锁时执行了sleep或阻塞I/O。
处理：只把共享计数更新放在锁里，I/O放在锁外。
扩展练习
把队列容量改成1，并让消费者每条等待0.2秒，观察生产者入队会等待。
提示：在put前后记录perf_counter，不能只记录消费者耗时。
参考答案：
生产者前几次put后会因队列满等待；消费者释放位置后继续。最终仍完成20项且线程正常退出。
技术文档
05 在线程池中调用阻塞 I/O
限制提交数量，为等待设置超时，并验证取消 Future 不会强制终止已经运行的线程。
本节目标
- 使用ThreadPoolExecutor执行同步读取
- 限制正在执行与已提交的工作数量
- 区分等待超时、取消成功与线程实际退出
必要概念
线程池只限制同时执行数
max_workers不会限制任意多次submit累积的待执行Future。调用方仍需限批提交、有界队列或接入限流。
Future超时不等于函数停止
future.result(timeout=...)只限制调用方等待时间。cancel只能取消尚未运行的工作，无法安全强杀运行中的线程。
取消需要协作与I/O超时
给同步库设置超时，长循环定期检查Event。with退出线程池默认等待工作完成，不能靠离开with逃避仍在运行的阻塞函数。
动手实现
1. 建立同步I/O模拟函数
创建 pool_io.py。函数每小段等待后检查停止事件；本例0.1秒是可控制的模拟I/O，真实库须使用其自身timeout参数。
文件：pool_io.py
import time
from threading import Event
from concurrent.futures import ThreadPoolExecutor, TimeoutError
def read_sync(note_id, stop):
for _ in range(20):
if stop.is_set():
return {"id": note_id, "cancelled": True}
time.sleep(0.1)
return {"id": note_id, "cancelled": False}
2. 观察运行中Future不能被取消
追加并执行。只有一个任务提交到一个线程，确保实验没有无限排队。0.2秒时通常已经开始执行，cancel返回False；设置stop后等待实际退出。
文件：pool_io.py
if __name__ == "__main__":
stop = Event()
with ThreadPoolExecutor(max_workers=1) as pool:
future = pool.submit(read_sync, 1, stop)
try:
print(future.result(timeout=0.2))
except TimeoutError:
print("wait timed out; cancel:", future.cancel())
stop.set()
print("worker result:", future.result(timeout=1))
print("pool closed")
3. 按批限制待执行Future
另建 batch_io.py，导入同一同步函数。每次只提交四项，完成后才提交下一批；这会牺牲少量流水线利用率，但不会无限占用内存。不要把本例的循环直接放进async路由阻塞事件循环。
文件：batch_io.py
from concurrent.futures import ThreadPoolExecutor, as_completed
from threading import Event
from pool_io import read_sync
def export_batches(note_ids):
stop = Event()
results = []
with ThreadPoolExecutor(max_workers=2) as pool:
for start in range(0, len(note_ids), 4):
batch = note_ids[start:start + 4]
futures = [pool.submit(read_sync, item, stop) for item in batch]
for future in as_completed(futures):
results.append(future.result())
return results
if __name__ == "__main__":
print(len(export_batches(list(range(8)))))
4. 在异步调用侧保留同样的退出边界
在独立脚本中用asyncio.to_thread调用read_sync，并在超时后设置Event。to_thread使用的底层线程不会因为await被取消就自动消失；业务确认停止必须来自工作函数结束或持久任务状态。
文件：async_thread.py
import asyncio
from threading import Event
from pool_io import read_sync
async def demo():
stop = Event()
try:
async with asyncio.timeout(0.2):
return await asyncio.to_thread(read_sync, 1, stop)
except TimeoutError:
stop.set()
print("stop requested, thread exits cooperatively")
if __name__ == "__main__":
asyncio.run(demo())
验收结果
- 等待超时被捕获，运行中的cancel返回False。
- 协作停止后工作结果为cancelled=True，线程池最终关闭。
- 批处理每批最多4个Future，同时执行最多2个。
生产约束与故障演练
导出调用只有同步接口的存储SDK，接口请求可能先超时。
实现决策
- 同步SDK必须有连接和读写超时。
- 线程数、待提交数量与外部服务额度一起限制。
- 超时之后的外部副作用依靠幂等与核对，不假定取消已经回滚。
隔离环境演练
本地让read_sync运行2秒，0.2秒停止等待，记录cancel返回值和线程实际结束时间。
恢复验收
- 没有声称Future.cancel能强杀线程。
- 线程池退出后没有遗留工作线程。
- 超时状态不被误当成外部操作未执行。
常见错误
接口已返回超时，外部写入却仍发生
原因：调用方取消了等待，没有停止底层线程。
处理：外部操作需要幂等键和明确状态，给库设置超时并使用协作取消。
离开with仍长时间卡住
原因：线程池正在等待没有超时的阻塞调用完成。
处理：先修正I/O本身的时间上限，不把shutdown当强制终止接口。
扩展练习
增加一个尚未运行的Future，验证排队任务cancel可能为True，而正在运行的任务为False。
提示：max_workers=1，先提交一个长任务，再立即提交第二个。
参考答案：
第二个尚未运行时cancel返回True，不能再对它当作成功结果调用；首个仍需stop.set并等待实际退出。
技术文档
06 用进程池处理 CPU 密集导出
把纯 Python 文本统计移入独立进程，明确序列化、spawn 与进程入口要求。
本节目标
- 将CPU函数定义在可导入模块顶层
- 使用spawn和主入口保护启动进程池
- 比较串行与进程处理的真实开销
必要概念
CPU工作与I/O等待不同
在通常带GIL的CPython构建中，多个线程不会让纯PythonCPU循环稳定获得多核并行；某些C扩展会释放GIL，需实际测量。进程池提供独立解释器，代价是启动与数据复制。
参数需要跨进程序列化
传可序列化的文本、编号和结构，不传Request、锁、数据库连接或打开的文件。每个进程需要自己的连接生命周期。
显式spawn避免依赖平台默认值
Python不同版本与平台的默认启动方式不同。本模块明确使用spawn，目标函数放模块顶层，并用if name == 'main'保护启动代码。
动手实现
1. 写一个可导入的CPU函数
新建 cpu_work.py。使用纯Python循环产生可比较的校验结果，不借助网络或数据库。规模可以调小，不能为了展示加速耗尽机器资源。
文件：cpu_work.py
def score_text(text: str, rounds: int) -> int:
total = 0
for _ in range(rounds):
for character in text:
total = (total * 33 + ord(character)) % 1000000007
return total
2. 比较串行与进程池
创建 process_export.py。在终端运行文件，避免直接在交互式Console定义worker函数。先用两个进程、小输入测量；小任务可能更慢，这是有效结果。
文件：process_export.py
import multiprocessing as mp
from concurrent.futures import ProcessPoolExecutor
from time import perf_counter
from cpu_work import score_text
def main():
texts = ["note export " * 100 for _ in range(4)]
started = perf_counter()
serial = [score_text(text, 2000) for text in texts]
print("serial", perf_counter() - started)
started = perf_counter()
with ProcessPoolExecutor(max_workers=2, mp_context=mp.get_context("spawn")) as pool:
parallel = list(pool.map(score_text, texts, [2000] * len(texts)))
print("process", perf_counter() - started)
assert serial == parallel
if __name__ == "__main__":
main()
3. 复现不可序列化与启动错误
把目标函数临时改为main内部的局部函数，观察序列化失败，然后恢复模块顶层定义。不要删除入口保护后在繁忙机器试验递归启动；用代码审查确认创建池只发生在受保护入口。
文件：process_notes_review.py
# 错误结构，只用于辨认，勿替换为实际入口：
# def main():
# def local_score(text):
# return len(text)
# pool.submit(local_score, "note")
# 正确结构：
from cpu_work import score_text
assert callable(score_text)
4. 给进程失败定义批次结论
在独立调用侧捕获BrokenProcessPool并把整批视为失败。Future等待超时也不会自动结束已运行CPU函数；大工作应分块，控制单块长度，避免在API每次请求里创建一整池进程。
文件：verify_process.py
from concurrent.futures.process import BrokenProcessPool
from process_export import main
if __name__ == "__main__":
try:
main()
except BrokenProcessPool:
print("batch failed: worker process lost")
raise
验收结果
- serial与parallel结果一致。
- 脚本使用显式spawn、顶层目标函数和主入口保护。
- 记录实际耗时，不把进程池预设为所有输入都更快。
生产约束与故障演练
笔记导出需要大量纯Python清洗或统计，阻塞API事件循环。
实现决策
- 先测量CPU占比，再选择进程数。
- 传入有限数据块，不共享数据库连接。
- 池在受控生命周期中管理，失败批次进入可恢复任务流程。
隔离环境演练
仅在隔离脚本把目标函数换为局部定义，确认序列化失败可见且不会产出成功结果。
恢复验收
- 结果可与串行版本交叉验证。
- CPU工作没有被描述成async能够自动加速。
- 跨进程输入和启动方式清晰可复现。
常见错误
Can't pickle local object
原因：任务函数是闭包、lambda或局部定义。
处理：把函数移入可导入模块顶层，仅传可序列化参数。
每个API请求创建进程池导致CPU和内存暴涨
原因：并发请求各自放大worker数量。
处理：使用受控的应用级池或后续独立Celery进程，限制整体并发。
扩展练习
把rounds分别设为10和2000，比较小任务与大任务的进程启动占比。
提示：两次保持输入、进程数与计时范围一致。
参考答案：
两个版本均先验证结果相同，再记录耗时。小任务可能明显更慢；只有在任务足够大时，多核收益才可能覆盖序列化与启动成本。
技术文档
07 选择协程、线程、进程或独立任务队列
用同一导出需求比较执行位置，并确认何时需要 RQ、Dramatiq 或 Celery。
本节目标
- 根据等待、计算和持久交付选择机制
- 区分执行器与消息队列
- 为本模块写出Celery选型及不解决的问题
必要概念
执行器回答工作在哪里运行
asyncio适合可等待I/O，线程承接阻塞I/O，进程承接需要隔离或多核的工作。它们本身不提供机器重启后的任务恢复。
任务队列增加独立消费者
RQ围绕Redis/Valkey队列和Python函数，Dramatiq围绕actor与中间件，Celery提供任务路由、重试、调度等能力。不同产品都需要明确版本、broker与故障交付语义。
本模块选择Celery并保留业务状态
主线采用Celery5.x、Redis7作为练习broker与结果缓存，PostgreSQL16保存任务真源。选择工具不能消除重复执行、外部副作用和数据库一致性问题。
动手实现
1. 把四类工作写成可检验的需求
建立decision.md，逐项填写现有notes-service的工作量，而不是只写“异步”。本例行数是练习限制，不是普遍容量建议。
文件：decision.md
| 工作 | 主要约束 | 起点 |
|---|---|---|
| 同一请求并发读20条笔记 | 可等待I/O、请求内完成 | asyncio + 有界队列 |
| 同步SDK上传一个导出文件 | 阻塞I/O、有超时 | 受限线程池 |
| 大批纯Python文本统计 | CPU、数据可分块 | 受限进程池 |
| 关闭API后仍需完成导出 | 持久状态、独立消费者 | 任务队列 + 状态表 |
2. 固定消息边界
先用json.dumps验证任务参数。工作函数接收job_id和小型快照；生产阶段只传job_id，从状态表读取不可变输入。不要把密码、请求对象或数据库连接放入消息。
文件：message_contract.py
import json
from uuid import uuid4
message = {
"job_id": str(uuid4()),
"rows": [{"id": 1, "title": "note", "content": "", "pinned": False}],
}
encoded = json.dumps(message, ensure_ascii=False).encode("utf-8")
assert len(encoded) < 64 * 1024
print(len(encoded))
3. 比较三种队列的实际运行边界
阅读下方官方资料，记录所选版本的broker、worker启动方式、超时、重试和结果保留方式。只安装下一课Celery，不同时引入三套消费者。Celery5.x不把原生Windows作为支持平台；本模块worker统一在Linux或WSL2运行。
文件：decision.md
# 记录到decision.md：
# broker: Redis 7，仅本机隔离练习
# worker: Celery 5.x / Linux 或 WSL2
# durable status: PostgreSQL 16 export_jobs
# delivery: 允许重复，业务处理必须幂等
# result backend: 临时查询缓存，不能替代export_jobs
4. 把迁移条件写进验收
运行前几课的小脚本并记录：进程退出是否丢任务、CPU是否饱和、待处理数量是否有界。若要求跨重启恢复，不以继续增加线程数作为解决方案；从下一课分离producer与worker。
文件：终端 / 当前课时
python bounded_async.py
python pool_io.py
python process_export.py
python message_contract.py
验收结果
- 四类工作都有明确执行位置和适用边界。
- 消息能够JSON编码且没有连接对象或凭据。
- Celery运行平台与状态真源已写入decision.md。
生产约束与故障演练
从单机小导出演进到API与导出worker独立运行。
实现决策
- 工具选择同时考虑运行平台、维护成本和失败恢复。
- 统一JSON消息和业务job_id，降低执行器替换成本。
- 队列不是状态数据库，外部输出还需幂等处理。
隔离环境演练
对比结束本地线程脚本与停止独立worker后的任务存留，用观察结果修订选型表。
恢复验收
- 没有用异步语法替代持久交付设计。
- 平台约束明确，worker不按原生Windows支持说明。
- 迁移的需求来自可测量约束。
常见错误
用了Celery就宣称任务只执行一次
原因：忽略ack、重投和外部副作用窗口。
处理：按允许重复设计，后续用幂等输入与状态CAS约束结果。
把Result TTL当任务生命周期
原因：缓存过期被误读为任务不存在。
处理：查询任务真源使用PostgreSQL，结果缓存只用于有限期辅助诊断。
扩展练习
给“同步SDK耗时3秒且必须跨API重启完成”选择方案。
提示：执行方式和交付方式可组合，并非只能选一个。
参考答案：
独立队列承诺持久受理与恢复；worker内部按SDK性质使用同步调用或受限线程。仍须设置SDK超时、幂等键和持久状态。
技术文档
08 启动 Celery broker、worker 与结果缓存
创建带认证的独立 Redis broker，把小型笔记快照发给 worker，再查询导出结果。
本节目标
- 配置独立broker与result backend
- 启动Linux/WSL2 Celery worker并检查注册任务
- 从producer提交JSON任务并获取结果
必要概念
三个角色分别运行
producer提交消息，broker暂存与转交消息，worker执行函数。result backend保存短期执行结果；它的可用性与保留时间不等于业务记录的完整性。
函数导入路径属于部署契约
worker必须能导入celery_app和export_tasks。装饰器指定稳定任务名，producer与worker使用相同代码和兼容参数。
先验证最小链路
本课发送一条笔记快照并生成本地 JSON 文件，暂不读取 PostgreSQL。第 11 课改为从数据库读取任务快照。命令都在同一 Linux/WSL2 环境运行；宿主连接 broker 6384，Docker Compose 内连接 broker:6379，不能混用两种网络地址。
动手实现
1. 准备 Linux/WSL2 环境与独立 Redis
在 notes-service/tasks_lab 的 Python 3.11+ 虚拟环境执行。Celery worker 使用 Linux 或 WSL2，不以原生 Windows 为支持环境。先通过 docker version 确认当前环境能访问 Docker Engine；Windows 使用 Docker Desktop 的 WSL 集成。本课另建 Redis 7 broker，宿主端口 6384、容器端口 6379，与缓存练习的 6380 分开。端口或 notes-task-broker 容器名已被其他练习占用时，先核对所属用途，不删除未知容器。
文件：终端 / 当前课时
python -m pip install "celery[redis]>=5.5,<6" "psycopg[binary]>=3.2,<4" "python-dotenv>=1,<2"
docker version
docker ps --format 'table {{.Names}}\t{{.Ports}}'
python -m pip freeze > requirements-tasks-lock.txt
2. 创建仅用于本地练习的认证配置
保存 prepare_broker.py，下一步运行。它生成 TASK_BROKER_PASSWORD、带认证的 Celery URL 和 Redis 配置，不打印凭据，也不会覆盖现有文件。broker-local 目录只允许当前宿主用户进入；配置文件本身可由容器内 Redis 用户读取。请在 WSL 的 Linux 文件系统中运行，避免 Windows 挂载盘权限语义差异。把 .env.tasks 与 broker-local/ 加入当前项目 .gitignore；不要提交、分享配置或把 URL 复制进日志。两个 Redis DB 只分离键空间，仍共享容量与可用性。
文件：prepare_broker.py
from pathlib import Path
import secrets
from urllib.parse import quote
folder = Path("broker-local")
env_file = Path(".env.tasks")
if env_file.exists() or folder.exists():
raise SystemExit("local broker configuration exists; reuse it")
folder.mkdir(mode=0o700)
password = secrets.token_urlsafe(24)
encoded = quote(password, safe="")
env_file.write_text(
f"TASK_BROKER_PASSWORD={password}\n"
f"CELERY_BROKER_URL=redis://:{encoded}@127.0.0.1:6384/0\n"
f"CELERY_RESULT_BACKEND=redis://:{encoded}@127.0.0.1:6384/1\n",
encoding="utf-8",
)
env_file.chmod(0o600)
conf = folder / "redis.conf"
conf.write_text(
"bind 0.0.0.0\nprotected-mode yes\nport 6379\n"
f"requirepass {password}\n"
"dir /data\nappendonly yes\nappendfsync everysec\n"
"maxmemory 256mb\nmaxmemory-policy noeviction\n",
encoding="utf-8",
)
conf.chmod(0o644)
ignore = Path(".gitignore")
existing = ignore.read_text(encoding="utf-8") if ignore.exists() else ""
with ignore.open("a", encoding="utf-8") as stream:
for rule in (".env.tasks", "broker-local/"):
if rule not in existing.splitlines():
stream.write("\n" + rule + "\n")
print("local broker files created; credentials were not printed")
3. 启动 broker 并验证认证连接
先运行 python preparebroker.py，再执行以下命令。数据使用独立命名卷；Redis 配置只读挂载，本课不执行 CONFIG REWRITE。宿主映射限制为 127.0.0.1:6384。启动后用 Python 客户端读取 .env.tasks 并 PING，只输出 PONG。若连接仍指向旧端口，打开未设置旧 CELERY* 变量的新终端；load_dotenv 默认保留已注入环境变量。停止后保留容器可用 docker start notes-task-broker 恢复，不重复 docker run。
文件：终端 / 当前课时
python prepare_broker.py
docker run -d --name notes-task-broker \
-p 127.0.0.1:6384:6379 \
--mount type=bind,src="$PWD/broker-local/redis.conf",dst=/usr/local/etc/redis/redis.conf,readonly \
--mount type=volume,src=notes-task-broker-data,dst=/data \
redis:7 redis-server /usr/local/etc/redis/redis.conf
python - <<'PY'
import os
from dotenv import load_dotenv
from redis import Redis
load_dotenv(".env.tasks")
client = Redis.from_url(
os.environ["CELERY_BROKER_URL"],
socket_connect_timeout=2, socket_timeout=2,
)
try:
print("PONG" if client.ping() else "unexpected reply")
finally:
client.close()
PY
4. 创建Celery应用
新建 celery_app.py，每次从 tasks_lab 目录运行时先加载本地 .env.tasks。部署环境已注入的变量优先，不会被文件覆盖。结果保留 1 小时；worker 只接受 JSON，消息内不传凭据。此处尚未启用 late ack，下一课演示交付窗口。
文件：celery_app.py
import os
from celery import Celery
from dotenv import load_dotenv
load_dotenv(".env.tasks")
app = Celery(
"notes_service",
broker=os.environ["CELERY_BROKER_URL"],
backend=os.environ["CELERY_RESULT_BACKEND"],
include=["export_tasks"],
)
app.conf.update(
task_serializer="json", result_serializer="json",
accept_content=["json"], result_expires=3600,
task_default_queue="exports", task_track_started=True,
broker_connection_timeout=3,
)
5. 实现快照导出并启动worker
新建export_tasks.py。校验job_id，文件只写入当前练习目录的artifacts。另一个相同环境终端启动 python -m celery -A celery_app:app worker -l INFO -Q exports --concurrency=2。确认启动列表出现notes.export。
文件：export_tasks.py
import json
from pathlib import Path
from uuid import UUID
from celery_app import app
@app.task(name="notes.export")
def export_notes(job_id: str, rows: list[dict]):
safe_id = str(UUID(job_id))
folder = Path("artifacts")
folder.mkdir(exist_ok=True)
target = folder / f"{safe_id}.json"
target.write_text(
json.dumps(rows, ensure_ascii=False, sort_keys=True),
encoding="utf-8",
)
return {"job_id": safe_id, "path": str(target), "rows": len(rows)}
6. 从独立producer提交并查询
创建produce_export.py，在有相同环境变量的另一终端运行。get仅用于本地命令行验收，不放进FastAPI请求或Celery任务中同步等待。SUCCESS表示此任务执行结果已写缓存，文件仍位于worker本地路径。
文件：produce_export.py
from uuid import uuid4
from export_tasks import export_notes
job_id = str(uuid4())
rows = [{"id": 1, "title": "first", "content": "note body", "pinned": False}]
result = export_notes.apply_async(args=[job_id, rows], task_id=job_id)
print("submitted", result.id)
print(result.get(timeout=15))
print("state", result.state)
验收结果
- 认证客户端 PING 返回 PONG，连接目标为独立 broker 6384；worker 注册列表包含 notes.export。
- producer获得job_id，worker成功后返回rows=1和相对文件路径。
- artifacts内JSON与提交快照一致，API线程没有等待result.get。
生产约束与故障演练
API提交导出后立即响应，独立worker完成文件生成。
实现决策
- API不调用AsyncResult.get阻塞请求。
- 生产文件保存到所有消费者可访问的持久存储；本地路径只用于单机实验。
- broker持久性、结果TTL和任务状态分别配置并验收。
- 任务 broker 与缓存实例分开；本地 AOF everysec 与 noeviction 仍不能消除丢失窗口，后续用持久状态和 outbox 恢复。
隔离环境演练
仅停止本机练习worker，提交一条后恢复，观察受理、等待超时与最终完成的区别。
恢复验收
- producer和worker确实通过独立进程通信。
- 文件内容对应输入，任务ID能追踪。
- 没有把Redis结果缓存描述为状态真源。
常见错误
Received unregistered task
原因：worker未导入模块或producer任务名不一致。
处理：检查include、当前目录与notes.export注册名，更新代码后重启worker。
查询随机ID显示PENDING
原因：结果backend对未知或已过期ID可能没有记录。
处理：PENDING不等于确定在排队；第11课使用状态表区分未知与已受理。
连接拒绝、NOAUTH 或一直连旧端口
原因：宿主与容器网络地址混用、没有加载认证 URL，或旧环境变量覆盖 .env.tasks。
处理：宿主使用 127.0.0.1:6384；同一 Compose 网络内使用 broker:6379。确认当前目录、已注入环境和凭据来源，不把完整 URL 打印到日志。
扩展练习
先停止worker再提交一条，记录producer拿到ID但不能立刻完成，恢复worker后再查询。
提示：使用同一broker与队列，不把get的等待超时当作任务执行失败。
参考答案：
暂停期间消息可留在broker；producer等待可能超时，但恢复worker后任务仍可能成功。随后用相同ID核对结果与文件。
技术文档
09 理解 ack、重投与幂等输出
故意重复投递同一导出任务，验证稳定输入与输出标识如何限制重复副作用。
本节目标
- 找出执行完成与消息确认之间的重复窗口
- 配置late ack并说明worker丢失行为
- 让重复任务产生相同逻辑结果
必要概念
确认时机会改变丢失与重复窗口
early ack可能在执行前确认；late ack在工作完成后确认，更便于恢复未确认消息，但完成副作用后、确认前崩溃仍可能重复。不要把任一配置称为exactly once。
at-least-once有前提
可用且按要求配置的broker、未确认消息恢复和消费者重试共同支持至少一次交付目标。Celery即使late ack，部分worker子进程丢失场景默认仍确认；reject_on_worker_lost改变该行为，也可能制造毒消息循环。
幂等约束逻辑副作用
同一job_id使用不可变快照和稳定输出名称，重复执行产生相同内容。本地原子替换只避免读到半文件，不解决多个版本竞争、断电持久性或跨机器文件一致性。
动手实现
1. 配置确认与可见性时间
在celery_app.py追加配置并重启练习worker。此处最长任务应小于30秒，设置60秒仅便于本地恢复观察；真实配置需覆盖执行上界且考虑恢复等待。Redis的三处visibility_timeout统一。
文件：celery_app.py
app.conf.update(
task_acks_late=True,
task_reject_on_worker_lost=True,
worker_prefetch_multiplier=1,
broker_transport_options={"visibility_timeout": 60, "socket_timeout": 3, "socket_connect_timeout": 3},
result_backend_transport_options={"visibility_timeout": 60},
visibility_timeout=60,
)
2. 用临时文件完成原子替换
用下面整体替换第08课export_tasks.py。每次执行使用自己的临时文件，最终名字只由job_id决定；输入仍是已固定的小快照。异常时清理本次临时文件，不删除其他任务的输出。
文件：export_tasks.py
import hashlib
import json
import os
from pathlib import Path
from uuid import UUID, uuid4
from celery_app import app
@app.task(name="notes.export")
def export_notes(job_id: str, rows: list[dict]):
safe_id = str(UUID(job_id))
payload = json.dumps(rows, ensure_ascii=False, sort_keys=True).encode("utf-8")
folder = Path("artifacts")
folder.mkdir(exist_ok=True)
target = folder / f"{safe_id}.json"
temporary = folder / f"{safe_id}.{uuid4()}.tmp"
try:
temporary.write_bytes(payload)
os.replace(temporary, target)
finally:
temporary.unlink(missing_ok=True)
return {"job_id": safe_id, "path": str(target),
"sha256": hashlib.sha256(payload).hexdigest()}
3. 重复投递同一业务任务
在独立producer提交两次相同job_id和快照，但使用不同Celery消息ID，避免把共享结果缓存覆盖当作去重。应比较逻辑输出路径与摘要；Celery task_id相同本身也不是业务去重保证。
文件：duplicate_export.py
from uuid import uuid4
from export_tasks import export_notes
job_id = str(uuid4())
rows = [{"id": 1, "title": "fixed snapshot", "content": "", "pinned": False}]
first = export_notes.apply_async(args=[job_id, rows], task_id=str(uuid4()))
second = export_notes.apply_async(args=[job_id, rows], task_id=str(uuid4()))
a, b = first.get(timeout=15), second.get(timeout=15)
assert a["path"] == b["path"]
assert a["sha256"] == b["sha256"]
print("same logical output", a)
4. 观察重投窗口并记录限制
在本地任务里临时于os.replace后增加5秒sleep，记录执行worker子进程PID，只终止该隔离进程以模拟完成后未确认。重投时再次进入任务；观察日志、最终文件和摘要，恢复后删除故障代码。不要给永久崩溃消息无限恢复机会，后续用持久attempts和死信限制。
文件：终端 / 当前课时
python -m celery -A celery_app:app inspect active
python -m celery -A celery_app:app inspect reserved
# 只对已确认属于本次隔离任务的worker子进程做故障注入。
# 使用日志与inspect中的任务ID核对，不按模糊进程名批量终止。
验收结果
- 两个不同Celery消息ID携带同一job_id，结果路径和摘要一致。
- 正常读取最终文件得到完整JSON，没有半文件。
- 能够解释late ack的重复窗口和worker丢失时的配置差异。
生产约束与故障演练
导出文件已生成，worker在发送确认前崩溃。
实现决策
- 按可能重复交付设计，每个副作用明确业务键。
- late ack和worker丢失重投策略需与毒消息隔离一起启用。
- 本地原子重命名只解决部分写入观察，数据库中的版本权属在第11课建立。
隔离环境演练
重复消息确定性验收后，再在隔离worker完成输出到ack之间终止执行进程，观察重复进入而非假定只执行一次。
恢复验收
- 重复执行不会产生两个不同逻辑导出资源。
- 输入快照不可变约束被明确记录。
- 失败恢复说明没有宣称绝对不丢失或exactly once。
常见错误
固定task_id后仍看到函数执行两次
原因：消息ID或结果缓存键不提供业务级只执行一次保证。
处理：用不可变输入、稳定业务键和后续数据库CAS保护最终结果。
任务持续崩溃并不断重投
原因：reject_on_worker_lost重排同一毒消息，没有持久尝试预算。
处理：限制持久attempts，把不可恢复输入转失败或死信，修复后受控重放。
扩展练习
给第二次投递同一job_id却不同title，观察原子替换无法保证业务版本正确。
提示：本实验故意违反输入不可变约束，只使用新建隔离job_id。
参考答案：
最终文件取决于最后完成者，不能靠os.replace解决。第11课由状态表固定快照，消息仅传job_id，并用租约token防旧执行者覆盖结果引用。
技术文档
10 为重试设置退避、抖动与总预算
只重试暂时性错误，在每次执行和下一次等待前检查截止时间。
本节目标
- 区分可重试故障与永久输入错误
- 为Celery retry设置次数与指数退避
- 将业务总截止时间跨消息保存
必要概念
重试只针对有机会恢复的失败
连接暂不可用、受控限流等可以重试；输入校验失败、无权访问通常直接失败。重试范围应围绕幂等的最小操作，不应重复整个含不可逆副作用的业务流程。
次数与总时间都要限制
max_retries只限制retry次数，不限制首次排队、每次执行和等待的总耗时。deadline_at随业务任务持久保存；跨worker用明确时间基准，本地耗时测量使用monotonic。
退避与抖动减少同步冲击
等待上限随尝试次数增长，再在区间内随机选择等待时间。长远调度不宜大量使用Redis broker的ETA/countdown，避免消息占用worker内存及可见性时间导致重投。
动手实现
1. 定义可控的暂时故障
新建retry_tasks.py，并把该模块加入celery_app.py的include列表。任务只模拟读取，不写文件，便于观察重试而不重复副作用。fail_until只用于本地演练，不对外开放。
文件：retry_tasks.py
import random
import time
from celery_app import app
class TemporaryReadError(Exception):
pass
def fetch_snapshot(attempt: int, fail_until: int):
if attempt < fail_until:
raise TemporaryReadError("temporary source failure")
return [{"id": 1, "title": "snapshot"}]
2. 加入跨重试截止时间
追加任务。deadline_epoch由producer确定并原样传递。进入执行前和计划下次重试前都检查；每次真实I/O还须设置不超过remaining的超时。手动retry会抛出Retry用于通知Celery，使用raise保持控制流。
文件：retry_tasks.py
@app.task(bind=True, name="notes.retry_demo", max_retries=3)
def retry_export(self, job_id: str, deadline_epoch: float, fail_until: int = 2):
remaining = deadline_epoch - time.time()
if remaining <= 0:
raise RuntimeError("export deadline exceeded")
try:
rows = fetch_snapshot(self.request.retries, fail_until)
return {"job_id": job_id, "rows": len(rows)}
except TemporaryReadError as exc:
if self.request.retries >= self.max_retries:
raise
cap = min(8.0, 2.0 ** self.request.retries)
delay = random.uniform(0, cap)
if time.time() + delay + 0.1 >= deadline_epoch:
raise RuntimeError("insufficient retry budget") from exc
raise self.retry(exc=exc, countdown=delay)
3. 分别验证恢复与预算耗尽
重启worker加载模块；在独立producer运行两次。第一次前两次失败后成功；第二次预算已经过期，应在调用外部读取前失败。get只用于本地命令行，异常时记录类型。
文件：verify_retry.py
import time
from uuid import uuid4
from retry_tasks import retry_export
ok = retry_export.delay(str(uuid4()), time.time() + 30, 2)
print(ok.get(timeout=35))
expired = retry_export.delay(str(uuid4()), time.time() - 1, 0)
try:
expired.get(timeout=10)
except Exception as exc:
print(type(exc).__name__, str(exc))
4. 验证不能恢复的永久错误
临时把fetch_snapshot改成raise ValueError，确认不会进入TemporaryReadError重试分支。再将fail_until设置10，观察最多初次加3次重试。记录开始时间、重试次数、等待区间和终态；恢复实验函数。
文件：retry_tasks.py
# 永久错误实验：
# raise ValueError("invalid export request")
# 次数预算实验：
# result = retry_export.delay(str(uuid4()), time.time() + 60, 10)
# 最多4次执行：首次 + 3次重试。
验收结果
- fail_until=2的任务经历重试后完成。
- 已过deadline的任务不执行读取，直接失败。
- 永久ValueError不重试，最多执行次数为首次加max_retries。
生产约束与故障演练
导出依赖短暂不可用的笔记存储，用户只接受10分钟内完成。
实现决策
- 持久任务记录保存deadline和尝试数，不只依赖消息内的request.retries。
- 每个外部操作配置阶段超时，预计等待超过预算就失败。
- 任务副作用保持幂等，避免重试重复发布或重复扣费。
隔离环境演练
对本地函数分别注入可恢复错误、永久错误、过期deadline，记录三种重试轨迹。
恢复验收
- 重试有次数上限和总时间上限。
- 不可恢复错误不会无限排队。
- 已超业务时限的任务不会继续消耗下游。
常见错误
每次重试都重置30秒deadline
原因：总预算实际永远不会耗尽。
处理：deadline随原始任务保存，重试传递原值，不在worker重新计算。
对所有Exception自动重试
原因：语法、校验或权限错误变成无用流量。
处理：明确可重试异常集合，永久故障直接进入失败处理。
扩展练习
把预算设为0.05秒，说明为什么次数预算尚余也可能直接失败。
提示：执行机会必须同时满足次数和时间条件。
参考答案：
remaining已不足下一次等待加最小操作时间，因此应结束；剩余retry次数不是必须消耗的额度。
技术文档
11 用 PostgreSQL 保存任务状态与租约
固定导出快照，用持久领取时间与租约控制执行，阻止过期执行者发布结果。
本节目标
- 建立export_jobs持久状态表
- 使用CAS、租约与attempts约束领取
- 将耗时文件处理放在数据库事务之外
必要概念
状态表是用户查询真源
PostgreSQL保存输入快照、状态、尝试数、截止时间和结果引用。Redis结果过期不改变业务任务是否存在。此后Celery消息只传job_id，不再传第08课rows参数。
领取与完成都要条件更新
领取只允许queued且next_attempt_at已到的任务，一次UPDATE原子生成lease_token并增加attempts。完成同时检查running、token、租约、deadline和取消标志。过期running不能直接被消息抢走，必须经过第15课恢复扫描重新排队并设置退避。
短事务包围状态转换
读取快照和领取状态分别在短事务中完成，文件生成放在事务外。不要持有数据库行锁等待长await、网络上传或CPU计算。租约是有时限的权属，不是自动杀死旧worker。
动手实现
1. 建立隔离状态表
在已有本机PostgreSQL16的notes_lab库执行SQL，账户app_user，端口5433。DATABASE_URL由终端环境设置，密码使用本地练习凭据。notes表需已有id/title/content/pinned；本课不覆盖原笔记表。 next_attempt_at决定最早可领取时间，last_dispatched_at记录最新一次被publisher确认的派发时间；这不等于worker已经领取或消息已经永久持久化。
文件：schema_jobs.sql
CREATE TABLE IF NOT EXISTS export_jobs (
id uuid PRIMARY KEY,
status text NOT NULL DEFAULT 'queued'
CHECK (status IN ('queued','running','succeeded','failed','cancelled')),
payload jsonb NOT NULL,
attempts integer NOT NULL DEFAULT 0,
lease_token uuid,
lease_until timestamptz,
next_attempt_at timestamptz NOT NULL DEFAULT now(),
last_dispatched_at timestamptz,
cancel_requested boolean NOT NULL DEFAULT false,
deadline_at timestamptz NOT NULL DEFAULT now() + interval '10 minutes',
result_ref text,
error_code text,
created_at timestamptz NOT NULL DEFAULT now(),
updated_at timestamptz NOT NULL DEFAULT now()
);
CREATE INDEX IF NOT EXISTS export_jobs_status_created
ON export_jobs(status, created_at);
-- 兼容已按旧版课件建立的本机练习表：
ALTER TABLE export_jobs ADD COLUMN IF NOT EXISTS next_attempt_at timestamptz NOT NULL DEFAULT now();
ALTER TABLE export_jobs ADD COLUMN IF NOT EXISTS last_dispatched_at timestamptz;
2. 在短事务中固定导出快照
创建state_store.py。设置DATABASE_URL指向postgresql://app_user:本地密码@127.0.0.1:5433/notes_lab；WSL2必须能访问该地址。限制100条，读取后保存JSON快照，后续修改笔记不会改变已受理的导出输入。 connect设置连接超时3秒、单条SQL超时3秒、锁等待1秒，用于本机小数据实验；连接建立、语句与事务总时间不是同一预算。生产根据服务SLO和批次大小调整，维护进程每批之间检查停机信号，不能依赖无限等待的数据库调用。
文件：state_store.py
import os
from uuid import uuid4
import psycopg
from psycopg.rows import dict_row
from psycopg.types.json import Jsonb
def connect():
return psycopg.connect(
os.environ["DATABASE_URL"],row_factory=dict_row,connect_timeout=3,
options="-c statement_timeout=3000 -c lock_timeout=1000",
)
def submit_snapshot(note_ids: list[int]) -> str:
ids = sorted(set(note_ids))
if not ids or len(ids) > 100:
raise ValueError("select 1..100 notes")
job_id = str(uuid4())
with connect() as conn:
rows = conn.execute(
"SELECT id,title,content,pinned FROM notes WHERE id=ANY(%s) ORDER BY id",
(ids,),
).fetchall()
if len(rows) != len(ids):
raise ValueError("one or more notes do not exist")
conn.execute(
"INSERT INTO export_jobs(id,payload) VALUES (%s,%s)",
(job_id, Jsonb(rows)),
)
return job_id
3. 添加领取与完成的CAS
追加两个函数。领取事务提交后才开始工作，返回的token用于完成校验。30秒是本地小文件租约；真实大导出需分块并续租，续租失败立即停止后续发布。 running租约过期后claim仍返回None；第15课扫描器负责queued转换，避免broker重复消息绕过持久退避时间。成功领取和完成清除当前error_code，历史错误由阶段日志保留。
文件：state_store.py
def claim(job_id: str):
token = str(uuid4())
with connect() as conn:
return conn.execute("""
UPDATE export_jobs SET status='running', attempts=attempts+1,
lease_token=%s, lease_until=now()+interval '30 seconds',
error_code=NULL, updated_at=now()
WHERE id=%s AND cancel_requested=false AND attempts<3
AND deadline_at>now()
AND next_attempt_at<=now() AND status='queued'
RETURNING id,payload,lease_token,attempts
""", (token, job_id)).fetchone()
def finish(job_id: str, token: str, result_ref: str) -> bool:
with connect() as conn:
row = conn.execute("""
UPDATE export_jobs SET status='succeeded', result_ref=%s, error_code=NULL, updated_at=now()
WHERE id=%s AND status='running' AND lease_token=%s
AND lease_until>now() AND deadline_at>now() AND cancel_requested=false
RETURNING id
""", (result_ref, job_id, token)).fetchone()
return row is not None
4. 替换worker为只接收job_id
整体替换export_tasks.py并重启worker，停止使用第08–09课携带rows的producer。每次领取写独立token文件，最终有效文件由数据库result_ref决定；旧worker最多留下未引用文件，不覆盖新结果。异常由Celery记录，业务状态暂留租约，恢复扫描在第15课补齐。不要把第10课retry直接套在未定义状态转换上。
文件：export_tasks.py
import json
from pathlib import Path
from celery_app import app
from state_store import claim, finish
@app.task(name="notes.export")
def export_notes(job_id: str):
job = claim(job_id)
if job is None:
return {"job_id": job_id, "claimed": False}
token = str(job["lease_token"])
folder = Path("artifacts")
folder.mkdir(exist_ok=True)
target = folder / f"{job['id']}.{token}.json"
target.write_text(json.dumps(job["payload"], ensure_ascii=False), encoding="utf-8")
published = finish(job_id, token, str(target))
return {"job_id": job_id, "published": published}
5. 从数据库查询并测试旧token失效
使用真实已存在的笔记ID执行submit_snapshot，再export_notes.delay(job_id)。查询只读状态表。当前提交后发送仍有双写窗口，下一课用outbox解决。接下来的专用租约实验不发送Celery消息，通过明确的一条记录模拟恢复转换。
文件：submit_db_export.py
from state_store import submit_snapshot, connect
from export_tasks import export_notes
job_id = submit_snapshot([1])
export_notes.delay(job_id)
with connect() as conn:
print(conn.execute(
"SELECT id,status,attempts,result_ref FROM export_jobs WHERE id=%s",
(job_id,),
).fetchone())
# HTTP查询接口应读取同一记录；无记录返回404。
# 本地故障注入只对刚创建的job_id修改租约，勿扫描修改其他任务。
6. 只对专用任务验证过期与旧令牌
使用已有笔记编号作为输入；下面通过最小编号选取样本，不假定id=1存在。只修改刚创建的job_id。过期running无法自动领取，实验显式模拟第15课的恢复转换；attempts不清零，随后旧token仍不能完成，新token生成真实文件后才发布。
文件：verify_lease.py
import json
from pathlib import Path
from state_store import submit_snapshot, connect, claim, finish
with connect() as conn:
note_id = conn.execute("SELECT min(id) AS id FROM notes").fetchone()["id"]
assert note_id is not None, "create a note first"
job_id = submit_snapshot([note_id])
old = claim(job_id)
assert old is not None and claim(job_id) is None
with connect() as conn:
conn.execute("UPDATE export_jobs SET lease_until=now()-interval '1 second' WHERE id=%s", (job_id,))
assert claim(job_id) is None
assert finish(job_id, str(old["lease_token"]), "unused") is False
with connect() as conn:
conn.execute("""
UPDATE export_jobs SET status='queued',lease_token=NULL,lease_until=NULL,
next_attempt_at=now(),updated_at=now()
WHERE id=%s AND status='running' AND lease_until<now()
""", (job_id,))
new = claim(job_id)
assert new is not None and new["attempts"] == 2
assert finish(job_id, str(old["lease_token"]), "unused") is False
folder = Path("artifacts")
folder.mkdir(exist_ok=True)
target = folder / f"{job_id}.{new['lease_token']}.json"
target.write_text(json.dumps(new["payload"],ensure_ascii=False),encoding="utf-8")
assert finish(job_id,str(new["lease_token"]),str(target)) is True
print("lease ownership verified",job_id)
验收结果
- 任务记录重启后仍存在，payload是提交时固定的笔记快照。
- 同一租约期内重复领取只有一方取得token。
- 旧token无法完成任务，工作期间不持有数据库长事务。
- next_attempt_at未到时所有重复消息都无法领取；过期running由恢复扫描转换后才能再次领取。
生产约束与故障演练
多个worker可能收到同一个导出job，旧worker在暂停后重新运行。
实现决策
- 业务job与Celery消息分开，任务状态保存在PostgreSQL。
- 输入快照固定；多租约输出隔离，CAS只发布当前token结果。
- 授权过滤应复用notes-service当前用户权限，本地单账户练习不能直接部署为多用户接口。
- finish同时检查业务deadline；超时后不能仅凭未过租约发布成功。
- 领取资格来自数据库状态与next_attempt_at，不由broker是否再次交付决定。
隔离环境演练
对专用任务令租约过期，先确认claim和旧token完成均失败，再模拟恢复为queued并重新领取，比较两次token与result_ref。
恢复验收
- 旧执行者不能发布结果。
- 数据库事务不跨耗时文件处理。
- attempts与deadline来自持久记录，不因消息重投重置。
常见错误
旧worker覆盖新worker结果
原因：完成时只按job_id更新，没有检查token或输出路径复用。
处理：文件名带token，结果发布使用状态、token和租约条件CAS。
Redis显示SUCCESS但数据库仍queued或running
原因：消息可能没有领取权，或者完成CAS失败；执行结果不等于业务终态。
处理：用户查询只认export_jobs，记录claimed/published用于诊断。
扩展练习
实现续租函数：仅当token匹配、status=running、当前租约未过期且未取消时延长30秒。
提示：过期租约不能自己复活，需要重新领取获取新token。
参考答案：
UPDATE export_jobs SET lease_until=now()+interval '30 seconds' WHERE id=%s AND lease_token=%s AND status='running' AND lease_until>now() AND deadline_at>now() AND cancel_requested=false RETURNING id；无返回记录即失去权属。
技术文档
12 用事务 outbox 连接状态提交与消息发布
把导出记录与待发布事件写入同一数据库事务，并处理发布成功但确认写回失败的重复。
本节目标
- 复现数据库与broker双写窗口
- 同事务写入export_jobs和outbox事件
- 用短领取租约与CAS确认发布结果
必要概念
两个系统之间存在双写窗口
先写数据库后发送消息，可能留下永不入队的queued任务；先发送后提交数据库，worker可能读不到记录。outbox把业务记录和发送意图放进同一数据库事务。
outbox不提供exactly once发布
发布成功后进程可能在写sent_at之前崩溃，事件会被再次发布。consumer继续依赖第11课的领取和结果CAS，不能因为用了outbox就移除幂等逻辑。
发布期间不持数据库事务
先短事务领取outbox事件，提交后访问broker，最后短事务按claim_token标记sent。租约过期允许其他publisher接手；SKIP LOCKED用于队列式领取，不用于普通业务一致性查询。
动手实现
1. 建立outbox表
在同一notes_lab数据库执行。外键保证job存在；同一个job可以有后续重放事件，因此不对job_id单独加唯一约束。事件ID与业务job_id分开记录。
文件：schema_outbox.sql
CREATE TABLE IF NOT EXISTS export_outbox (
id uuid PRIMARY KEY,
job_id uuid NOT NULL REFERENCES export_jobs(id),
created_at timestamptz NOT NULL DEFAULT now(),
available_at timestamptz NOT NULL DEFAULT now(),
sent_at timestamptz,
claim_token uuid,
claim_until timestamptz,
attempts integer NOT NULL DEFAULT 0
);
CREATE INDEX IF NOT EXISTS export_outbox_pending
ON export_outbox(available_at, created_at) WHERE sent_at IS NULL;
2. 让提交记录与事件一起提交
在state_store.py的submit_snapshot里，把下面INSERT放到原有with connect()事务中、export_jobs INSERT之后。删除producer中直接export_notes.delay的调用。HTTP202只能在数据库事务提交后返回，暂时broker不可用仍可保留待发布事件。
文件：submit_outbox_export.py
# 放在submit_snapshot的同一个with connect() as conn内：
# conn.execute(
# "INSERT INTO export_outbox(id,job_id) VALUES (%s,%s)",
# (str(uuid4()), job_id),
# )
# 提交侧现在只调用：
from state_store import submit_snapshot
job_id = submit_snapshot([1])
print("accepted", job_id)
3. 领取一条待发布事件
创建publisher.py。CTE锁定一行并更新领取token，连接上下文退出后才发消息。该函数一次处理一条，便于观察；循环调用时需停机信号、空队列等待与有限批量。
文件：publisher.py
from uuid import uuid4
from state_store import connect
from celery_app import app
def claim_event():
token = str(uuid4())
with connect() as conn:
return conn.execute("""
WITH candidate AS (
SELECT id FROM export_outbox
WHERE sent_at IS NULL AND available_at<=now()
AND (claim_until IS NULL OR claim_until<now())
ORDER BY created_at FOR UPDATE SKIP LOCKED LIMIT 1
)
UPDATE export_outbox o SET claim_token=%s,
claim_until=now()+interval '30 seconds', attempts=attempts+1
FROM candidate c WHERE o.id=c.id
RETURNING o.id,o.job_id,o.claim_token
""", (token,)).fetchone()
4. 发送后按token确认
追加并执行python publisher.py。消息只带job_id。broker调用不包在数据库事务内；失败时不标sent，等待领取租约到期后可再次尝试。生产还应限制发布重试总量并报警，不能无日志地永久积压。
文件：publisher.py
def publish_one():
event = claim_event()
if event is None:
return False
app.send_task(
"notes.export", args=[str(event["job_id"])],
task_id=str(event["id"]), queue="exports",
retry=False,
)
with connect() as conn:
conn.execute("""
UPDATE export_outbox SET sent_at=now(), claim_until=NULL
WHERE id=%s AND claim_token=%s AND sent_at IS NULL
""", (event["id"], event["claim_token"]))
return True
if __name__ == "__main__":
print("published", publish_one())
5. 注入发送后未标记的故障
只在本地publisher临时把发送之后的数据库更新跳过一次。等待该事件租约到期再执行publisher，它会再次发送；检查export_jobs只有一个有效结果引用。恢复更新代码；另试停止broker后提交，确认任务和outbox同时存在。
文件：终端 / 当前课时
SELECT j.id, j.status, j.result_ref, o.id AS event_id, o.attempts, o.sent_at
FROM export_jobs j JOIN export_outbox o ON o.job_id=j.id
ORDER BY o.created_at DESC LIMIT 10;
验收结果
- 提交事务提交后，job与outbox事件同时存在。
- broker不可用时sent_at保持NULL，事件可被后续处理。
- 重复发布不会让旧worker越过租约CAS覆盖终态结果。
生产约束与故障演练
数据库正常、Redis暂不可用，API不能受理后永久漏发任务。
实现决策
- 数据库提交的发送意图与业务记录同生共灭。
- publisher网络调用有超时和重试策略，长期未发布事件报警。
- 重复消息是常规故障路径，consumer继续使用持久幂等与租约。
隔离环境演练
分别模拟事务回滚、broker不可用、发送成功但sent_at未写回，检查三种窗口的恢复结果。
恢复验收
- 没有只存在job而无发送意图的已受理任务。
- 没有跨broker等待的数据库长事务。
- 重复发布可恢复且最终业务结果唯一可定位。
常见错误
outbox行已标sent但broker根本没收到
原因：先标记再发送，或把异常忽略。
处理：发送成功后再条件更新sent_at，失败保留待处理事件。
发送很慢导致数据库锁一直占用
原因：把broker发布放在FOR UPDATE事务内部。
处理：分成领取、网络发送、确认三个阶段，用领取token防旧publisher误确认。
扩展练习
在导出记录INSERT后、outbox INSERT前抛异常，验证两个表都没有这次提交的记录。
提示：使用同一个连接事务，让异常退出with触发回滚。
参考答案：
保存本次job_id进行查询；两表计数都应为0。若只回滚一张表，说明事务边界没有统一。
技术文档
13 限制受理量并拆分长短任务队列
在数据库事务中控制待处理容量，用独立worker减少大导出对小任务的影响。
本节目标
- 区分排队容量、执行并发和外部调用速率
- 用事务内受理检查避免并发超卖容量
- 按队列隔离大导出并观察等待时间
必要概念
增加worker不等于无限扩容
系统吞吐还受数据库连接、存储I/O和CPU限制。到达速度长期高于完成速度时，队列持续增长，必须在入口拒绝、等待或降级。
限并发与限速分别处理
concurrency控制同时执行，rate limit控制时间窗口速率；Celery任务rate_limit通常按worker实例生效，不等于全局额度。全局限制应在共享状态或独立入口实现。
队列拆分需要对应消费者容量
给fast和bulk命名不会自动隔离资源；分别启动受限worker，并配置CPU、连接等预算。prefetch降低预取囤积，但不能代替入队容量限制。
动手实现
1. 在受理事务内检查容量
在state_store.py添加admit_export，并在submit_snapshot的with connect()内、INSERT之前调用。所有提交路径必须使用同一检查，包括之后定时任务和重放。该本地全局锁只持有短事务；多租户系统可设计分租户额度。
文件：state_store.py
class CapacityExceeded(Exception):
pass
def admit_export(conn, limit: int = 100):
conn.execute("SELECT pg_advisory_xact_lock(831701)")
row = conn.execute("""
SELECT count(*) AS total FROM export_jobs
WHERE status IN ('queued','running')
""").fetchone()
if row["total"] >= limit:
raise CapacityExceeded("export capacity reached")
# submit_snapshot内部：
# with connect() as conn:
# admit_export(conn)
# ...读取快照、INSERT job、INSERT outbox...
2. 把事件路由写进outbox
执行SQL并在提交outbox时根据快照条数选择队列。本例≤30条走fast，其余走bulk。修改publisher的RETURNING包含queue_name，send_task的queue改为event['queue_name']；不要仅修改任务装饰器却保留显式旧queue。
文件：schema_capacity.sql
ALTER TABLE export_outbox
ADD COLUMN IF NOT EXISTS queue_name text NOT NULL DEFAULT 'exports.bulk'
CHECK (queue_name IN ('exports.fast','exports.bulk'));
-- 在提交事务中使用参数化INSERT：
-- INSERT INTO export_outbox(id,job_id,queue_name) VALUES (%s,%s,%s);
-- Python: queue_name = "exports.fast" if len(rows) <= 30 else "exports.bulk"
3. 分别启动两个受限worker
先完成第12课旧exports队列中的实验任务，再停旧worker。两个命令分别放在终端运行，各一个执行进程；worker_prefetch_multiplier=1继续保留。正常发布的事件必须进入其中一条被消费的队列。
文件：终端 / 当前课时
# Linux / WSL2，终端A
python -m celery -A celery_app:app worker -l INFO -Q exports.fast --concurrency=1 -n fast@%h
# Linux / WSL2，终端B
python -m celery -A celery_app:app worker -l INFO -Q exports.bulk --concurrency=1 -n bulk@%h
4. 用数据库观察积压与最老等待
暂停本地消费者后少量提交，逐渐达到本课100条容量；CapacityExceeded应由API转换为503并附Retry-After，而不是继续受理。先统计之前练习遗留的queued/running，不能假定数据库为空。恢复worker和publisher后观察最老等待下降。
文件：终端 / 当前课时
SELECT status, count(*) AS jobs,
extract(epoch FROM now()-min(created_at)) AS oldest_age_seconds
FROM export_jobs
WHERE status IN ('queued','running')
GROUP BY status;
SELECT queue_name, count(*) AS unpublished,
extract(epoch FROM now()-min(created_at)) AS oldest_unpublished_seconds
FROM export_outbox WHERE sent_at IS NULL GROUP BY queue_name;
5. 给HTTP受理失败返回明确反馈
将CapacityExceeded映射放在现有提交接口，不创建第二个状态真源。不同用户配额用429或业务约定，整个服务容量不足本课使用503。数据库事务回滚后不得返回202和虚假job_id。
文件：admission_adapter.py
from fastapi import HTTPException
from state_store import CapacityExceeded, submit_snapshot
def accept_export(note_ids: list[int]):
try:
return {"job_id": submit_snapshot(note_ids)}
except CapacityExceeded as exc:
raise HTTPException(
503, "导出容量暂满，请稍后重试",
headers={"Retry-After": "5"},
) from exc
验收结果
- 并发提交不会让受理记录超过配置容量。
- fast与bulk事件分别被对应worker消费。
- 过载时返回503而不是202，恢复后最老等待逐步下降。
生产约束与故障演练
大批量导出占满worker，小型用户导出等待过久。
实现决策
- 入口容量、worker并发、数据库连接和下游速率分别设限。
- 队列隔离有对应的消费者资源，不能只换名称。
- 积压监控同时看数量与最老等待，outbox未发布也计入受理容量。
隔离环境演练
仅暂停本地bulk消费者，提交大小两类任务，观察fast仍处理；随后压到小容量上限确认拒绝。
恢复验收
- 受理总量有可验证上限。
- 关键队列保留处理能力。
- 503请求没有留下已受理但无状态的记录。
常见错误
队列分了级，小任务仍一直等待
原因：所有队列仍由同一批已被长任务占满的worker处理。
处理：为关键小任务保留独立消费者和资源预算。
检查count后仍突破容量
原因：多个producer并发读取相同count，检查与INSERT不在同一串行化范围。
处理：同一事务获取约定锁，检查后写job与outbox，所有入口遵守。
扩展练习
将本地受理上限改为5，用两个producer同时提交，核对queued/running总数没有超过5。
提示：先处理或隔离此前活动任务，再做小容量实验。
参考答案：
超过容量的事务抛CapacityExceeded并回滚；只有实际写入状态表和outbox的请求能返回202。
技术文档
14 用持久时间槽合并周期执行，并验证跨分钟重投边界
Celery beat负责唤醒，数据库唯一键合并同一执行分钟；跨分钟重投和停机补跑明确作为不同问题验收。
本节目标
- 配置Celery beat周期与UTC时间
- 阻止重复调度信号创建重复导出
- 区分调度去重与执行租约
- 复现同一信号跨分钟执行生成不同时间槽的边界
必要概念
beat负责发送，worker负责执行
本课按worker实际开始处理时的UTC分钟生成slot_key，数据库唯一约束保证同一槽只创建一个job。这是按执行分钟合并，不是按原始beat信号去重：同一消息如果跨分钟重投，会落到不同槽并可能新建任务。
时间槽固定一个业务周期
本地例子用worker处理信号时的UTC分钟作为slot_key，同一分钟最多创建一份快照。它不补齐停机错过的分钟；需要严格补跑时，应由持久调度表保存due_at与补跑规则。
不重叠执行还需要租约
同一时间槽不重复创建，并不阻止不同时间槽的大导出重叠。复用任务领取租约，并按业务决定跳过、合并或限制同一导出类型的并发。
需要计划时间语义时使用持久调度游标
若业务要求每个计划分钟最多一次且停机后补跑，应在数据库保存schedule_id、next_due_at和唯一(schedule_id,scheduled_at)槽。调度事务锁定配置、按有限批次推进游标并写job/outbox；worker消费持久scheduled_at，不能再用执行时now猜测原计划。本课保留较小的执行分钟模型，下面用实际两次调用证明边界。
动手实现
1. 创建时间槽唯一约束
执行SQL。外键延迟到事务提交时检查，允许先争取slot_key，再插入对应job。冲突意味着该时间槽已有任务，不能再创建第二条。
文件：schema_schedule.sql
CREATE TABLE IF NOT EXISTS export_schedule_slots (
slot_key text PRIMARY KEY,
job_id uuid NOT NULL UNIQUE REFERENCES export_jobs(id)
DEFERRABLE INITIALLY DEFERRED,
created_at timestamptz NOT NULL DEFAULT now()
);
2. 在同一事务中创建周期任务
新建schedule_store.py。复用connect、admit_export；生产参数应包含固定租户和授权范围，本地只取给定笔记。没有长任务或broker调用进入这个事务。
文件：schedule_store.py
from uuid import uuid4
from psycopg.types.json import Jsonb
from state_store import connect, admit_export
def schedule_once(slot_key: str, note_ids: list[int]):
ids = sorted(set(note_ids))
if not ids or len(ids) > 100:
raise ValueError("select 1..100 notes")
job_id = str(uuid4())
with connect() as conn:
won = conn.execute("""
INSERT INTO export_schedule_slots(slot_key,job_id) VALUES (%s,%s)
ON CONFLICT(slot_key) DO NOTHING RETURNING slot_key
""", (slot_key, job_id)).fetchone()
if won is None:
return None
admit_export(conn)
rows = conn.execute(
"SELECT id,title,content,pinned FROM notes WHERE id=ANY(%s) ORDER BY id",
(ids,),
).fetchall()
if len(rows) != len(ids):
raise ValueError("one or more scheduled notes do not exist")
conn.execute("INSERT INTO export_jobs(id,payload) VALUES (%s,%s)",
(job_id, Jsonb(rows)))
conn.execute("""
INSERT INTO export_outbox(id,job_id,queue_name) VALUES (%s,%s,%s)
""", (str(uuid4()), job_id, "exports.bulk"))
return job_id
3. 注册周期信号并配置beat
新建schedule_tasks.py；将模块加入Celery include并重启worker。把下方app.conf部分放入celery_app.py。此信号使用fast队列，实际导出由outbox进入bulk；继续运行publisher才能把事件变成消息。 execution_slot接受显式时间，便于下一个实验固定跨分钟时序；正式tick仍使用开始执行时的now，因此不能宣称保留原beat计划时间。
文件：schedule_tasks.py
from datetime import datetime, timezone
from celery_app import app
from state_store import connect
from schedule_store import schedule_once
def execution_slot(when: datetime) -> str:
if when.tzinfo is None:
raise ValueError("timezone-aware timestamp required")
return when.astimezone(timezone.utc).strftime("notes-export:%Y%m%dT%H%M")
@app.task(name="notes.schedule_tick")
def schedule_tick():
slot = execution_slot(datetime.now(timezone.utc))
with connect() as conn:
rows = conn.execute("SELECT id FROM notes ORDER BY id LIMIT 100").fetchall()
ids = [row["id"] for row in rows]
return schedule_once(slot, ids) if ids else None
# 以下配置写入celery_app.py：
# app.conf.timezone = "UTC"
# app.conf.beat_schedule = {
# "export-each-minute": {
# "task": "notes.schedule_tick", "schedule": 60.0,
# "options": {"queue": "exports.fast"},
# }
# }
4. 启动一个beat并验证重复信号
Linux/WSL2单独终端执行命令，启动文件只给本次beat使用。不要为了高可用复制同一个本地schedule文件。用schedule_once对同一个测试slot连续调用两次，第一次返回job_id，第二次None。
文件：终端 / 当前课时
python -m celery -A celery_app:app beat -l INFO --schedule=./beat-lab.db
# 另一个终端在Python里测试同槽去重：
# from schedule_store import schedule_once
# print(schedule_once("drill:unique-slot-01", [1]))
# print(schedule_once("drill:unique-slot-01", [1]))
5. 证明同槽去重与跨分钟重投的区别
本实验直接调用同一slot生成逻辑，并注入两个明确的UTC时间，不必等待真实分钟切换。每次生成新的drill前缀避免与真实调度或此前实验混用。相同delivery_label表示模拟同一信号；两次不同分钟仍会创建两个job，结果是当前模型的边界，不应标成原消息exactly-once。
文件：verify_schedule_slots.py
from datetime import datetime, timezone
from uuid import uuid4
from state_store import connect
from schedule_store import schedule_once
from schedule_tasks import execution_slot
with connect() as conn:
note_id = conn.execute("SELECT min(id) AS id FROM notes").fetchone()["id"]
assert note_id is not None
prefix = f"drill:{uuid4()}:"
delivery_label = "same-redelivered-signal"
t1 = datetime(2026,1,1,10,0,59,tzinfo=timezone.utc)
t2 = datetime(2026,1,1,10,1,1,tzinfo=timezone.utc)
first = schedule_once(prefix+execution_slot(t1),[note_id])
same_slot = schedule_once(prefix+execution_slot(t1),[note_id])
later_slot = schedule_once(prefix+execution_slot(t2),[note_id])
assert first is not None and same_slot is None
assert later_slot is not None and later_slot != first
print(delivery_label,{"first":first,"same_slot":same_slot,"later_slot":later_slot})
# 此处会创建两条真实本地job与outbox事件；按正常publisher/worker流程处理。
6. 验证停机与时区行为
停止本地beat两分钟再启动，查询时间槽表。本例不会自动补发停机期间每一分钟的导出；明确记录这一行为。若需要补跑，先设计持久due_at游标和补跑容量，再修改任务。
文件：终端 / 当前课时
SELECT slot_key,job_id,created_at
FROM export_schedule_slots ORDER BY created_at DESC LIMIT 10;
验收结果
- 同一slot_key最多对应一个job_id与初始outbox事件。
- beat只发送信号，文件由worker和publisher链路完成。
- 时区固定UTC，停机遗漏的分钟不会被伪装成已执行。
- 同一模拟信号跨分钟执行会生成两个槽，明确记录这与同槽去重的区别。
生产约束与故障演练
每分钟为笔记生成一份小快照，调度进程偶尔重启。
实现决策
- 数据库唯一键合并同一执行分钟的任务；本课不承诺原始beat信号跨分钟重投只创建一次。
- 定义时区、时间槽和停机补跑语义，不能依靠默认本地时间。
- 慢任务的重叠策略和补跑容量单独控制。
- 需要固定计划时间和补跑时，升级为持久next_due_at游标与唯一scheduled_at槽，受理额度不足时不推进游标。
隔离环境演练
依次验证同槽重复调用、同一模拟信号跨分钟调用、beat停机两分钟三种情形；当前模型分别得到合并、新槽、无自动补跑。
恢复验收
- 重复调度不能创建两份同槽业务导出。
- 没有把beat当任务执行器或持久业务状态真源。
- 停机后的补跑规则可明确说明。
- 不把执行分钟合并误写成原始信号exactly-once；跨分钟行为有实验证据。
常见错误
启动两个beat后定时任务数量翻倍
原因：多个调度器同时发送同一计划。
处理：常态运行单beat，数据库slot唯一约束兜住重复业务创建。
以为周期调度天然不重叠
原因：上一分钟任务未完成，下一分钟仍会生成新任务。
处理：按业务增加导出类型并发限制或跳过策略，保留任务执行租约。
扩展练习
用两个producer并发调用同一全新slot_key，确认只有一个返回job_id。
提示：唯一约束需要在数据库事务中争取，不能先SELECT判断再INSERT。
参考答案：
一方成功创建job和outbox，另一方ON CONFLICT返回None。最终一个slot、一条业务任务；数据库事务保证外键成立。
技术文档
15 实现优雅退出、取消与失败隔离恢复
按持久退避恢复过期租约，重新派发久未领取的queued任务，并保留取消、失败与人工重放记录。
本节目标
- 区分请求取消与实际停止
- 恢复过期租约并限制重复尝试
- 隔离耗尽任务并保留重放审计
- 用next_attempt_at阻止重复消息绕过退避
- 恢复已确认派发但仍未被领取的queued任务
必要概念
取消需要执行者配合
取消标志表示用户请求停止，不能证明线程、子进程或外部写入已经停止。执行者在分块边界检查权属与取消，发布结果前再次CAS；不要把terminate当普通业务取消。
正常退出先停止取新任务
Celery worker收到TERM进行warm shutdown，等待正在运行工作；强制终止不能保证finally执行。退出宽限期要覆盖有界单次任务，恢复依赖状态租约和未确认消息机制。
死信是可核对的业务隔离
Redis broker没有自动替本模块建立业务死信流程。本课用export_dead_letters保存尝试耗尽或过期任务，原任务保持failed；修复后创建新job重放，保留原始失败证据。
派发成功与任务领取是不同事实
publisher收到发送成功后，消息仍可能在broker故障中丢失；worker也可能在claim之前因数据库连接错误退出。两者都可能留下queued和已sent事件。恢复扫描还要查久未领取、没有未发送事件的queued任务，原子新建outbox。重复派发允许存在，数据库claim决定谁能执行。
动手实现
1. 增加取消标记与隔离表
请求取消只修改活跃状态记录，终态不回退。取消路由仍应复用notes-service身份与任务归属校验。运行中的文件操作可能继续片刻，但第11课finish必须检查取消和deadline。
文件：schema_recovery.sql
CREATE TABLE IF NOT EXISTS export_dead_letters (
job_id uuid PRIMARY KEY REFERENCES export_jobs(id),
reason text NOT NULL,
quarantined_at timestamptz NOT NULL DEFAULT now()
);
ALTER TABLE export_jobs ADD COLUMN IF NOT EXISTS replay_of uuid REFERENCES export_jobs(id);
-- 用参数化语句处理一个已授权的job_id：
-- UPDATE export_jobs SET cancel_requested=true, updated_at=now()
-- WHERE id=%s AND status IN ('queued','running') RETURNING id,status;
2. 在处理边界检查权属
把can_continue追加到state_store.py，在生成每个数据块前调用，返回False就停止后续工作。小文件实验可在write_text前检查；完成CAS仍不可省略。有效租约也必须处于业务deadline以内。
文件：state_store.py
def can_continue(job_id: str, token: str) -> bool:
with connect() as conn:
row = conn.execute("""
SELECT id FROM export_jobs
WHERE id=%s AND lease_token=%s AND status='running'
AND lease_until>now() AND deadline_at>now()
AND cancel_requested=false
""", (job_id, token)).fetchone()
return row is not None
# export_tasks.py生成文件前：
# if not can_continue(job_id, token):
# return {"job_id": job_id, "published": False}
3. 在publisher确认时记录派发时间
替换publisher.py中的publish_one，保留原claim_event，并确认第13课已让RETURNING包含queue_name。事件确认与last_dispatched_at在同一短事务中写入，网络发送仍在事务外。UPDATE未匹配token时不写job时间，旧publisher不能延迟新一轮恢复判断。last_dispatched_at只表示发送响应已确认，不保证消息永不丢失。
文件：publisher.py
def publish_one():
event = claim_event()
if event is None:
return False
app.send_task(
"notes.export",args=[str(event["job_id"])],
task_id=str(event["id"]),queue=event["queue_name"],retry=False,
)
with connect() as conn:
confirmed = conn.execute("""
UPDATE export_outbox SET sent_at=now(),claim_until=NULL
WHERE id=%s AND claim_token=%s AND sent_at IS NULL
RETURNING job_id
""",(event["id"],event["claim_token"])).fetchone()
if confirmed is not None:
conn.execute("UPDATE export_jobs SET last_dispatched_at=now() WHERE id=%s",
(confirmed["job_id"],))
return confirmed is not None
4. 扫描租约与未领取任务，原子生成恢复事件
新建recovery.py。每次最多50条；锁内只做数据库操作。running租约过期后设next_attempt_at=10秒后，outbox同时使用该时间；所有claim都检查此字段。queued超过最近一次确认派发60秒仍未领取、且不存在待发送事件时重新派发，解决消息丢失或claim前失败。旧queued缺last_dispatched_at时兼容读取已有sent_at。60秒是本机观察阈值，生产应结合队列正常等待SLO设置，过短会制造重复消息。 无消费能力时最多按阈值重新派发直到业务deadline，派发不增加执行attempts；attempts只在claim成功时增加。仍有未发送outbox时由publisher租约重试，扫描器不重复堆新事件。课件一次性命令只处理一批；部署时应以监督进程或调度器每5秒调用扫描器并持续运行publisher，空批次等待、异常告警、关闭时停止取新工作。
文件：recovery.py
from uuid import uuid4
from state_store import connect
def recover_batch():
with connect() as conn:
rows = conn.execute("""
SELECT j.*,j.deadline_at<=now() AS expired
FROM export_jobs j
WHERE (j.status='running' AND j.lease_until<now())
OR (j.status='queued' AND
(j.cancel_requested OR j.deadline_at<=now() OR j.attempts>=3))
OR (j.status='queued' AND j.next_attempt_at<=now()
AND j.cancel_requested=false AND j.deadline_at>now() AND j.attempts<3
AND NOT EXISTS (
SELECT 1 FROM export_outbox pending
WHERE pending.job_id=j.id AND pending.sent_at IS NULL)
AND GREATEST(j.updated_at,j.next_attempt_at,
COALESCE(j.last_dispatched_at,
(SELECT max(o.sent_at) FROM export_outbox o WHERE o.job_id=j.id),
j.created_at)) < now()-interval '60 seconds')
ORDER BY j.updated_at FOR UPDATE OF j SKIP LOCKED LIMIT 50
""").fetchall()
for row in rows:
if row["cancel_requested"]:
status,reason,delay = "cancelled","user_cancelled",0
elif row["expired"] or row["attempts"] >= 3:
status = "failed"
reason = "deadline_exceeded" if row["expired"] else "attempts_exhausted"
delay = 0
elif row["status"] == "running":
status,reason,delay = "queued","lease_expired",10
else:
status,reason,delay = "queued","dispatch_overdue",0
updated = conn.execute("""
UPDATE export_jobs SET status=%s,error_code=%s,
lease_token=NULL,lease_until=NULL,
next_attempt_at=now()+(%s * interval '1 second'),updated_at=now()
WHERE id=%s RETURNING next_attempt_at
""",(status,reason,delay,row["id"])).fetchone()
if status == "failed":
conn.execute("""
INSERT INTO export_dead_letters(job_id,reason) VALUES (%s,%s)
ON CONFLICT(job_id) DO NOTHING
""",(row["id"],reason))
if status == "queued":
conn.execute("""
INSERT INTO export_outbox(id,job_id,queue_name,available_at)
VALUES (%s,%s,'exports.bulk',%s)
""",(str(uuid4()),row["id"],updated["next_attempt_at"]))
return len(rows)
if __name__ == "__main__":
print("recovered",recover_batch())
5. 验证重复消息不能跳过持久退避
先暂停本机publisher与consumer，防止实验记录被正常流程抢先处理。只针对新建实验job设置next_attempt_at为未来10秒；两次直接claim模拟已有重复消息。两次均为None，attempts保持0。到期后claim成功且attempts变1。演练需要10秒等待，只在本地执行；任务后续通过正常worker输出或恢复扫描到终态。
文件：verify_retry_gate.py
import time
from state_store import connect,submit_snapshot,claim
with connect() as conn:
note_id=conn.execute("SELECT min(id) AS id FROM notes").fetchone()["id"]
assert note_id is not None
job_id=submit_snapshot([note_id])
with connect() as conn:
conn.execute("UPDATE export_jobs SET next_attempt_at=now()+interval '10 seconds' WHERE id=%s",(job_id,))
assert claim(job_id) is None
assert claim(job_id) is None
with connect() as conn:
assert conn.execute("SELECT attempts FROM export_jobs WHERE id=%s",(job_id,)).fetchone()["attempts"] == 0
time.sleep(10.1)
job=claim(job_id)
assert job is not None and job["attempts"] == 1
print("persistent retry gate verified",job_id)
# 此处人为持有新租约；不伪造成功结果，等待租约到期后由恢复流程重新排队。
6. 模拟派发后消息丢失，不删除broker数据
暂停本机publisher和consumer后执行本实验，避免正常进程抢先处理。使用新建job的outbox行模拟发送已确认，但不真的发送Celery消息；只修改该job时间。扫描后必须出现新的pending outbox，第二次扫描不能再给它新增事件。恢复publisher和consumer后应正常执行。这覆盖排队消息丢失，也覆盖worker在claim前数据库连接失败的相同持久状态。
文件：verify_queued_recovery.py
from state_store import connect,submit_snapshot
from recovery import recover_batch
with connect() as conn:
note_id=conn.execute("SELECT min(id) AS id FROM notes").fetchone()["id"]
assert note_id is not None
job_id=submit_snapshot([note_id])
with connect() as conn:
conn.execute("""
UPDATE export_outbox SET sent_at=now()-interval '61 seconds',claim_until=NULL
WHERE job_id=%s
""",(job_id,))
conn.execute("""
UPDATE export_jobs SET last_dispatched_at=now()-interval '61 seconds',
updated_at=now()-interval '61 seconds',next_attempt_at=now()-interval '61 seconds'
WHERE id=%s
""",(job_id,))
recover_batch()
with connect() as conn:
pending=conn.execute("SELECT count(*) AS n FROM export_outbox WHERE job_id=%s AND sent_at IS NULL",(job_id,)).fetchone()["n"]
assert pending == 1
recover_batch()
with connect() as conn:
assert conn.execute("SELECT count(*) AS n FROM export_outbox WHERE job_id=%s AND sent_at IS NULL",(job_id,)).fetchone()["n"] == 1
assert conn.execute("SELECT attempts FROM export_jobs WHERE id=%s",(job_id,)).fetchone()["attempts"] == 0
print("queued delivery recovery verified",job_id)
# 恢复正常publisher和consumer，再核对该job终态与输出。
7. 先正常停机，再验证租约恢复
对本次隔离worker先停止取新任务或直接发送TERM，观察warm shutdown等待当前有界任务。只在故障演练时终止已核对的执行进程；30秒租约过期后运行recovery.py，再让publisher发布到期事件。不要使用模糊进程名批量终止。
文件：终端 / 当前课时
python -m celery -A celery_app:app inspect active
python -m celery -A celery_app:app inspect reserved
# 正常关闭：对已核对的本次worker PID发送TERM，或首次Ctrl+C。
python recovery.py
python publisher.py
8. 修复后创建新任务重放
新建replay_job.py，在命令行传一个明确的failed任务ID。事务中保留payload、记录replay_of、建立新outbox；原dead letter不删除。自动批量重放前必须有故障已修复的证据，本模块只逐条本地重放。
文件：replay_job.py
import sys
from uuid import UUID, uuid4
from state_store import connect, admit_export
def replay(old_id: str):
old_id = str(UUID(old_id))
new_id = str(uuid4())
with connect() as conn:
admit_export(conn)
row = conn.execute("""
INSERT INTO export_jobs(id,payload,replay_of)
SELECT %s,payload,id FROM export_jobs WHERE id=%s AND status='failed'
RETURNING id
""", (new_id, old_id)).fetchone()
if row is None:
raise ValueError("failed source job not found")
conn.execute("""
INSERT INTO export_outbox(id,job_id,queue_name)
VALUES (%s,%s,'exports.bulk')
""", (str(uuid4()), new_id))
return new_id
if __name__ == "__main__":
print(replay(sys.argv[1]))
验收结果
- 取消任务不能被finish发布为成功，最终由执行者或扫描器转cancelled。
- 过期租约恢复产生新事件，attempts不归零，最多3次领取。
- 耗尽任务进入failed和dead letter，重放使用新job_id并记录replay_of。
- 已sent但未领取的queued任务能产生新的恢复事件；已有pending事件时不会每轮扫描继续堆积。
- next_attempt_at到期前重复消息无法增加attempts，派发次数与实际执行次数分开。
生产约束与故障演练
发布新版本需要停worker，用户同时请求取消导出，部分任务已多次崩溃。
实现决策
- 取消、正常退出、强制终止分别定义结果和恢复方式。
- 扫描器只处理过期或明确终止条件，使用短事务与有限批次。
- 故障修复后以新任务重放，保留旧失败状态和审计关联。
- queued久未领取和running租约到期分别恢复；派发确认不作为完成证据。
- 持久next_attempt_at约束所有消息，包括已有重复投递；broker倒计时不是唯一退避控制。
- publisher与恢复扫描必须由持续监督运行的进程定期执行，单次脚本不是生产调度器。
隔离环境演练
依次验证过期租约、未来next_attempt_at收到重复消息、已sent却queued、已有pending事件抑制重复恢复、3次耗尽与单条重放；每次仅操作记录的本地job_id。
恢复验收
- 有效租约不会被扫描器抢占。
- 尝试耗尽不会形成无限重投循环。
- 重放有来源记录且不会抹掉原始失败证据。
- 派发已确认但消息未领取的任务可恢复，不只等deadline后失败。
- 退避时间未到时重复消息不消耗执行attempts。
- 暂时无消费者时重新派发有阈值和deadline限制，不无限快速重投。
常见错误
revoke后立即在页面显示已取消
原因：revoke或发送取消信号不等于所有执行者与外部操作已停止。
处理：先显示取消申请，待持久状态确认后显示cancelled。
恢复脚本把所有running直接改queued
原因：仍有有效租约的执行者会与新worker竞争。
处理：只恢复过期租约，旧执行者发布必须受token、时间和状态CAS约束。
outbox已sent但任务一直queued
原因：消息可能丢失，或worker在claim之前因数据库错误退出；只恢复running无法覆盖。
处理：按持久派发时间扫描久未领取且无pending事件的queued任务，原子重建发送意图；仍用claim防重复执行。
扩展练习
连续让同一隔离任务领取后租约过期，验证第3次后进入failed，不产生第4次领取。
提示：只操作本次新建任务，并通过正常claim增加attempts。
参考答案：
前两次过期产生恢复事件；第三次过期被扫描器隔离，attempts保持3，后续claim返回None；原错误和重放链可查询。
技术文档
16 完成故障注入与可观测性验收
用同一组任务验证受理、发布、领取、执行、取消、恢复与结果查询的完整链路。
本节目标
- 为每个阶段记录同一业务job_id
- 用可重复的故障矩阵验证状态与输出
- 建立队列等待、运行耗时和恢复指标
必要概念
指标要对应用户等待过程
排队数量之外，还要测受理到开始、运行时长、总完成时间、最老待发布年龄、失败率和重试次数。只看worker存活无法知道任务是否积压。
追踪业务ID与交付ID
日志同时记录job_id、outbox_id或Celery task_id、lease_token和attempt，才能区分同一个任务的重复消息与重新执行。日志不打印完整笔记正文、数据库密码或访问密钥。
验收包括故障后状态一致性
任务成功需要有效结果引用和可解析文件；取消或失败不能被旧worker覆盖。单元测试能检查状态转换，真实broker、worker和数据库故障窗口仍需隔离集成演练。
动手实现
1. 给worker添加结构化阶段日志
在worker领取后和完成后记录下面格式，使用真实变量。错误日志保留异常类型与关联ID，正文快照不进入日志。要记录Celery消息ID可将任务改为bind=True，并使用self.request.id；同时保持对外任务名notes.export。
文件：task_logging.py
import json
import logging
logger = logging.getLogger("notes.exports")
def log_phase(job_id, phase, message_id=None, lease_token=None, attempt=None, **fields):
logger.info(json.dumps({
"event": "export_phase", "job_id": str(job_id),
"phase": phase, "message_id": message_id,
"lease_token": str(lease_token) if lease_token else None,
"attempt": attempt, **fields,
}, ensure_ascii=False))
# log_phase(job_id, "claimed", self.request.id, token, job["attempts"])
# log_phase(job_id, "published", self.request.id, token, job["attempts"],
# success=published)
2. 查询积压、租约和失败
保存SQL为monitor_exports.sql，在本地演练前后各执行一次。age和数量一起看；注意取消等待清理、outbox未发布和running租约过期是不同阶段。当前updated_at不能直接当精确执行耗时，需另记started_at/finished_at或使用结构化阶段时间戳。
文件：monitor_exports.sql
SELECT status,count(*) AS total
FROM export_jobs GROUP BY status;
SELECT count(*) AS unpublished,
extract(epoch FROM now()-min(created_at)) AS oldest_unpublished_seconds
FROM export_outbox WHERE sent_at IS NULL;
SELECT count(*) AS expired_leases
FROM export_jobs WHERE status='running' AND lease_until<now();
SELECT reason,count(*) AS failed
FROM export_dead_letters GROUP BY reason;
SELECT count(*) AS queued_without_pending,
extract(epoch FROM now()-min(COALESCE(j.last_dispatched_at,j.created_at))) AS oldest_dispatch_age
FROM export_jobs j
WHERE j.status='queued' AND NOT EXISTS (
SELECT 1 FROM export_outbox o WHERE o.job_id=j.id AND o.sent_at IS NULL
);
SELECT id,status,attempts,next_attempt_at,last_dispatched_at
FROM export_jobs WHERE status IN ('queued','running') ORDER BY created_at LIMIT 20;
3. 以任务ID检查终态和文件
新建verify_terminal.py，运行时传本次job_id。成功时只读取artifacts目录内的受控结果，核对文件内容与固定payload。查询不是SUCCESS时明确显示实际状态，不用RedisPENDING代替数据库结果。
文件：verify_terminal.py
import json
import sys
from pathlib import Path
from uuid import UUID
from state_store import connect
job_id = str(UUID(sys.argv[1]))
with connect() as conn:
job = conn.execute("SELECT * FROM export_jobs WHERE id=%s", (job_id,)).fetchone()
assert job is not None, "job missing"
print("state", job["status"], "attempts", job["attempts"])
if job["status"] == "succeeded":
root = Path("artifacts").resolve()
target = Path(job["result_ref"]).resolve()
assert target.is_relative_to(root), "unexpected artifact location"
assert json.loads(target.read_text(encoding="utf-8")) == job["payload"]
assert not job["cancel_requested"]
print("artifact verified")
elif job["status"] in {"failed", "cancelled"}:
assert job["result_ref"] is None
print("terminal without published artifact")
4. 按矩阵逐个注入故障
每一行使用新建job_id，先记基线，再暂停或修改本次隔离组件，最后恢复并运行publisher/recovery。不要同时注入多种故障，否则无法确定恢复由哪个机制保证。下面是待执行清单，不代表已经跑过。
文件：failure-matrix.md
| 故障点 | 操作 | 应核对的结果 |
|---|---|---|
| DB事务提交前 | 本次提交主动抛异常 | job与outbox同时回滚 |
| broker不可用 | 暂停本地Redis后提交 | job与outbox保留，sent_at为空 |
| 发布后未标sent | 跳过一次确认写回 | 可重复发布，业务结果仍受CAS保护 |
| worker持租约退出 | 终止本次隔离执行进程 | 到期恢复，attempts增加 |
| 旧token晚到 | 新领取后用旧token完成 | finish=False，不改result_ref |
| 取消 | 设置本次cancel_requested | 最终cancelled，不发布成功结果 |
| 次数耗尽 | 3次领取均失败/过期 | failed及dead letter，无第4次领取 |
| 结果缓存过期 | 缩短本地result_expires | PostgreSQL状态仍可查询 |
| 队列满 | 临时小容量并发提交 | 503，无虚假202 |
| 同槽周期重复 | 相同slot并发调度 | 只创建一个周期job |
| 周期跨分钟重投 | 同一模拟信号用两个执行分钟 | 当前模型创建两个槽，不宣称原信号exactly-once |
| 已派发但未领取 | 专用job模拟sent且queued、时间超过阈值 | 扫描生成新outbox，恢复后完成 |
| claim之前失败 | worker进入claim前抛连接错误 | queued路径能重新派发 |
| 退避期间重复消息 | next_attempt_at未来时直接重复claim | 均返回None，attempts不增加 |
| pending事件仍在 | 同一个queued重复运行扫描器 | 不为已有pending事件继续新增恢复事件 |
5. 保存观测证据并核对恢复预算
为每次演练填写实际值：操作时间、job_id、消息ID、租约、状态转换、输出摘要、恢复耗时。只有实际运行后才能标pass。设置本地验收阈值，例如过期租约路径在60秒内重新进入处理、queued丢消息路径在90秒内重新进入处理；若超出，检查可见性时间、outbox租约、扫描周期与队列等待之和。 queued恢复包含60秒未领取阈值，不能拿租约恢复阈值直接套用；两条路径都要加入扫描周期、10秒退避（仅过期运行任务）和实际队列等待。
文件：verification-record.json
{
"case": "publish_after_send_before_mark",
"job_id": "填写本次实际ID",
"observed_states": [],
"duplicate_deliveries": null,
"recovery_seconds": null,
"artifact_sha256": null,
"result": "not_run",
"notes": "实际演练后填写；不要用预计行为替代观测"
}
验收结果
- 每个故障案例有实际job_id、时间、状态和恢复证据才标通过。
- 成功文件与payload一致，失败或取消不会被旧token发布成功。
- 结果缓存过期不导致业务记录404，容量与调度去重也经过验证。
- 已sent未领取、退避重复消息与跨分钟周期重投都有独立案例，不被普通worker存活检查替代。
- 任务执行attempts不会因单纯重新派发增加；等待有持久时间和deadline边界。
生产约束与故障演练
准备把导出服务交给他人维护，需要证明故障可恢复且结果可追踪。
实现决策
- 先设受理、开始、完成和恢复的可量化目标，再配置阈值。
- 日志以业务ID关联，不把敏感笔记内容当诊断材料。
- 把未执行的演练标为not_run，只有实际记录支持通过结论。
隔离环境演练
逐项执行本课矩阵，恢复所有本地组件与临时配置后，再跑一次无故障导出作为回归。
恢复验收
- 每项恢复结论都有观测证据。
- 已受理任务能定位到待发布、运行或明确终态。
- 不会把语法校验或eager函数测试描述为生产可靠性证明。
- 恢复覆盖queued和running两类滞留状态，时间槽去重范围与实际实现一致。
常见错误
只有worker心跳正常就判定服务正常
原因：worker可能活着但队列积压、outbox不发布或任务卡住。
处理：同时观察最老等待、过期租约、失败和用户完成时间。
测试使用task_always_eager却声称broker恢复已验证
原因：eager在当前进程执行，绕过真实交付链路。
处理：eager只做函数级测试，消息重投与worker故障使用隔离集成环境。
扩展练习
选择“发布成功但sent_at未写回”案例，写出日志应包含的三个ID，并说明如何判断重复消息没有改变最终结果。
提示：区分业务job_id、outbox消息ID和租约token。
参考答案：
同一job_id关联重复投递日志，消息ID可能相同或不同；新领取有新token，只有有效token完成CAS。最终唯一result_ref指向与payload一致的文件，旧token完成返回False。