即使在连接中断之后仍会留下的 running：如何设计流式执行的终止

流式 API 的正常路径看起来很简单。启动外部操作，发送结果片段，最后将执行状态更改为 succeeded。

但如果用户在响应中途关闭了标签页，情况就不同了。HTTP 连接已经结束，但模型调用可能仍在继续；生成器可能已关闭，但执行记录可能仍保留为 running。接收完成事件的界面与数据库中的最终状态也可能不一致。

要解决此问题，需要分别定义连接、执行和持久化记录的生命周期，并连接它们之间的终止责任。 本文从 Python 异步生成器的最小复现出发，扩展到取消处理、事务边界以及多工作进程的执行所有权。

示例已在 Python 3.12.13 中验证。关于 AnyIO 和 SQLAlchemy 的说明分别基于 4.x 和 2.0 文档。后面的作业租约和事件存储结构是设计示例，与前部分的复现实验区分开来。

1. 首先分离四种生命周期

无需始终让这四种生命周期保持一致。相反，应根据产品行为确定连接规则。

• 依赖于请求的执行： 将连接断开作为取消信号。清理责任在于处理请求的执行者。
• 需要继续进行的作业： 将连接断开视为取消订阅。执行责任在于独立于请求的执行者。

如果这里的判断不明确，同一个程序中就会存在相互矛盾的代码。一个层级在连接结束时取消任务，另一个层级则等待完成通知。虽然两者单独来看都是自然的代码，但合并后的系统契约无法成立。

2. 从 async for 退出并不意味着上游生成器立即关闭

嵌套流式传输通常按如下方式构建。

외부 응답 생성기 → 실행 이벤트 생성기 → HTTP 응답 생성기

此时，如果各层级仅简单地进行 async for 读取并仅执行 yield，则外部生成器的终止可能无法连接到内部生成器的显式关闭。特别是当外部生成器在传递值后立即停止时，这一点尤为重要。

以下示例验证上游资源清理的时间点。

import asyncio
from contextlib import aclosing


async def upstream(closed):
    try:
        yield "first chunk"
        await asyncio.Event().wait()
    finally:
        closed.append("upstream closed")


async def relay(source):
    async with aclosing(source) as chunks:
        async for chunk in chunks:
            yield chunk


async def check():
    closed = []
    response = relay(upstream(closed))

    assert await anext(response) == "first chunk"
    await response.aclose()

    # 이벤트 루프 종료나 GC를 기다리기 전에 정리가 끝나야 한다.
    assert closed == ["upstream closed"]


asyncio.run(check())

在移除 aclosing() 的单独复现中，当保持对上游生成器的引用仅关闭外部生成器时，清理列表为空。应用 aclosing() 后，上游清理在 response.aclose() 返回前结束。当消费任务被 cancel() 时，也确认了清理和取消传播。

contextlib.aclosing() 在块退出时调用异步生成器的 aclose()。其关键在于将清理时间点绑定到与循环相同的执行上下文中。Python contextlib 官方文档

此规则应应用于每个流式适配器。不仅包括直接创建的生成器，还必须确定由谁关闭 SDK 返回的流。仅凭 HTTP 框架关闭最外部响应这一事实，并不能保证所有上游资源的生命周期。

3. 错误和取消是不同的控制流

考虑以下处理执行记录的代码。

try:
    result = await execute()
except Exception:
    record_failure()
else:
    record_success(result)

在 Python 3.12 中，asyncio.CancelledError 是 BaseException 的子类型。因此，上述 except Exception 不会处理取消。如果异步生成器被 aclose()，则在停止点会插入 GeneratorExit。这就是为什么必须单独验证取消和显式生成器关闭的原因。asyncio 任务取消文档、异步生成器方法文档

执行状态记录至少需要以下不变量。

1. 正常清理的执行具有终止状态。
2. 已确定为成功的执行不会被随后的连接结束 stopped 覆盖。
3. 如果有部分响应，则保留部分响应及停止记录。
4. 取消信号在清理完成后向上传播至上层。

在单个执行者内部，像 finalized 这样的局部变量有助于防止重复关闭。在提交成功或失败后将此变量设置为 true，并在最后的清理路径中处理尚未关闭的执行。

然而，不能仅因为到达了 finally 就将所有情况归类为取消。终止记录本身的数据库错误、进程中断、外部作业结果状态不明确等情况需要单独处理。单个局部变量所能解决的范围限于同一执行者内部的控制流。

4. finally 内的 await 也可能被取消

如果将清理代码移动到 finally，就会出现以下问题。

finally:
    await save_terminal_state()

这个 await 在什么取消策略下执行？

在 AnyIO 的取消作用域内，在取消状态持续期间，每次到达等待点都可能重新引发取消异常。因此，异步终止记录可能需要受限的清理部分。就像 move_on_after(..., shield=True) 那样，暂时阻止外部作用域的取消，但设置自身的超时限制。这种保护是根据 AnyIO 作用域的含义而定的，不应假设它会无条件地屏蔽任意的 asyncio.Task.cancel()。AnyIO 取消与清理文档

在清理部分中，最好区分以下内容。

• 资源清理： 关闭套接字·响应流·临时文件。
• 状态关闭： 持久化存储执行结果。
• 标记为恢复对象： 如果因超时或数据库故障未能关闭，则使恢复路径能够发现它。

如果保护整个作业免受取消，用户的停止请求将失效。相反，如果无限制地扩大保护部分，终止处理将永远不会结束。保护范围应保持简短，并将关闭失败作为可观测的错误保留下来。

如果使用同步数据库调用，即使 Python 级别的 await 取消未介入，查询也可能长时间阻塞事件循环。在这种情况下，也需要数据库连接·查询超时和进程终止策略。

5. 发送完成的事实与存储完成的事实

以下顺序会产生失败区间。

완료 이벤트 전송 → DB 커밋

如果进程在发送事件后立即终止，用户看到了完成，但刷新后的界面可能显示为正在执行。

如果必须能够重新查询终止结果，则以下顺序更容易处理。

결과·종결 상태 저장 → 커밋 → 완료 이벤트 전송

在这种顺序下，提交后也可能断开连接。但是，可以重新查询已保存的结果。这是将失败区间转移到可以恢复传递失败的方向。

使用 SQLAlchemy 时，必须区分 flush() 和 commit()。flush() 是将更改 SQL 反映到当前事务的阶段。这并不意味着该更改已最终提交。仅凭在同一会话中看到更新的对象，就不能结束存储验证。SQLAlchemy 会话与事务文档

验证应在另一个连接的新事务中执行。即使生产者会话仍处于打开状态，也必须能够查询到终止状态、结束时间和部分响应。由于长时间打开的读事务可能会根据隔离级别看到之前的快照，因此测试的读边界也必须明确。

还需要注意将长时间的外部调用包裹在单个数据库事务中的设计。如果将执行开始、进度检查点和结束记录为短事务，则可以减少连接·锁占用时间，并更容易建立进程中断后的恢复标准。

6. 在多工作进程环境中，甚至必须存储终止权限

如果增加一个执行者，仅靠局部 finalized 变量是不够的。

워커 A: 작업 인수 → 긴 지연
복구기: A의 리스 만료 판단
워커 B: 같은 작업 재인수 → 완료 저장
워커 A: 뒤늦게 복귀 → 오래된 결과 저장 시도

不应通过“谁最后写入”来解决此问题。必须确保只有拥有当前执行权限的尝试才能进行存储。

一种设计是在执行记录中保留以下值。

作业重新获取是原子性执行的，并递增代数编号。执行者使用接收到的编号进行存储。例如，终止更新具有以下条件。SQL 是使用绑定变量表示法的概念示例。

UPDATE execution_run
SET state = :terminal_state,
    finished_at = CURRENT_TIMESTAMP,
    result_ref = :result_ref
WHERE run_id = :run_id
  AND execution_epoch = :my_epoch
  AND state = 'running'
  AND lease_until > CURRENT_TIMESTAMP;

如果更新计数为 0，则意味着已失去终止权限或状态已更改。该执行者不应记录新的终止通知或后续结果指针。进度事件和检查点也需要相同的权限规则。如果先保存了文件，则应使用按尝试划分的存储路径，并且最好在权限验证成功的事务中连接最终结果指针。

在租约续期和过期判断中，应尽可能一致地使用数据库时间。这可以减少因进程间时钟差异导致执行所有权动摇的情况。代数编号起到栅栏令牌的作用，防止旧执行者的写入。

然而，此 SQL 并不能保证外部副作用仅执行一次。如果 A 在成功外部请求后、在保存结果前停止，则 B 的重试可能会产生相同的效果。必须单独设计外部 API 的幂等性键、执行效果前后的阶段记录、结果查询或补偿作业。幂等性键也应与每次重试都会改变的代数编号分开，并且对于相同的逻辑效果应使用相同的键。

7. 将需要继续执行的作业与订阅连接分离

如果即使关闭标签页也需要继续执行，则可以分离以下三条路径。

명령: 실행 요청 → 요청 식별자 중복 검사 → 작업 저장
실행: 작업 인수 → 단계 실행 → 상태·결과·이벤트 저장
관측: 저장된 진행 내역 조회 + 실시간 이벤트 구독

这里，仅添加 create_task() 调用可以创建进程内部的并发执行。但是，进程重启后的作业恢复、执行获取、重试标准等不会自动生成。

重新打开订阅时，也必须能够恢复自上次接收事件之后的内容。SSE 具有标准的 id 字段以及在重连时发送 Last-Event-ID 的标准操作。但是，服务器端的事件存储和重放必须由应用程序实现。此外，在非浏览器默认 EventSource 的 fetch 基础流客户端中，必须直接实现重连和光标传递。WHATWG 服务器发送事件标准

如果设计持久化事件，则可以基于 (run_id, sequence) 消除重复，并恢复同一执行内的顺序。如果同时存储尝试编号，则更容易区分重试前后的记录。但是，分配递增的数字并不能保证提交顺序。如果有多个写入者，则需要考虑按执行序列化或提交顺序的光标规则。

并非必须将所有令牌保留为持久化事件。可以选择存储步骤的开始·结束·结果，并将令牌合并为固定大小的块或周期性快照。需要一起决定必要的重放精度和写入成本。

8. 完成通知应从已保存的状态派生

在保存完成状态和发送通知之间也存在失败区间。

• 保存状态后、发送通知前发生故障：已完成但没有通知。
• 发送通知后、保存传递确认前发生故障：在重试过程中同一通知重复发送。

一种方法是在更改终止状态的事务中同时保存用于发送通知的记录。单独的传递者读取此记录，发送通知并记录传递结果。仅当终止更新通过前面的所有权条件时，才应创建通知记录。

由于传输本身可能重复，因此通知中应包含稳定的 event_id，并在接收端消除重复。创建同一执行终止通知的唯一性约束也有帮助。这种方式可以一起保留状态和通知意图，但并不保证对外部通知渠道进行恰好一次传递。

界面的通知文案也必须反映此契约。“完成”通知必须能够打开已保存的结果，“停止”不应被解释为外部系统的所有更改均已取消。作业进行中已发生的效果和响应生成中断是截然不同的事实。

9. 测试终止边界比测试正常响应更重要

如果将验证单位设为 HTTP 200，则很容易忽略这些问题。更有效的方法是将注入失败的位置和之后必须成立的状态作为一对记录下来。

在本文的最小复现中，直接验证了第一个块后立即关闭、通过 aclosing() 的立即清理以及等待期间任务取消的传播。在单独使用数据库的集成验证中，确认了在关闭流后立即通过独立连接查询到停止状态、部分响应和结束时间已提交。租约重新获取和通知重传项目是在实际实现提议的设计时需要添加的失败注入场景。

仅凭平均响应时间的运营指标是不够的。租约过期的执行数、终止记录失败数、终止状态与通知传递之间的时间差、旧执行代数写入被拒绝的次数等值，显示了恢复路径的健康状况。

总结

在流式系统中，终止由多个步骤组成：停止消费、清理上游资源、确定执行状态，并将此事实传达给用户。

在设计审查中，值得首先确认以下三个问题。

1. 负责结束此作业的是谁？
2. 即使该执行者消失，是否留下了供下一个执行者判断的记录？
3. 重新查询发送给用户的最终状态时，是否能获得相同的结果？

如果这三点明确，取消就不会被隐藏在单行异常处理中。它将成为一个可以与正常完成同等程度验证的执行协议。