CoderPath

幂等任务与 ASGI 生命周期 ​

本站纵览 ｜ 练习答案

用数据库唯一约束记录事件身份：

python
from django.db import models


class ProcessedEvent(models.Model):
    event_id = models.CharField(max_length=64, unique=True)
    processed_at = models.DateTimeField(auto_now_add=True)
python
from django.db import IntegrityError, transaction
from django.tasks import task
from .models import ProcessedEvent


@task
def process_ticket_event(event_id, ticket_id):
    try:
        with transaction.atomic():
            ProcessedEvent.objects.create(event_id=event_id)
            # 在同一事务中执行可重放的数据库状态迁移。
    except IntegrityError:
        return {"status": "duplicate", "ticket_id": ticket_id}
    return {"status": "processed", "ticket_id": ticket_id}

深水边界 ​

唯一记录只能保护同一数据库内的状态；外部邮件/支付仍存在“调用成功、状态写回失败”窗口。ASGI 应用关闭时要停止接流量、等待有限时间内的请求并释放池；后台 Task worker 是另一套进程生命周期，不能依赖 Web 进程内全局变量。

验收 ​

相同 event_id 执行两次只有一次状态迁移；worker 重启后幂等记录仍在；应用关闭不会接收新长任务；超时和取消有日志。

练习（先独立完成） ​

  1. 并发执行相同 event_id，记录唯一冲突结果。
  2. 设计外部邮件的幂等键与补偿策略。
  3. 写 ASGI 与 Task worker 各自的就绪/存活检查。

完成后查看练习答案。

深度加餐：任务重试需要幂等，不需要祈祷 ​

网络重试、worker 崩溃和用户重复点击都会让同一任务执行多次；业务结果必须可重复计算。

先做一个可观察实验 ​

python
from django.db import transaction

def reserve_once(event_key, user_id):
    with transaction.atomic():
        event, created = OutboxEvent.objects.get_or_create(
            key=event_key,
            defaults={'user_id': user_id, 'status': 'PENDING'},
        )
        if not created and event.status == 'DONE':
            return 'already-done'
        event.attempts = event.attempts + 1
        event.save(update_fields=['attempts'])
        return 'run'

预期：唯一 key 把重复请求合并；已完成事件直接返回，不重复执行副作用。

原理下潜与工程边界 ​

幂等键的唯一约束必须落在数据库；任务状态机至少区分 PENDING、RUNNING、DONE、FAILED。ASGI 生命周期中的启动/关闭钩子要释放连接、线程和订阅，不要把状态藏在单进程内存。

标记 DONE 与外部支付成功之间可能出现崩溃窗口；需要 outbox、幂等 API 或供应商侧幂等键共同覆盖。取消是协作信号，不保证已经运行的代码立刻停止。

反例排错路线 ​

为每次尝试记录 key、attempt、worker、开始/结束时间和异常 cause；重复执行时检查数据库唯一冲突与外部响应，而不是只看日志行数。

练习加码（仍不提供答案） ​

加码：为任务状态机画出失败恢复图，覆盖 worker 崩溃、超时、重复投递和人工重放。

验收时请补充：正常路径、权限或失败路径、以及一条能证明事务/资源/配置边界没有被绕过的测试。

继续查阅：Django 官方文档（外部资料已收录于证据账本） · Django 安全清单（外部资料已收录于证据账本）

课程合同：任务幂等与 ASGI ​

学习目标 ​

学完本篇，读者应能把「重复投递同一幂等键不会重复扣费，取消可被安全处理」拆成输入、可信边界、Django 机制和可复查证据；同时能说明 Django 6.0 与 5.2 LTS 的差异不会改变业务不变量。

心智模型 ​

请求/数据输入 → 边界检查 → Django 核心机制 → 可观察输出 → 恢复动作。配置、数据库提交和外部副作用分别拥有自己的生命周期；没有证据的“应该可以”不算完成。

本地机制图 ​

任务幂等与 ASGI机制图

图中的“输入”是不可信的 HTTP、表单、任务或配置；“边界”负责身份、格式、权限、大小、超时与版本；“机制”才是 URL、View、ORM、事务、缓存、Tasks 或 ASGI；“证据”是测试、响应、日志、查询计划和发布记录。

最小实验 ​

在全新虚拟环境中运行：

bash
python -m django --version
python manage.py check
python manage.py test

实验只验证一个机制：输入一条有效值，再输入一条空值、重复值或无权限值。每次实验都记录 Django 版本、数据库、命令和退出码。

完整场景 ​

把实验接到 tickets 工单流：请求携带用户身份和工单 ID，经过当前专题负责的边界，完成一次读或写，并留下响应状态、数据库变化、日志字段和清理动作。完整场景必须能第二次执行，或明确说明幂等键、事务回滚、缓存失效和任务重试策略。

运行输出 ​

成功路径应出现：

text
Django 6.0.x
System check identified no issues (0 silenced).
OK

失败路径要保留真实信号，例如 400（输入无效）、403（权限/CSRF）、404（资源不存在）、405（方法不允许）或数据库约束异常；不要把失败改写成含糊的 200。

失败边界 ​

至少验证：空输入、重复提交、匿名用户、越权对象、并发写入、下游不可用、超时和版本不兼容。记录“谁拒绝、何时拒绝、状态是否回滚、重试是否安全”。开发模式的 traceback 不得进入生产响应，缓存不得跨用户复用敏感结果。

原理与排错 ​

先确认版本和 settings，再用 manage.py check 定位启动问题；接着确认 URL/任务入口、输入解析和权限；最后检查 SQL、事务提交、响应和副作用。若结果异常，用最小复现打开 SQL/请求 ID/任务 ID 观测，沿「复现 → 缩小 → 观测 → 修复 → 回归」闭环，不靠重装依赖猜测。

练习（正文不泄答案） ​

  1. 为上述完整场景补一条正常路径和两条失败路径测试，并记录完整输出。
  2. 画出本篇机制图，标出身份 owner、数据提交点、缓存/任务副作用和恢复动作。
  3. 故意制造一次重复、越权或超时，说明应返回的状态、日志字段和是否可重试。
  4. 写一条 Django 5.2 LTS 兼容验收标准，再说明 Django 6.0 的新能力是否需要条件分支。

验收标准：命令可重复、输出可复查、正文没有练习答案；完成后再打开本阶段答案册。

答案册 ​

本阶段答案册提供参考实现、关键推理、失败边界和替代方案；先独立完成练习再对照。

下一篇 ​

下一篇是09-testing-and-security/01-testing-security-csp，继续沿同一条输入—边界—机制—证据链前进。

用心记录每一步成长

Copyright © 2026-present Twiink