321 lines
13 KiB
Python
321 lines
13 KiB
Python
"""P1 concurrency + single-transaction launch tests for MAF roundtable.
|
|
|
|
Covers:
|
|
- **Concurrent run launch**: ``asyncio.gather`` of ≥ ``create_run`` calls must
|
|
yield exactly one active run; the rest get ``MafConflictError``.
|
|
- **Single-transaction atomicity**: flush / pre-commit failures roll back fully —
|
|
session stays at its prior status, no orphan run, no residual ``starting``.
|
|
- **Success shape**: session ``running``, unique run ``pending``.
|
|
- **Cancel then re-launch** works.
|
|
- **Concurrent approval decision**: parallel approve + reject →exactly one winner.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
|
|
|
from app.maf_roundtable.service import (
|
|
MafConflictError,
|
|
)
|
|
from deerflow.persistence.base import Base
|
|
from deerflow.persistence.maf_roundtable import make_maf_roundtable_store
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def store(tmp_path):
|
|
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'maf_conc.db'}")
|
|
async with engine.begin() as conn:
|
|
import deerflow.persistence.maf_roundtable.model # noqa: F401
|
|
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
sf = async_sessionmaker(engine, expire_on_commit=False)
|
|
yield make_maf_roundtable_store(sf)
|
|
await engine.dispose()
|
|
|
|
|
|
def _svc(store, *, enabled=True, admin_diagnostic_access=True):
|
|
from maf_test_helpers import make_maf_service
|
|
|
|
return make_maf_service(store, enabled=enabled, admin_diagnostic_access=admin_diagnostic_access)
|
|
|
|
|
|
# ── P1: concurrent run launch (single-tx CAS) ─────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_concurrent_create_run_only_one_wins(store):
|
|
"""8 parallel ``create_run`` calls →exactly one active run, 7 conflicts."""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
|
|
async def _attempt() -> tuple[str, object]:
|
|
try:
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
return ("ok", run)
|
|
except MafConflictError as exc:
|
|
return ("conflict", exc)
|
|
|
|
results = await asyncio.gather(*[_attempt() for _ in range(8)])
|
|
oks = [r for r in results if r[0] == "ok"]
|
|
conflicts = [r for r in results if r[0] == "conflict"]
|
|
|
|
assert len(oks) == 1, f"expected exactly one winner, got {len(oks)}"
|
|
assert len(conflicts) == 7
|
|
|
|
runs = await svc.list_runs(session_id=s["id"], user_id="u1")
|
|
active = [r for r in runs if r["status"] in ("pending", "running", "waiting_approval", "waiting_input")]
|
|
assert len(active) == 1
|
|
assert active[0]["id"] == oks[0][1]["id"]
|
|
assert active[0]["status"] == "pending"
|
|
|
|
refreshed = await svc.get_session(session_id=s["id"], user_id="u1")
|
|
assert refreshed["status"] == "running"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_run_success_session_running_run_pending(store):
|
|
svc = _svc(store)
|
|
s = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
assert run["status"] == "pending"
|
|
refreshed = await svc.get_session(session_id=s["id"], user_id="u1")
|
|
assert refreshed["status"] == "running"
|
|
assert len(await svc.list_runs(session_id=s["id"], user_id="u1")) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_concurrent_create_run_second_launch_after_cancel(store):
|
|
"""After the winning run is cancelled, a fresh launch succeeds."""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
|
|
results = await asyncio.gather(
|
|
*[svc.create_run(session_id=s["id"], user_id="u1") for _ in range(3)],
|
|
return_exceptions=True,
|
|
)
|
|
winners = [r for r in results if not isinstance(r, Exception)]
|
|
assert len(winners) == 1
|
|
|
|
await svc.cancel_run(run_id=winners[0]["id"], user_id="u1")
|
|
run2 = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
assert run2["status"] == "pending"
|
|
refreshed = await svc.get_session(session_id=s["id"], user_id="u1")
|
|
assert refreshed["status"] == "running"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_flush_failure_rolls_back_session_and_no_run(store):
|
|
"""If flush() fails after the claim UPDATE, the whole transaction rolls back:
|
|
session stays at its prior status and no run row is persisted.
|
|
"""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
prior_status = s["status"]
|
|
assert prior_status == "draft"
|
|
|
|
async def boom_flush(self, *args, **kwargs):
|
|
raise RuntimeError("simulated flush failure")
|
|
|
|
with patch.object(AsyncSession, "flush", boom_flush):
|
|
with pytest.raises(RuntimeError, match="simulated flush failure"):
|
|
await svc.create_run(session_id=s["id"], user_id="u1")
|
|
|
|
refreshed = await svc.get_session(session_id=s["id"], user_id="u1")
|
|
assert refreshed["status"] == prior_status
|
|
assert refreshed["status"] != "starting"
|
|
assert await svc.list_runs(session_id=s["id"], user_id="u1") == []
|
|
|
|
# A subsequent healthy launch still works.
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
assert run["status"] == "pending"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pre_commit_failure_leaves_no_starting_residue(store):
|
|
"""If an error occurs after flush but before commit, the transaction rolls
|
|
back: session must not residual ``starting`` / ``running``, and no run row.
|
|
"""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
|
|
real_flush = AsyncSession.flush
|
|
|
|
async def flush_then_boom(self, *args, **kwargs):
|
|
await real_flush(self, *args, **kwargs)
|
|
raise RuntimeError("simulated post-flush pre-commit failure")
|
|
|
|
with patch.object(AsyncSession, "flush", flush_then_boom):
|
|
with pytest.raises(RuntimeError, match="simulated post-flush pre-commit failure"):
|
|
await svc.create_run(session_id=s["id"], user_id="u1")
|
|
|
|
refreshed = await svc.get_session(session_id=s["id"], user_id="u1")
|
|
assert refreshed["status"] == "draft"
|
|
assert refreshed["status"] != "starting"
|
|
assert await svc.list_runs(session_id=s["id"], user_id="u1") == []
|
|
|
|
|
|
# ── P1: concurrent approval decision ──────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_concurrent_approval_decision_only_one_wins(store):
|
|
svc = _svc(store)
|
|
s = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
appr = await store.approvals.create_approval(
|
|
{"session_id": s["id"], "run_id": run["id"], "summary": "高风险写入", "status": "pending"}
|
|
)
|
|
|
|
async def _decide(decision: str, who: str) -> tuple[str, object]:
|
|
try:
|
|
decided = await svc.decide_approval(approval_id=appr["id"], decision=decision, user_id=who)
|
|
return ("ok", decided)
|
|
except MafConflictError as exc:
|
|
return ("conflict", exc)
|
|
|
|
results = await asyncio.gather(
|
|
*[_decide("approved", "u1") for _ in range(4)],
|
|
*[_decide("rejected", "u1") for _ in range(4)],
|
|
)
|
|
oks = [r for r in results if r[0] == "ok"]
|
|
conflicts = [r for r in results if r[0] == "conflict"]
|
|
|
|
assert len(oks) == 1, f"expected exactly one winning decision, got {len(oks)}"
|
|
assert len(conflicts) == 7
|
|
|
|
winner = oks[0][1]
|
|
final = await store.approvals.get_approval(appr["id"])
|
|
assert final["status"] == winner["status"]
|
|
assert final["decided_by"] == winner["decided_by"] == "u1"
|
|
assert final["decided_at"] == winner["decided_at"]
|
|
assert final["decided_at"] is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approval_decision_cannot_be_overwritten(store):
|
|
svc = _svc(store)
|
|
s = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
appr = await store.approvals.create_approval(
|
|
{"session_id": s["id"], "run_id": run["id"], "summary": "x", "status": "pending"}
|
|
)
|
|
first = await svc.decide_approval(approval_id=appr["id"], decision="approved", user_id="u1")
|
|
with pytest.raises(MafConflictError):
|
|
await svc.decide_approval(approval_id=appr["id"], decision="rejected", user_id="u1")
|
|
final = await store.approvals.get_approval(appr["id"])
|
|
assert final["status"] == first["status"] == "approved"
|
|
assert final["decided_by"] == "u1"
|
|
|
|
|
|
# ── P1: cancel_run single-transaction atomicity ───────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_run_flush_failure_rolls_back_run_and_session(store):
|
|
"""If the cancel transaction's flush fails, run stays pending and session
|
|
stays running —nothing is half-committed."""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
assert run["status"] == "pending"
|
|
|
|
async def boom_flush(self, *args, **kwargs):
|
|
raise RuntimeError("simulated cancel flush failure")
|
|
|
|
with patch.object(AsyncSession, "flush", boom_flush):
|
|
with pytest.raises(RuntimeError, match="simulated cancel flush failure"):
|
|
await svc.cancel_run(run_id=run["id"], user_id="u1")
|
|
|
|
refreshed_run = await svc.get_run(run_id=run["id"], user_id="u1")
|
|
assert refreshed_run["status"] == "pending"
|
|
assert refreshed_run["finished_at"] is None
|
|
refreshed_session = await svc.get_session(session_id=s["id"], user_id="u1")
|
|
assert refreshed_session["status"] == "running"
|
|
|
|
# A healthy cancel afterwards succeeds.
|
|
cancelled = await svc.cancel_run(run_id=run["id"], user_id="u1")
|
|
assert cancelled["status"] == "cancelled"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_repairs_historical_inconsistent_session(store):
|
|
"""Historical half-completed data: run=cancelled but session still running.
|
|
Re-cancelling returns the cancelled run and repairs the session."""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
|
|
# Manufacture the inconsistency: run cancelled, session left running.
|
|
await store.runs.set_status(run["id"], "cancelled")
|
|
await store.sessions.set_status(s["id"], "running")
|
|
assert (await svc.get_session(session_id=s["id"], user_id="u1"))["status"] == "running"
|
|
|
|
result = await svc.cancel_run(run_id=run["id"], user_id="u1")
|
|
assert result["status"] == "cancelled"
|
|
assert (await svc.get_session(session_id=s["id"], user_id="u1"))["status"] == "cancelled"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_terminal_run_does_not_change_run(store):
|
|
"""A completed/failed run is not re-cancelled; its status is returned as-is."""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
await store.runs.set_status(run["id"], "completed")
|
|
await store.sessions.set_status(s["id"], "completed")
|
|
|
|
result = await svc.cancel_run(run_id=run["id"], user_id="u1")
|
|
assert result["status"] == "completed"
|
|
assert (await svc.get_run(run_id=run["id"], user_id="u1"))["status"] == "completed"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_concurrent_cancel_same_run_consistent(store):
|
|
"""Two concurrent cancels of the same run →run and session both cancelled,
|
|
no exceptions, no inconsistent state."""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
|
|
results = await asyncio.gather(
|
|
*[svc.cancel_run(run_id=run["id"], user_id="u1") for _ in range(2)],
|
|
return_exceptions=True,
|
|
)
|
|
assert not any(isinstance(r, Exception) for r in results), results
|
|
assert all(r["status"] == "cancelled" for r in results)
|
|
|
|
assert (await svc.get_run(run_id=run["id"], user_id="u1"))["status"] == "cancelled"
|
|
assert (await svc.get_session(session_id=s["id"], user_id="u1"))["status"] == "cancelled"
|
|
|
|
|
|
# ── P2: pending run time semantics ────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pending_run_has_no_started_at(store):
|
|
"""A freshly-created pending run is queued, not started: started_at is None."""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
assert run["status"] == "pending"
|
|
assert run["started_at"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_pending_run_sets_finished_at(store):
|
|
"""Cancelling a pending run sets finished_at (terminal) while started_at
|
|
stays None (it never actually ran)."""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
cancelled = await svc.cancel_run(run_id=run["id"], user_id="u1")
|
|
assert cancelled["status"] == "cancelled"
|
|
assert cancelled["finished_at"] is not None
|
|
assert cancelled["started_at"] is None
|
|
|