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() |