"""Phase 5 task/quota exception matrix (isolated, fault-injection). Boots an isolated business API (uvicorn) against a dedicated Postgres database and a local fake TTS upstream, then exercises the Phase 3 leftover exception matrix. It never touches the real TTS provider and uses throwaway accounts in the isolated database only. Required environment: P5_DATABASE_URL isolated Postgres DSN (role must own the DB) P5_API_DIR path to the services/api package to boot (default: repo services/api) P5_AUDIO_DIR isolated audio dir (default: /tmp/kaotings-p5-audio) P5_REPORT_PATH optional; if set, a markdown summary is written here Each scenario records: task status, upstream call-count delta, used, reserved, available (current-period quota account) and the audio-file result. Exit code 0 => every recorded invariant held; non-zero => at least one failure. """ from __future__ import annotations import json import os import signal import subprocess import sys import threading import time from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from uuid import uuid4 import httpx import psycopg API_PORT = int(os.environ.get("P5_API_PORT", "8200")) FAKE_PORT = int(os.environ.get("P5_FAKE_PORT", "9200")) API_ROOT = f"http://127.0.0.1:{API_PORT}/api/v1" FAKE_ROOT = f"http://127.0.0.1:{FAKE_PORT}" DB_URL = os.environ["P5_DATABASE_URL"] API_DIR = os.environ.get("P5_API_DIR") or str(Path(__file__).resolve().parents[1]) AUDIO_DIR = Path(os.environ.get("P5_AUDIO_DIR", "/tmp/kaotings-p5-audio")) REPORT_PATH = os.environ.get("P5_REPORT_PATH", "") DEFAULT_TIMEOUT = 30 # generous so the "slow"/"running" scenario stays in-flight SHORT_TIMEOUT = 1 # for the explicit upstream-timeout scenario class FakeTts(BaseHTTPRequestHandler): mode = "success" sleep_s = 0.0 calls = 0 _lock = threading.Lock() protocol_version = "HTTP/1.1" @classmethod def bump(cls): with cls._lock: cls.calls += 1 @classmethod def reset_calls(cls): with cls._lock: cls.calls = 0 @classmethod def call_count(cls): with cls._lock: return cls.calls def do_POST(self): # noqa: N802 self.bump() if self.sleep_s: time.sleep(self.sleep_s) if self.mode == "non_audio": payload, ctype = b"not audio", "text/plain" elif self.mode == "corrupt": payload, ctype = b"bad wav", "audio/wav" elif self.mode == "oversized": # the code checks size after httpx has read the body, so send a real # > 50 MiB payload (localhost, fast) to exercise UPSTREAM_RESPONSE_TOO_LARGE payload, ctype = b"x" * (50 * 1024 * 1024 + 1), "audio/wav" elif self.mode == "http_500": self.send_response(500) self.send_header("Content-Type", "text/plain") self.send_header("Content-Length", "4") self.end_headers() try: self.wfile.write(b"boom") self.wfile.flush() except (BrokenPipeError, ConnectionResetError): pass return else: # success payload = b"RIFF" + (36).to_bytes(4, "little") + b"WAVEfmt " + bytes(36) ctype = "audio/wav" try: self.send_response(200) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(payload))) self.end_headers() chunk = 1024 * 1024 for off in range(0, len(payload), chunk): self.wfile.write(payload[off:off + chunk]) self.wfile.flush() except (BrokenPipeError, ConnectionResetError): pass def log_message(self, *_args): return def db(): return psycopg.connect(DB_URL, row_factory=psycopg.rows.dict_row) def db_query(sql, params=()): with db() as c: return c.execute(sql, params).fetchall() def db_exec(sql, params=()): with db() as c: c.execute(sql, params) c.commit() def start_fake() -> ThreadingHTTPServer: server = ThreadingHTTPServer(("127.0.0.1", FAKE_PORT), FakeTts) threading.Thread(target=server.serve_forever, daemon=True).start() return server def boot_api(storage_dir: str, timeout: int = DEFAULT_TIMEOUT) -> subprocess.Popen: env = os.environ.copy() env.update({ "DATABASE_URL": DB_URL, "TTS_UPSTREAM_URL": FAKE_ROOT, "TTS_API_KEY": "test-only-fake-key", "TTS_TIMEOUT_SECONDS": str(timeout), "AUDIO_STORAGE_DIR": str(storage_dir), "SESSION_COOKIE_SECURE": "false", "APP_ENV": "test", "CSRF_SECRET": "p5-matrix-test-secret", }) proc = subprocess.Popen( [sys.executable, "-m", "uvicorn", "app.main:app", "--host", "127.0.0.1", "--port", str(API_PORT), "--workers", "1"], cwd=API_DIR, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) for _ in range(80): try: if httpx.get(f"http://127.0.0.1:{API_PORT}/healthz", timeout=1).status_code == 200: return proc except httpx.HTTPError: time.sleep(0.2) proc.kill() raise RuntimeError("isolated API did not start") def stop_api(proc: subprocess.Popen | None, force: bool = False) -> None: if proc is None or proc.poll() is not None: return proc.send_signal(signal.SIGKILL if force else signal.SIGTERM) try: proc.wait(timeout=10) except subprocess.TimeoutExpired: proc.kill() proc.wait(timeout=10) def make_client(email: str | None = None, password: str | None = None) -> httpx.Client: email = email or f"p5-{uuid4().hex[:12]}@example.com" password = password or f"Test-{uuid4().hex[:10]}-Aa1!" s = httpx.Client(base_url=API_ROOT, timeout=20, follow_redirects=True) csrf = s.get("/auth/csrf").json()["csrf_token"] assert s.post("/auth/register", json={"email": email, "password": password}, headers={"X-CSRF-Token": csrf}).status_code == 201 assert s.post("/auth/login", json={"email": email, "password": password}, headers={"X-CSRF-Token": csrf}).status_code == 200 s.headers.update({"X-CSRF-Token": csrf}) return s def user_id(session) -> str: return session.get("/auth/me").json()["id"] def make_admin() -> httpx.Client: session = make_client() uid = user_id(session) db_exec("UPDATE users SET role = 'admin', email_verified = true WHERE id = %s", (uid,)) return session def grant_vip(session) -> None: uid = user_id(session) db_exec("INSERT INTO membership_grants(user_id, plan, starts_at, reason) VALUES (%s, 'vip', now(), 'p5-matrix-test')", (uid,)) session.get("/account/usage") # materialize the current-period quota account def set_limit(session, limit: int) -> None: uid = user_id(session) with db() as c: c.execute( "UPDATE quota_accounts SET limit_snapshot = %s WHERE user_id = %s " "AND period_start = (SELECT max(period_start) FROM quota_accounts WHERE user_id = %s)", (limit, uid, uid), ) c.commit() def prepare_worker(session, limit: int) -> str: """Grant VIP (concurrency>1), materialize the current-period quota account, reset its counters and set the limit. Returns the user id.""" grant_vip(session) uid = user_id(session) with db() as c: c.execute( "UPDATE quota_accounts SET used=0, reserved=0, adjustment=0, limit_snapshot=%s " "WHERE user_id=%s AND period_start=(SELECT max(period_start) FROM quota_accounts WHERE user_id=%s)", (limit, uid, uid), ) c.commit() return uid def current_quota(uid: str) -> dict: row = db_query( "SELECT limit_snapshot, adjustment, used, reserved FROM quota_accounts WHERE user_id = %s " "AND period_start = (SELECT max(period_start) FROM quota_accounts WHERE user_id = %s)", (uid, uid), ) assert row, "no current quota account" q = dict(row[0]) q["available"] = max(0, q["limit_snapshot"] + q["adjustment"] - q["used"] - q["reserved"]) return q def invariant_ok(q: dict) -> bool: return (q["limit_snapshot"] + q["adjustment"]) - q["used"] - q["reserved"] >= 0 def create_task(session, text: str, idem: str | None = None): return session.post( "/tts/tasks", headers={"Idempotency-Key": idem or f"p5-{uuid4().hex}"}, json={"text": text, "voice_id": "default", "parameters": {"format": "wav", "speed": 1}}, ) def wait_task(session, task_id: str, timeout: float = 12) -> dict: end = time.time() + timeout while time.time() < end: t = session.get(f"/tts/tasks/{task_id}").json() if t["status"] in {"succeeded", "failed"}: return t time.sleep(0.15) return session.get(f"/tts/tasks/{task_id}").json() def audio_status(session, task_id: str) -> int: return session.get(f"/tts/tasks/{task_id}/audio").status_code def audio_file_exists(task_id: str) -> bool: return any(p.is_file() for p in AUDIO_DIR.rglob(f"{task_id}.*")) @dataclass class Scenario: name: str calls: int = 0 status: str = "" used: int | None = None reserved: int | None = None available: int | None = None audio: str = "n/a" ok: bool = True note: str = "" extra: dict = field(default_factory=dict) results: list[Scenario] = [] FAILURES: list[str] = [] API_HANDLE: subprocess.Popen | None = None POOL: dict[str, httpx.Client] = {} def record(name: str, **kw) -> Scenario: sc = Scenario(name=name, **kw) results.append(sc) if not sc.ok: FAILURES.append(name) return sc def check(cond: bool, sc: Scenario, msg: str) -> None: if not cond: sc.ok = False sc.note = (sc.note + " | " if sc.note else "") + msg def main() -> int: global API_HANDLE AUDIO_DIR.mkdir(parents=True, exist_ok=True) for p in AUDIO_DIR.rglob("*"): if p.is_file(): p.unlink() fake = start_fake() API_HANDLE = boot_api(str(AUDIO_DIR), DEFAULT_TIMEOUT) # Pre-create a small pool on the first API instance (register rate limit is # 5/hour per source IP, and S1-S6 share this single instance). S7-S10 each # restart the API, which resets the in-memory limiter, so they register fresh. POOL["admin"] = make_admin() POOL["w1"] = make_client() POOL["w2"] = make_client() POOL["w3"] = make_client() try: s1_success() s2_last_quota_contention() s3_admin_reduce_concurrent() s4_freeze_reduce_stress() s5_bad_upstream_responses() s6_cross_period_settlement() s7_storage_failure() s8_upstream_timeout() s9_queued_restart() s10_running_lease_recovery() finally: stop_api(API_HANDLE, force=True) fake.shutdown() emit_report() return 0 if not FAILURES else 1 def s1_success() -> None: session = POOL["w1"] uid = prepare_worker(session, 10000) FakeTts.mode, FakeTts.sleep_s = "success", 0.0 FakeTts.reset_calls() before = FakeTts.call_count() r = create_task(session, "测试") tid = r.json()["id"] task = wait_task(session, tid) q = current_quota(uid) audio = audio_status(session, tid) sc = record("S1 成功基线", calls=FakeTts.call_count() - before, status=task["status"], used=q["used"], reserved=q["reserved"], available=q["available"], audio=f"GET /audio={audio} file={audio_file_exists(tid)}") check(r.status_code == 202, sc, f"create={r.status_code}") check(task["status"] == "succeeded", sc, f"final={task['status']}") check(q["used"] == 2 and q["reserved"] == 0, sc, f"used={q['used']} reserved={q['reserved']} (expect used=2 reserved=0)") check(audio == 200 and audio_file_exists(tid), sc, f"audio={audio}") def s2_last_quota_contention() -> None: session = POOL["w2"] uid = prepare_worker(session, 5) FakeTts.mode, FakeTts.sleep_s = "success", 0.0 FakeTts.reset_calls() before = FakeTts.call_count() L = 3 with ThreadPoolExecutor(max_workers=2) as ex: resp = [f.result() for f in (ex.submit(create_task, session, "测" * L) for _ in range(2))] codes = sorted(x.status_code for x in resp) created = [x.json()["id"] for x in resp if x.status_code == 202] for tid in created: wait_task(session, tid) q = current_quota(uid) sc = record("S2 最后额度并发争用", calls=FakeTts.call_count() - before, status=f"codes={codes}", used=q["used"], reserved=q["reserved"], available=q["available"], audio="n/a") check(codes == [202, 409], sc, f"codes={codes} (expect one 202 + one 409 QUOTA_EXCEEDED)") check(len(created) == 1, sc, f"created={len(created)} (expect 1)") check(invariant_ok(q), sc, f"available negative: {q}") check(q["reserved"] == 0 and q["used"] == L, sc, f"used={q['used']} reserved={q['reserved']} (expect used={L} reserved=0)") def s3_admin_reduce_concurrent() -> None: admin = POOL["admin"] session = POOL["w3"] uid = prepare_worker(session, 100) iters, worst, violations = 8, None, 0 for _ in range(iters): with db() as c: c.execute("UPDATE quota_accounts SET used=0, reserved=0, adjustment=0, limit_snapshot=100 WHERE user_id=%s AND period_start=(SELECT max(period_start) FROM quota_accounts WHERE user_id=%s)", (uid, uid)) c.commit() idems = [f"p5r-{uuid4().hex}" for _ in range(4)] def reduce(i): return admin.post(f"/admin/users/{uid}/quota-adjustments", json={"amount": -30, "reason": "p5-concurrency-reduce", "idempotency_key": idems[i]}).status_code with ThreadPoolExecutor(max_workers=4) as ex: list(ex.map(reduce, range(4))) q = current_quota(uid) if not invariant_ok(q): violations += 1 if worst is None or q["available"] < worst["available"]: worst = q sc = record("S3 管理员并发调减额度", calls=0, status=f"iters={iters}", used=(worst or {}).get("used"), reserved=(worst or {}).get("reserved"), available=(worst or {}).get("available"), audio="n/a", extra={"invariant_violations": violations}) check(violations == 0, sc, f"额度不变式被破坏 {violations}/{iters} 次 (available 出现负值)") def s4_freeze_reduce_stress() -> None: admin = POOL["admin"] session = POOL["w1"] uid = prepare_worker(session, 100) FakeTts.mode, FakeTts.sleep_s = "success", 0.0 FakeTts.reset_calls() before = FakeTts.call_count() def mk(_): return create_task(session, "测" * 8).status_code def red(_): return admin.post(f"/admin/users/{uid}/quota-adjustments", json={"amount": -10, "reason": "p5-stress-reduce", "idempotency_key": f"p5s-{uuid4().hex}"}).status_code with ThreadPoolExecutor(max_workers=12) as ex: futs = [ex.submit(mk, i) for i in range(6)] + [ex.submit(red, i) for i in range(6)] codes = [f.result() for f in futs] time.sleep(3) # let all queued tasks settle q = current_quota(uid) sc = record("S4 冻结与调减并发压测", calls=FakeTts.call_count() - before, status=f"codes={sorted(codes)}", used=q["used"], reserved=q["reserved"], available=q["available"], audio="n/a") check(invariant_ok(q), sc, f"最终额度不变式被破坏: {q}") check(q["reserved"] == 0, sc, f"reserved 未清零: {q['reserved']}") def s5_bad_upstream_responses() -> None: session = POOL["w2"] uid = prepare_worker(session, 1000) FakeTts.sleep_s = 0.0 expect = {"non_audio": "UPSTREAM_NOT_AUDIO", "corrupt": "UPSTREAM_AUDIO_CORRUPT", "oversized": "UPSTREAM_RESPONSE_TOO_LARGE", "http_500": "UPSTREAM_HTTP_500"} for mode, code in expect.items(): FakeTts.mode = mode r = create_task(session, f"bad-{mode}") tid = r.json()["id"] task = wait_task(session, tid) q = current_quota(uid) sc = record(f"S5 上游{mode}", calls=1, status=task["status"], used=q["used"], reserved=q["reserved"], available=q["available"], audio=f"file={audio_file_exists(tid)}") check(task["status"] == "failed" and task["error_code"] == code, sc, f"status={task['status']} err={task['error_code']} (expect {code})") check(not audio_file_exists(tid), sc, "失败任务不应有音频文件") FakeTts.mode = "success" def s6_cross_period_settlement() -> None: session = POOL["w3"] uid = prepare_worker(session, 100) FakeTts.mode, FakeTts.sleep_s = "success", 0.0 with db() as c: qid = c.execute( "INSERT INTO quota_accounts(user_id, period_start, period_end, limit_snapshot, reserved) " "VALUES (%s, date_trunc('month', now() - interval '1 month'), date_trunc('month', now()), 100, 5) RETURNING id", (uid,), ).fetchone()["id"] c.execute( "INSERT INTO tts_tasks(user_id, text, text_length, voice_id, provider_voice_id, parameters, idempotency_key, request_hash, policy_version, quota_account_id, reserved_amount, status) " "VALUES (%s, 'cross', 5, (SELECT id FROM tts_voices WHERE provider_voice_id='default'), 'default', '{}'::jsonb, %s, %s, 1, %s, 5, 'queued')", (uid, f"p5xp-{uuid4().hex}", f"p5xp-{uuid4().hex}", qid), ) c.commit() time.sleep(3) # worker claims + settles prior = db_query("SELECT used, reserved FROM quota_accounts WHERE id = %s", (qid,)) cur = current_quota(uid) sc = record("S6 跨周期结算", calls=1, status="settled", used=prior[0]["used"], reserved=prior[0]["reserved"], available=cur["available"], audio="n/a", extra={"prior_used": prior[0]["used"], "current_used": cur["used"]}) check(prior[0]["used"] == 5 and prior[0]["reserved"] == 0, sc, f"上一周期账户 used={prior[0]['used']} reserved={prior[0]['reserved']} (expect used=5 reserved=0)") check(cur["used"] == 0, sc, f"当前周期账户不应被结算, used={cur['used']}") def s7_storage_failure() -> None: global API_HANDLE stop_api(API_HANDLE, force=True) blocker = Path("/tmp/kaotings-p5-storage-blocker") if blocker.exists() and blocker.is_dir(): blocker.rmdir() blocker.write_text("not a directory") API_HANDLE = boot_api(str(blocker), DEFAULT_TIMEOUT) session = make_client() uid = user_id(session) FakeTts.mode, FakeTts.sleep_s = "success", 0.0 r = create_task(session, "存储失败") tid = r.json()["id"] task = wait_task(session, tid) q = current_quota(uid) blocker.unlink(missing_ok=True) sc = record("S7 存储失败", calls=1, status=task["status"], used=q["used"], reserved=q["reserved"], available=q["available"], audio=f"file={audio_file_exists(tid)}") check(task["status"] == "failed", sc, f"status={task['status']}") check(q["reserved"] == 0, sc, f"reserved 未释放: {q['reserved']}") check(invariant_ok(q), sc, f"不变式: {q}") def s8_upstream_timeout() -> None: global API_HANDLE stop_api(API_HANDLE, force=True) API_HANDLE = boot_api(str(AUDIO_DIR), SHORT_TIMEOUT) session = make_client() uid = user_id(session) FakeTts.mode, FakeTts.sleep_s = "success", 5 # sleeps past the 1s client timeout r = create_task(session, "超时") tid = r.json()["id"] task = wait_task(session, tid, timeout=12) FakeTts.sleep_s = 0.0 q = current_quota(uid) sc = record("S8 上游超时", calls=1, status=task["status"], used=q["used"], reserved=q["reserved"], available=q["available"], audio=f"file={audio_file_exists(tid)}") check(task["status"] == "failed" and task["error_code"] == "UPSTREAM_TIMEOUT", sc, f"status={task['status']} err={task['error_code']} (expect UPSTREAM_TIMEOUT)") check(q["reserved"] == 0, sc, f"reserved 未释放: {q['reserved']}") stop_api(API_HANDLE, force=True) API_HANDLE = boot_api(str(AUDIO_DIR), DEFAULT_TIMEOUT) def s9_queued_restart() -> None: global API_HANDLE session = make_client() uid = user_id(session) grant_vip(session) FakeTts.mode, FakeTts.sleep_s = "success", 4 # first task stays running r_a = create_task(session, "运行中") r_b = create_task(session, "排队中") tid_a, tid_b = r_a.json()["id"], r_b.json()["id"] time.sleep(1.5) # A running (blocked in fake), B queued stop_api(API_HANDLE, force=True) # unclean shutdown FakeTts.sleep_s = 0.0 API_HANDLE = boot_api(str(AUDIO_DIR), DEFAULT_TIMEOUT) task_b = wait_task(session, tid_b, timeout=15) task_a = session.get(f"/tts/tasks/{tid_a}").json() q = current_quota(uid) sc = record("S9 重启恢复-queued", calls=2, status=f"queued->{task_b['status']} running->{task_a['status']}", used=q["used"], reserved=q["reserved"], available=q["available"], audio=f"b_file={audio_file_exists(tid_b)}") check(task_b["status"] == "succeeded", sc, f"queued 任务重启后未恢复: {task_b['status']}") check(audio_file_exists(tid_b), sc, "queued 任务重启后音频缺失") def s10_running_lease_recovery() -> None: global API_HANDLE session = make_client() uid = user_id(session) grant_vip(session) FakeTts.mode, FakeTts.sleep_s = "success", 6 # keep it running r = create_task(session, "租约过期") tid = r.json()["id"] time.sleep(1.5) stop_api(API_HANDLE, force=True) # worker dies while task running db_exec("UPDATE tts_tasks SET lease_expires_at = now() - interval '1 second' WHERE id = %s", (tid,)) q_before = current_quota(uid) API_HANDLE = boot_api(str(AUDIO_DIR), DEFAULT_TIMEOUT) task = wait_task(session, tid, timeout=20) q = current_quota(uid) sc = record("S10 重启恢复-running租约过期", calls=0, status=task["status"], used=q["used"], reserved=q["reserved"], available=q["available"], audio=f"file={audio_file_exists(tid)}") check(task["status"] == "failed" and task["error_code"] == "WORKER_LEASE_EXPIRED", sc, f"status={task['status']} err={task['error_code']} (expect WORKER_LEASE_EXPIRED)") check(q["reserved"] < q_before["reserved"], sc, f"reserved 未释放 ({q_before['reserved']}->{q['reserved']})") check(invariant_ok(q), sc, f"不变式: {q}") def emit_report() -> None: lines = ["| 场景 | 上游调用 | 任务状态 | 已用 | 冻结 | 可用 | 音频结果 | 通过 | 备注 |", "| --- | --- | --- | --- | --- | --- | --- | --- | --- |"] for sc in results: lines.append(f"| {sc.name} | {sc.calls} | {sc.status} | {sc.used} | {sc.reserved} | {sc.available} | {sc.audio} | {'通过' if sc.ok else '失败'} | {sc.note} |") blob = "\n".join(lines) + f"\n\n失败项: {FAILURES if FAILURES else '无'}\n" print(blob) if REPORT_PATH: Path(REPORT_PATH).write_text(blob, encoding="utf-8") print(json.dumps({"failures": FAILURES, "scenarios": len(results)}, ensure_ascii=False)) if __name__ == "__main__": raise SystemExit(main())