383 lines
16 KiB
Python
383 lines
16 KiB
Python
|
# ============================================================
|
|||
|
|
# round-39 方向 P：Saga/补偿编排框架黄金测试
|
|||
|
|
# 固化：状态机 / 幂等 / 重试 / 超时 / 补偿链 / 人工接管 / 持久化恢复
|
|||
|
|
# 矩阵 78：任一步失败可重试或补偿；中间态可见；人工接管点和审计记录明确
|
|||
|
|
# 矩阵 119：每个 P3 动作定义幂等键、超时、重试、补偿、人工接管和审计事件
|
|||
|
|
# ============================================================
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import copy
|
|||
|
|
from pathlib import Path
|
|||
|
|
|
|||
|
|
from fastapi.testclient import TestClient
|
|||
|
|
|
|||
|
|
from server.aps_domain import saga
|
|||
|
|
from server.state.checkpoints import CheckpointStore
|
|||
|
|
from server.state.seed import seed_world
|
|||
|
|
|
|||
|
|
|
|||
|
|
class _MemStore:
|
|||
|
|
def __init__(self, data=None, checkpoint_path: Path | None = None):
|
|||
|
|
self.data = data if data is not None else seed_world()
|
|||
|
|
self.checkpoints = CheckpointStore(str(checkpoint_path or Path(__file__).parent / "_saga_cp.json"))
|
|||
|
|
|
|||
|
|
def next_id(self, kind: str) -> int:
|
|||
|
|
key = f"_c_{kind}"
|
|||
|
|
self.data[key] = self.data.get(key, 1000) + 1
|
|||
|
|
return self.data[key]
|
|||
|
|
|
|||
|
|
def save(self) -> None:
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
|
|||
|
|
def make_action(name: str, log: list[str], *, fails: int = 0, timeout: bool = False):
|
|||
|
|
state = {"calls": 0}
|
|||
|
|
|
|||
|
|
def fn(ctx) -> dict:
|
|||
|
|
state["calls"] += 1
|
|||
|
|
log.append(name)
|
|||
|
|
if timeout:
|
|||
|
|
raise saga.SagaStepTimeout("simulated timeout")
|
|||
|
|
if state["calls"] <= fails:
|
|||
|
|
raise saga.SagaStepFailed("simulated failure")
|
|||
|
|
return {"done": name, "calls": state["calls"]}
|
|||
|
|
|
|||
|
|
return fn
|
|||
|
|
|
|||
|
|
|
|||
|
|
def make_registry(log: list[str], *, s2_fails: int = 0, s3_fails: int = 0,
|
|||
|
|
timeout: bool = False, comp_fails: bool = False) -> dict:
|
|||
|
|
reg = {
|
|||
|
|
"s1": {"fn": make_action("s1", log), "compensation": "c1"},
|
|||
|
|
"c1": {"fn": make_action("c1", log)},
|
|||
|
|
"s2": {"fn": make_action("s2", log, fails=s2_fails), "compensation": "c2"},
|
|||
|
|
"c2": {"fn": make_action("c2", log)},
|
|||
|
|
"s3": {"fn": make_action("s3", log, fails=s3_fails, timeout=timeout), "compensation": "c3"},
|
|||
|
|
"c3": {"fn": make_action("c3", log)},
|
|||
|
|
}
|
|||
|
|
if comp_fails:
|
|||
|
|
def broken_c1(_ctx):
|
|||
|
|
log.append("c1")
|
|||
|
|
raise saga.SagaCompensationError("simulated compensation failure")
|
|||
|
|
reg["c1"]["fn"] = broken_c1
|
|||
|
|
return reg
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _demo_steps() -> list[saga.SagaStep]:
|
|||
|
|
return [
|
|||
|
|
saga.SagaStep("a", "s1", idem_key="k1", compensation_action="c1"),
|
|||
|
|
saga.SagaStep("b", "s2", idem_key="k2", max_retries=2, compensation_action="c2"),
|
|||
|
|
saga.SagaStep("c", "s3", idem_key="k3", compensation_action="c3"),
|
|||
|
|
]
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ------------------------------------------------------------
|
|||
|
|
# ① 创建：记录结构 + 随世界状态持久化
|
|||
|
|
# ------------------------------------------------------------
|
|||
|
|
def test_saga_record_created_with_steps_and_persisted():
|
|||
|
|
store = _MemStore()
|
|||
|
|
coord = saga.SagaCoordinator(store, actor="test")
|
|||
|
|
rec = coord.create_saga("demo", _demo_steps(), context={"x": 1})
|
|||
|
|
assert rec["status"] == "PENDING"
|
|||
|
|
assert rec["steps"][0]["status"] == "PENDING"
|
|||
|
|
assert rec["steps"][0]["idemKey"] == "k1"
|
|||
|
|
assert rec["steps"][0]["timeoutSec"] == 30.0
|
|||
|
|
assert rec["steps"][0]["maxRetries"] == 2
|
|||
|
|
assert rec["steps"][0]["compensationAction"] == "c1"
|
|||
|
|
assert rec["auditRefs"] == [f"saga:{rec['id']}"]
|
|||
|
|
# 随世界状态持久化
|
|||
|
|
assert store.data["sagas"][0]["id"] == rec["id"]
|
|||
|
|
assert any(e["action"] == "saga.created" for e in store.data["auditEvents"])
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_saga_dedupe_key_reuses_existing_record():
|
|||
|
|
store = _MemStore()
|
|||
|
|
coord = saga.SagaCoordinator(store, actor="test")
|
|||
|
|
r1 = coord.create_saga("demo", _demo_steps(), dedupe_key="EV-1")
|
|||
|
|
r2 = coord.create_saga("demo", _demo_steps(), dedupe_key="EV-1")
|
|||
|
|
assert r2["id"] == r1["id"]
|
|||
|
|
assert len(store.data["sagas"]) == 1
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ------------------------------------------------------------
|
|||
|
|
# ② 全链成功 + 审计（含 idemKey/status/compensation）
|
|||
|
|
# ------------------------------------------------------------
|
|||
|
|
def test_saga_runs_all_steps_to_succeeded_with_audit():
|
|||
|
|
store = _MemStore()
|
|||
|
|
log: list[str] = []
|
|||
|
|
coord = saga.SagaCoordinator(store, registry=make_registry(log), actor="test")
|
|||
|
|
rec = coord.create_saga("demo", _demo_steps())
|
|||
|
|
out = coord.run(rec["id"])
|
|||
|
|
assert out["status"] == "SUCCEEDED"
|
|||
|
|
assert [s["status"] for s in out["steps"]] == ["SUCCEEDED"] * 3
|
|||
|
|
assert log == ["s1", "s2", "s3"]
|
|||
|
|
# 幂等键登记
|
|||
|
|
assert set(store.data["sagaIdem"]) == {"k1", "k2", "k3"}
|
|||
|
|
actions = [e["action"] for e in store.data["auditEvents"]]
|
|||
|
|
assert actions.count("saga.step.succeeded") == 3
|
|||
|
|
assert "saga.succeeded" in actions
|
|||
|
|
# 每次 step 迁移审计含 idemKey/status/compensation
|
|||
|
|
step_audit = next(e for e in store.data["auditEvents"] if e["action"] == "saga.step.succeeded")
|
|||
|
|
assert step_audit["rationale"]["idemKey"]
|
|||
|
|
assert step_audit["rationale"]["status"] == "SUCCEEDED"
|
|||
|
|
assert "compensation" in step_audit["rationale"]
|
|||
|
|
assert f"saga.step:{rec['id']}:a" in step_audit["evidenceRefs"]
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ------------------------------------------------------------
|
|||
|
|
# ③ 重试：瞬时失败重试成功后 saga 完成
|
|||
|
|
# ------------------------------------------------------------
|
|||
|
|
def test_saga_step_retry_after_transient_failure():
|
|||
|
|
store = _MemStore()
|
|||
|
|
log: list[str] = []
|
|||
|
|
coord = saga.SagaCoordinator(store, registry=make_registry(log, s2_fails=2), actor="test")
|
|||
|
|
rec = coord.create_saga("demo", _demo_steps())
|
|||
|
|
out = coord.run(rec["id"])
|
|||
|
|
assert out["status"] == "SUCCEEDED"
|
|||
|
|
b = next(s for s in out["steps"] if s["name"] == "b")
|
|||
|
|
assert b["attempts"] == 3 # 1 次初始 + 2 次重试
|
|||
|
|
assert b["error"] is None
|
|||
|
|
assert "saga.step.retry" in [e["action"] for e in store.data["auditEvents"]]
|
|||
|
|
assert "saga.step.timeout" not in [e["action"] for e in store.data["auditEvents"]]
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ------------------------------------------------------------
|
|||
|
|
# ④ 失败超限 → 自动补偿链（逆序）
|
|||
|
|
# ------------------------------------------------------------
|
|||
|
|
def test_saga_exhausts_retries_then_compensates_in_reverse():
|
|||
|
|
store = _MemStore()
|
|||
|
|
log: list[str] = []
|
|||
|
|
coord = saga.SagaCoordinator(store, registry=make_registry(log, s3_fails=99), actor="test")
|
|||
|
|
rec = coord.create_saga("demo", _demo_steps())
|
|||
|
|
out = coord.run(rec["id"])
|
|||
|
|
assert out["status"] == "COMPENSATED"
|
|||
|
|
assert out["compensationReason"].startswith("step:c")
|
|||
|
|
c = next(s for s in out["steps"] if s["name"] == "c")
|
|||
|
|
assert c["status"] == "FAILED"
|
|||
|
|
assert c["compensation"]["status"] == "SUCCEEDED"
|
|||
|
|
# 逆序补偿：c3（失败步自身的部分效果）→ c2 → c1（s3 重试预算=2 → 3 次尝试）
|
|||
|
|
assert log == ["s1", "s2", "s3", "s3", "s3", "c3", "c2", "c1"]
|
|||
|
|
actions = [e["action"] for e in store.data["auditEvents"]]
|
|||
|
|
assert "saga.compensating" in actions and "saga.compensated" in actions
|
|||
|
|
assert "saga.step.compensated" in actions
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ------------------------------------------------------------
|
|||
|
|
# ⑤ 超时：动作主动超时 / 真实时钟超时 → 重试 → 补偿
|
|||
|
|
# ------------------------------------------------------------
|
|||
|
|
def test_saga_timeout_triggers_retry_then_compensation():
|
|||
|
|
store = _MemStore()
|
|||
|
|
log: list[str] = []
|
|||
|
|
coord = saga.SagaCoordinator(store, registry=make_registry(log, timeout=True), actor="test")
|
|||
|
|
rec = coord.create_saga("demo", _demo_steps())
|
|||
|
|
out = coord.run(rec["id"])
|
|||
|
|
assert out["status"] == "COMPENSATED"
|
|||
|
|
c = next(s for s in out["steps"] if s["name"] == "c")
|
|||
|
|
assert c["timeouts"] == 3 # 每次尝试都超时
|
|||
|
|
assert c["error"].startswith("timeout:")
|
|||
|
|
assert "saga.step.timeout" in [e["action"] for e in store.data["auditEvents"]]
|
|||
|
|
|
|||
|
|
|
|||
|
|
class _FakeClock:
|
|||
|
|
def __init__(self) -> None:
|
|||
|
|
self.t = 0.0
|
|||
|
|
|
|||
|
|
def __call__(self) -> float:
|
|||
|
|
return self.t
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_saga_clock_timeout_detection():
|
|||
|
|
store = _MemStore()
|
|||
|
|
log: list[str] = []
|
|||
|
|
clock = _FakeClock()
|
|||
|
|
reg = {
|
|||
|
|
"s1": {"fn": make_action("s1", log), "compensation": None},
|
|||
|
|
"slow": {"fn": _slow_action(log, clock), "compensation": "slow-comp"},
|
|||
|
|
"slow-comp": {"fn": make_action("slow-comp", log)},
|
|||
|
|
}
|
|||
|
|
coord = saga.SagaCoordinator(store, registry=reg, clock=clock, actor="test")
|
|||
|
|
rec = coord.create_saga("demo", [
|
|||
|
|
saga.SagaStep("a", "s1", idem_key="k1"),
|
|||
|
|
saga.SagaStep("b", "slow", idem_key="k2", timeout_sec=1.0, max_retries=1,
|
|||
|
|
compensation_action="slow-comp"),
|
|||
|
|
])
|
|||
|
|
out = coord.run(rec["id"])
|
|||
|
|
assert out["status"] == "COMPENSATED"
|
|||
|
|
b = next(s for s in out["steps"] if s["name"] == "b")
|
|||
|
|
assert b["timeouts"] == 2
|
|||
|
|
assert b["error"].startswith("timeout:")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _slow_action(log: list[str], clock: _FakeClock):
|
|||
|
|
def fn(_ctx) -> dict:
|
|||
|
|
log.append("slow")
|
|||
|
|
clock.t += 100.0 # 模拟耗时超过 timeout_sec
|
|||
|
|
return {"done": "slow"}
|
|||
|
|
return fn
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ------------------------------------------------------------
|
|||
|
|
# ⑥ 补偿失败 → MANUAL_TAKEOVER（人工接管点）
|
|||
|
|
# ------------------------------------------------------------
|
|||
|
|
def test_saga_compensation_failure_leads_to_manual_takeover():
|
|||
|
|
store = _MemStore()
|
|||
|
|
log: list[str] = []
|
|||
|
|
coord = saga.SagaCoordinator(store, registry=make_registry(log, s2_fails=99, comp_fails=True),
|
|||
|
|
actor="test")
|
|||
|
|
rec = coord.create_saga("demo", [
|
|||
|
|
saga.SagaStep("a", "s1", idem_key="k1", compensation_action="c1"),
|
|||
|
|
saga.SagaStep("b", "s2", idem_key="k2", max_retries=1), # 无补偿定义
|
|||
|
|
])
|
|||
|
|
out = coord.run(rec["id"])
|
|||
|
|
assert out["status"] == "MANUAL_TAKEOVER"
|
|||
|
|
assert out["manualTakeover"] is True
|
|||
|
|
assert "compensation-failed" in out["takeoverReason"]
|
|||
|
|
a = next(s for s in out["steps"] if s["name"] == "a")
|
|||
|
|
assert a["compensation"]["status"] == "FAILED"
|
|||
|
|
actions = [e["action"] for e in store.data["auditEvents"]]
|
|||
|
|
assert "saga.manual_takeover" in actions
|
|||
|
|
assert "saga.step.compensation_failed" in actions
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_saga_takeover_blocks_auto_run_then_retry_resumes():
|
|||
|
|
store = _MemStore()
|
|||
|
|
log: list[str] = []
|
|||
|
|
coord = saga.SagaCoordinator(store, registry=make_registry(log), actor="test")
|
|||
|
|
rec = coord.create_saga("demo", _demo_steps())
|
|||
|
|
out = coord.takeover(rec["id"], reason="operator pause")
|
|||
|
|
assert out["status"] == "MANUAL_TAKEOVER"
|
|||
|
|
assert out["manualTakeover"] is True
|
|||
|
|
assert out["takeoverReason"] == "operator pause"
|
|||
|
|
# 接管后自动 run 被阻断
|
|||
|
|
out2 = coord.run(rec["id"])
|
|||
|
|
assert out2["status"] == "MANUAL_TAKEOVER"
|
|||
|
|
assert log == [] # 没有任何步骤执行
|
|||
|
|
# 人工接管后显式 retry → 恢复自动化
|
|||
|
|
out3 = coord.retry(rec["id"])
|
|||
|
|
assert out3["status"] == "SUCCEEDED"
|
|||
|
|
assert out3["manualTakeover"] is False
|
|||
|
|
assert log == ["s1", "s2", "s3"]
|
|||
|
|
assert "saga.retry" in [e["action"] for e in store.data["auditEvents"]]
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ------------------------------------------------------------
|
|||
|
|
# ⑦ 恢复队列：中断的 saga 可 resume/replay
|
|||
|
|
# ------------------------------------------------------------
|
|||
|
|
def test_saga_persistence_resume_after_restart():
|
|||
|
|
store = _MemStore()
|
|||
|
|
log: list[str] = []
|
|||
|
|
coord = saga.SagaCoordinator(store, registry=make_registry(log), actor="test")
|
|||
|
|
rec = coord.create_saga("demo", _demo_steps())
|
|||
|
|
# 模拟进程中断：步骤 b 执行中崩溃（status=RUNNING，无 finishedAt）
|
|||
|
|
rec["status"] = "RUNNING"
|
|||
|
|
rec["steps"][0]["status"] = "SUCCEEDED"
|
|||
|
|
rec["steps"][0]["result"] = {"done": "s1"}
|
|||
|
|
rec["steps"][1]["status"] = "RUNNING"
|
|||
|
|
rec["steps"][1]["startedAt"] = "2026-08-02 10:00:00"
|
|||
|
|
store.save()
|
|||
|
|
# 重启：世界数据（含 sagas/sagaIdem）整体重建
|
|||
|
|
store2 = _MemStore(copy.deepcopy(store.data))
|
|||
|
|
log2: list[str] = []
|
|||
|
|
coord2 = saga.SagaCoordinator(store2, registry=make_registry(log2), actor="test")
|
|||
|
|
out = coord2.resume_all()
|
|||
|
|
assert len(out) == 1
|
|||
|
|
assert out[0]["status"] == "SUCCEEDED"
|
|||
|
|
assert log2 == ["s2", "s3"] # 已成功步骤不重放
|
|||
|
|
assert all(s["status"] == "SUCCEEDED" for s in out[0]["steps"])
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ------------------------------------------------------------
|
|||
|
|
# ⑧ 幂等重放：同 idem_key 已成功 → 复用记录，不重复执行外部效果
|
|||
|
|
# ------------------------------------------------------------
|
|||
|
|
def test_saga_idempotent_replay_reuses_recorded_result():
|
|||
|
|
store = _MemStore()
|
|||
|
|
store.data.setdefault("sagaIdem", {})["k1"] = {
|
|||
|
|
"sagaId": "saga-other", "step": "a", "result": {"replayed": True}, "at": "2026-08-02 10:00:00"}
|
|||
|
|
log: list[str] = []
|
|||
|
|
coord = saga.SagaCoordinator(store, registry=make_registry(log), actor="test")
|
|||
|
|
rec = coord.create_saga("demo", _demo_steps())
|
|||
|
|
out = coord.run(rec["id"])
|
|||
|
|
assert out["status"] == "SUCCEEDED"
|
|||
|
|
a = next(s for s in out["steps"] if s["name"] == "a")
|
|||
|
|
assert a["result"]["replayed"] is True
|
|||
|
|
assert log == ["s2", "s3"] # s1 未重新执行
|
|||
|
|
assert "saga.step.idempotent_replayed" in [e["action"] for e in store.data["auditEvents"]]
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ------------------------------------------------------------
|
|||
|
|
# ⑨ 中间态可见：gate 步骤停在 WAITING_HUMAN，恢复后继续
|
|||
|
|
# ------------------------------------------------------------
|
|||
|
|
def test_saga_gate_pauses_waits_human_and_resumes():
|
|||
|
|
store = _MemStore()
|
|||
|
|
log: list[str] = []
|
|||
|
|
coord = saga.SagaCoordinator(store, registry=make_registry(log), actor="test")
|
|||
|
|
rec = coord.create_saga("demo", [
|
|||
|
|
saga.SagaStep("a", "s1", idem_key="k1", gate=True),
|
|||
|
|
saga.SagaStep("b", "s2", idem_key="k2"),
|
|||
|
|
])
|
|||
|
|
out = coord.run(rec["id"], auto_approve=False)
|
|||
|
|
assert out["status"] == "WAITING_HUMAN"
|
|||
|
|
assert out["steps"][0]["status"] == "WAITING_HUMAN"
|
|||
|
|
assert out["steps"][1]["status"] == "PENDING"
|
|||
|
|
assert "saga.step.waiting_human" in [e["action"] for e in store.data["auditEvents"]]
|
|||
|
|
# 中间态可见：详情查询可看到门禁点
|
|||
|
|
detail = coord.get(rec["id"])
|
|||
|
|
assert detail["status"] == "WAITING_HUMAN"
|
|||
|
|
# 人工放行后继续
|
|||
|
|
out2 = coord.run(rec["id"], auto_approve=True)
|
|||
|
|
assert out2["status"] == "SUCCEEDED"
|
|||
|
|
assert log == ["s1", "s2"]
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ------------------------------------------------------------
|
|||
|
|
# ⑩ 网关 API：GET /api/saga、详情、retry/compensate/takeover
|
|||
|
|
# ------------------------------------------------------------
|
|||
|
|
def test_saga_gateway_api_endpoints(monkeypatch):
|
|||
|
|
import server.aps_domain.saga as saga_module
|
|||
|
|
import server.gateway.app as gateway_module
|
|||
|
|
from tests.auth_provider import install_test_auth
|
|||
|
|
|
|||
|
|
store = _MemStore()
|
|||
|
|
log: list[str] = []
|
|||
|
|
coord = saga.SagaCoordinator(store, registry=make_registry(log), actor="web")
|
|||
|
|
monkeypatch.setattr(gateway_module, "get_store", lambda: store)
|
|||
|
|
monkeypatch.setattr(saga_module, "get_coordinator", lambda _store, **kw: coord)
|
|||
|
|
install_test_auth(monkeypatch, "tenant-saga-test")
|
|||
|
|
|
|||
|
|
rec = coord.create_saga("api-demo", _demo_steps())
|
|||
|
|
client = TestClient(gateway_module.create_app())
|
|||
|
|
login = client.post("/api/auth/login", json={"username": "planner"})
|
|||
|
|
assert login.status_code == 200
|
|||
|
|
|
|||
|
|
# 人工接管
|
|||
|
|
r = client.post(f"/api/saga/{rec['id']}/takeover", json={"reason": "ops"})
|
|||
|
|
assert r.status_code == 200
|
|||
|
|
assert r.json()["status"] == "MANUAL_TAKEOVER"
|
|||
|
|
assert r.json()["manualTakeover"] is True
|
|||
|
|
|
|||
|
|
# 重试恢复自动化
|
|||
|
|
r = client.post(f"/api/saga/{rec['id']}/retry", json={})
|
|||
|
|
assert r.status_code == 200
|
|||
|
|
assert r.json()["status"] == "SUCCEEDED"
|
|||
|
|
|
|||
|
|
# 列表（含中间态 steps 明细）
|
|||
|
|
r = client.get("/api/saga")
|
|||
|
|
assert r.status_code == 200
|
|||
|
|
assert r.json()["count"] >= 1
|
|||
|
|
assert any(s["id"] == rec["id"] for s in r.json()["sagas"])
|
|||
|
|
|
|||
|
|
# 详情
|
|||
|
|
r = client.get(f"/api/saga/{rec['id']}")
|
|||
|
|
assert r.status_code == 200
|
|||
|
|
assert r.json()["id"] == rec["id"]
|
|||
|
|
assert [s["name"] for s in r.json()["steps"]] == ["a", "b", "c"]
|
|||
|
|
|
|||
|
|
# 手动补偿
|
|||
|
|
r = client.post(f"/api/saga/{rec['id']}/compensate", json={"reason": "manual"})
|
|||
|
|
assert r.status_code == 200
|
|||
|
|
assert r.json()["status"] == "COMPENSATED"
|
|||
|
|
|
|||
|
|
# 不存在 → error 语义
|
|||
|
|
r = client.get("/api/saga/saga-nope")
|
|||
|
|
assert r.status_code == 200
|
|||
|
|
assert "error" in r.json()
|