5fb7bfb1de
- 伏笔账本：纯函数状态机(OPEN/PARTIAL/CLOSED/OVERDUE) + ForeshadowLedger repo；验收后到期扫描(BackgroundTask 自建 session 置 OVERDUE)；登记/状态变更端点 - 节奏 + 三审齐：foreshadow-analyst + pace-checker 并入 LangGraph 并行审(REVIEW_SPECS)，collect 分列落 chapter_reviews(conflicts/foreshadow_sug/pace)，review SSE 加 foreshadow/pace 事件 - 大纲：outliner Agent 产 OutlineResult(含 foreshadow_windows)，POST /outline 逐章 upsert outline 表；GET /foreshadow?status= 看板 - 前端：伏笔四泳道看板(OVERDUE 琥珀) + 大纲编辑器(窗口徽标) + 节奏节拍图(▁▃▅) + 审稿页消费 foreshadow/pace SSE - bugfix(T3.8)：并行三审共用请求 session 记账触发 'Session is already flushing' → foreshadow/pace 静默丢失；SqlAlchemyLedgerSink.record 改 add-only(靠端点/事务 commit)，加并发回归测试 - M3 E2E：真实 DB + mock 网关零 token 走通 埋设→进展→验收后扫描 OVERDUE→看板 + 大纲含窗口 + 三审齐 SSE/留痕；E2E 暴露并钉住上述 bug - 门禁绿：mypy 111 / pytest 228(0 xfailed) / alembic 无漂移；前端 gen:api/lint/tsc/vitest 69/build
113 lines
4.4 KiB
Python
113 lines
4.4 KiB
Python
"""T3.8 回归：并行审记账并发安全（add-only，无 flush 重入）。
|
||
|
||
旧 bug（T3.7 真 pg E2E 暴露）：三审作为同一 LangGraph superstep 的并行分支，各自
|
||
`gateway.run()` → 共用**同一请求 session** 的 `SqlAlchemyLedgerSink.record`
|
||
(`session.add` + `await session.flush()`)。`AsyncSession` 非并发安全：`await flush()`
|
||
是并行路径里唯一让出控制权的 DB-IO，第二/三审的 flush 撞上「Session is already
|
||
flushing」（flush 重入）→ 被 `run_review` 失败隔离吞成 `incomplete` → foreshadow/pace
|
||
结果静默丢失。
|
||
|
||
本测试用一个会**侦测 flush 重入**的假 AsyncSession：`flush` 异步（`await asyncio.sleep`
|
||
让出控制权），若两协程同时在 flush 中即抛 `RuntimeError("Session is already flushing")`
|
||
——精确复刻 SQLAlchemy 的报错。在旧 add-then-flush 实现下，`asyncio.gather` 三并发
|
||
`record` 会触发该重入；修复后 `record` 只 `add`（同步、不让出），不再调 flush，故三行
|
||
全部入 session 且零重入。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import uuid
|
||
|
||
import pytest
|
||
from ww_llm_gateway.ledger import SqlAlchemyLedgerSink
|
||
from ww_llm_gateway.types import Scope, Usage
|
||
|
||
|
||
class _FlushReentryDetectingSession:
|
||
"""假 AsyncSession：记录 `add` 的行；`flush` 异步且侦测并发重入。
|
||
|
||
`add` 同步追加（镜像真 session）。`flush` 进入时若已有协程在 flush 中 → 抛
|
||
`RuntimeError("Session is already flushing")`（复刻 SQLAlchemy 行为），否则
|
||
`await asyncio.sleep(0)` 让出控制权模拟真实 DB-IO 让步点。
|
||
"""
|
||
|
||
def __init__(self) -> None:
|
||
self.added: list[object] = []
|
||
self.flush_calls = 0
|
||
self._flushing = False
|
||
|
||
def add(self, row: object) -> None:
|
||
self.added.append(row)
|
||
|
||
async def flush(self) -> None:
|
||
self.flush_calls += 1
|
||
if self._flushing:
|
||
raise RuntimeError("Session is already flushing")
|
||
self._flushing = True
|
||
try:
|
||
await asyncio.sleep(0) # 让出控制权——并行协程在此交错
|
||
finally:
|
||
self._flushing = False
|
||
|
||
|
||
def _usage(input_tokens: int) -> Usage:
|
||
return Usage(
|
||
provider="deepseek",
|
||
model="deepseek-chat",
|
||
input_tokens=input_tokens,
|
||
output_tokens=1,
|
||
cost_minor=0,
|
||
currency="USD",
|
||
)
|
||
|
||
|
||
def _scope() -> Scope:
|
||
return Scope(user_id=uuid.UUID(int=1), project_id=uuid.UUID(int=2))
|
||
|
||
|
||
async def test_concurrent_record_adds_all_rows_without_flush_reentry() -> None:
|
||
"""三并发 `record`（模拟三审并行）→ 三行全部入 session，绝不抛 flush 重入。
|
||
|
||
旧 add-then-flush 实现在此会因 `await flush()` 让出控制权而 flush 重入抛错；
|
||
add-only 实现下 `record` 不让出控制权，三行确定性入 session。
|
||
"""
|
||
session = _FlushReentryDetectingSession()
|
||
sink = SqlAlchemyLedgerSink(session) # type: ignore[arg-type]
|
||
|
||
await asyncio.gather(
|
||
sink.record(_scope(), _usage(100)),
|
||
sink.record(_scope(), _usage(200)),
|
||
sink.record(_scope(), _usage(300)),
|
||
)
|
||
|
||
# 三审用量全部入 session（一次调用一行，记账不丢）
|
||
assert len(session.added) == 3
|
||
input_tokens = sorted(row.input_tokens for row in session.added) # type: ignore[attr-defined]
|
||
assert input_tokens == [100, 200, 300]
|
||
# add-only：sink 不在并行路径 flush（持久化靠端点/事务 commit）
|
||
assert session.flush_calls == 0
|
||
|
||
|
||
async def test_record_does_not_flush_so_caller_commit_persists() -> None:
|
||
"""单次 `record` 只 add 不 flush——契约「网关 ledger 只 add、调用方负责 commit」。"""
|
||
session = _FlushReentryDetectingSession()
|
||
sink = SqlAlchemyLedgerSink(session) # type: ignore[arg-type]
|
||
|
||
await sink.record(_scope(), _usage(42))
|
||
|
||
assert len(session.added) == 1
|
||
assert session.flush_calls == 0
|
||
|
||
|
||
@pytest.mark.parametrize("n_concurrent", [2, 5, 10])
|
||
async def test_concurrent_record_scales_without_loss(n_concurrent: int) -> None:
|
||
"""N 并发 `record` 全部落 session，无丢失、无 flush 重入（不论并发度）。"""
|
||
session = _FlushReentryDetectingSession()
|
||
sink = SqlAlchemyLedgerSink(session) # type: ignore[arg-type]
|
||
|
||
await asyncio.gather(*(sink.record(_scope(), _usage(i + 1)) for i in range(n_concurrent)))
|
||
|
||
assert len(session.added) == n_concurrent
|
||
assert session.flush_calls == 0
|