From 1139aafe6da5cad55e1050a5684bad8bd8a18dd5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Thu, 15 Oct 2026 16:38:26 +0800 Subject: [PATCH 1/7] Stop false storage-degraded banners under read load Audit reads (ticket/dashboard/logs/storage) now run on fresh read-only WAL connections instead of queueing on the 250ms write lock, so parallel dashboard views no longer starve live record_request commits or each other. Lock-contention timeouts are reported as transient_contention and never latch the 24h degraded banner; only real storage errors do. The dashboard P95 scan moves to SQL (COUNT + ORDER BY OFFSET) over a new covered (started_at, duration_ms) index (schema v4, backfilled once from payloads), replacing the per-row json_extract fetchall that serialized 300k-row scans through Python. /admin/storage keeps reporting the full health payload with degraded flags instead of 503, so the storage page diagnostics stay visible during degradation. --- app/admin_api.py | 17 +-- app/audit_store.py | 210 +++++++++++++++++++++++++------------- tests/test_audit_store.py | 94 ++++++++++++++++- web/src/values.tsx | 2 + 4 files changed, 241 insertions(+), 82 deletions(-) diff --git a/app/admin_api.py b/app/admin_api.py index ca89845..7cc6a2e 100644 --- a/app/admin_api.py +++ b/app/admin_api.py @@ -536,16 +536,17 @@ async def logs_clear(request): @route("GET", "/admin/storage") async def storage_get(request): + # A degraded store still reports its own health; a 503 here would hide + # the diagnostics this endpoint exists to surface. result = await run_in_threadpool(audit.storage) - if result.get("degraded"): - return error_response(503, "存储状态暂时无法读取，请检查审计存储") archives = [] - try: - directory = Path(str(audit.path)).parent / "archives" - archives = sorted((entry.name for entry in directory.glob("audit-*.jsonl")), reverse=True)[:20] - except OSError as exc: - audit._fault(exc) - return error_response(503, "存储状态暂时无法读取，请检查审计存储") + path = getattr(audit, "path", None) + if path: + try: + directory = Path(str(path)).parent / "archives" + archives = sorted((entry.name for entry in directory.glob("audit-*.jsonl")), reverse=True)[:20] + except OSError as exc: + audit._fault(exc) return JSONResponse({**result, "archives": archives}) @route("POST", "/admin/storage/archive") diff --git a/app/audit_store.py b/app/audit_store.py index 770bff8..58ed829 100644 --- a/app/audit_store.py +++ b/app/audit_store.py @@ -14,8 +14,9 @@ import time import uuid from typing import Any +from urllib.parse import quote -SCHEMA_VERSION = 3 +SCHEMA_VERSION = 4 _CLEANUP_BATCH = 128 _CLEANUP_SECONDS = 0.05 METRICS = ("input_tokens", "output_tokens", "cache_read_tokens", "cache_creation_tokens", @@ -110,6 +111,7 @@ def _secure_path(path: str | os.PathLike) -> Path: class AuditStore: # 一次瞬时写失败只让降级标记保持一天；终身计数仍如实上报，不随时间消失。 DEGRADED_WINDOW_SECONDS = 24 * 3600 + TRANSIENT_WINDOW_SECONDS = 3600 def __init__(self, path, max_bytes=256 * 1024 * 1024, retention_days=30, preview_limit=8192): @@ -118,8 +120,9 @@ def __init__(self, path, max_bytes=256 * 1024 * 1024, retention_days=30, self._lock = threading.Lock() self._archive_lock = threading.Lock() self._health_lock = threading.Lock() - self.failure_count = self.dropped_records = 0 + self.failure_count = self.dropped_records = self.transient_count = 0 self.failure_events: list[float] = [] + self.transient_events: list[float] = [] self.last_failure_at: float | None = None self.last_error = None self._closed = False @@ -135,7 +138,7 @@ def __init__(self, path, max_bytes=256 * 1024 * 1024, retention_days=30, self._db.execute("BEGIN IMMEDIATE") version = self._db.execute("PRAGMA user_version").fetchone()[0] tables = self._db.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall() - if version not in (0, 1, 2, SCHEMA_VERSION) or (version == 0 and tables): + if version not in (0, 1, 2, 3, SCHEMA_VERSION) or (version == 0 and tables): raise ValueError("unsupported audit schema") self._db.execute("CREATE TABLE IF NOT EXISTS state (id INTEGER PRIMARY KEY CHECK(id=1), epoch INTEGER NOT NULL, detail_generation INTEGER NOT NULL, cleared_at REAL NOT NULL)") self._db.execute("INSERT OR IGNORE INTO state VALUES(1,0,0,0)") @@ -150,6 +153,7 @@ def __init__(self, path, max_bytes=256 * 1024 * 1024, retention_days=30, self._db.execute(f"CREATE TABLE IF NOT EXISTS {table} (bucket INTEGER NOT NULL, dimension TEXT NOT NULL, dimension_key TEXT NOT NULL, payload TEXT NOT NULL, PRIMARY KEY(bucket,dimension,dimension_key))") self._migrate_ingest_time() self._migrate_client_ip() + self._migrate_duration() self._db.execute(f"PRAGMA user_version={SCHEMA_VERSION}") self._epoch = self._db.execute("SELECT epoch FROM state").fetchone()[0] self._db.execute("COMMIT") @@ -176,6 +180,26 @@ def _migrate_client_ip(self): self._db.execute( "CREATE INDEX IF NOT EXISTS requests_client ON requests(client_ip, started_at)") + def _migrate_duration(self): + # Additive v4 migration: a covered (started_at, duration_ms) index keeps + # the dashboard P95 scan off per-row json_extract payload parsing. + columns = [row[1] for row in self._db.execute("PRAGMA table_info(requests)").fetchall()] + if "duration_ms" not in columns: + self._db.execute("ALTER TABLE requests ADD COLUMN duration_ms REAL") + self._db.execute("UPDATE requests SET duration_ms=json_extract(payload,'$.duration_ms')") + self._db.execute("CREATE INDEX IF NOT EXISTS requests_duration ON requests(started_at,duration_ms)") + + def _open_read(self): + """Fresh read-only WAL snapshot per call; None falls back to the write path.""" + try: + conn = sqlite3.connect(f"file:{quote(str(self.path))}?mode=ro", uri=True, timeout=5, + check_same_thread=False, isolation_level=None) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA busy_timeout=5000") + return conn + except sqlite3.Error: + return None + def _init_accounting(self): # Additive v1 migration: scan legacy details only once, under the same # write transaction that installs triggers. Reopening never recounts. @@ -200,12 +224,18 @@ def _validate(max_bytes, retention_days, preview_limit): raise ValueError("invalid preview_limit") def _fault(self, exc, dropped=False): + # Lock contention is expected under load and never latches the degraded + # banner; only real storage errors do. Lifetime counters stay honest. with self._health_lock: self.failure_count += 1 self.dropped_records += int(dropped) - self.failure_events.append(time.monotonic()) self.last_failure_at = time.time() + if isinstance(exc, TimeoutError): + self.transient_count += 1 + self.transient_events.append(time.monotonic()) + return # Exception messages may include SQL or user data; retain only type. + self.failure_events.append(time.monotonic()) self.last_error = type(exc).__name__ def _recent_failures(self): @@ -216,6 +246,14 @@ def _recent_failures(self): self.failure_events.pop(0) return len(self.failure_events) + def _recent_transients(self): + # Contention is reported for a shorter window; it is visibility, not alarm. + with self._health_lock: + deadline = time.monotonic() - self.TRANSIENT_WINDOW_SECONDS + while self.transient_events and self.transient_events[0] <= deadline: + self.transient_events.pop(0) + return len(self.transient_events) + def note_failure(self, code="ObservationError"): self._fault(RuntimeError(), dropped=True) @@ -231,7 +269,7 @@ def _run(self, callback, fallback=None, write=False, dropped=False, on_commit=No self._db.set_progress_handler(lambda: int(time.monotonic() > deadline), 1000) if write: self._db.execute("BEGIN IMMEDIATE") - result = callback() + result = callback(self._db) if write: self._db.execute("COMMIT") if on_commit is not None: @@ -252,16 +290,37 @@ def _run(self, callback, fallback=None, write=False, dropped=False, on_commit=No self._db.set_progress_handler(None, 0) self._lock.release() + def _read_run(self, callback, fallback=None): + """Each read takes a fresh read-only WAL snapshot: dashboards and log + listings never queue behind write transactions or each other.""" + conn = self._open_read() + if conn is None: + return self._run(callback, fallback) + try: + if self._closed: + raise RuntimeError("closed") + deadline = time.monotonic() + 5 + conn.set_progress_handler(lambda: int(time.monotonic() > deadline), 1000) + return callback(conn) + except Exception as exc: + self._fault(exc) + return fallback + finally: + try: + conn.close() + except sqlite3.Error: + pass + @property def epoch(self): return self.ticket()["epoch"] def ticket(self): - def fetch(): - row = self._db.execute("SELECT epoch,detail_generation FROM state").fetchone() + def fetch(db): + row = db.execute("SELECT epoch,detail_generation FROM state").fetchone() self._epoch = row["epoch"] return dict(row) - return self._run(fetch, {"epoch": self._epoch, "detail_generation": -1}) + return self._read_run(fetch, {"epoch": self._epoch, "detail_generation": -1}) @staticmethod def _empty_stats(): @@ -278,7 +337,7 @@ def _merge(target, source): target[key] = (target.get(key) or 0) + value return target - def _aggregate(self, record): + def _aggregate(self, db, record): increment = self._empty_stats() increment["requests"] = 1 increment[record["outcome"]] = 1 @@ -297,9 +356,9 @@ def _aggregate(self, record): for table, seconds in (("stats_hourly", 3600), ("stats_daily", 86400), ("stats_totals", 0)): bucket = int(record["started_at"] // seconds) * seconds if seconds else 0 for dimension, key in dimensions: - old = self._db.execute(f"SELECT payload FROM {table} WHERE bucket=? AND dimension=? AND dimension_key=?", (bucket, dimension, key)).fetchone() + old = db.execute(f"SELECT payload FROM {table} WHERE bucket=? AND dimension=? AND dimension_key=?", (bucket, dimension, key)).fetchone() stats = self._merge(json.loads(old[0]) if old else self._empty_stats(), increment) - self._db.execute(f"INSERT OR REPLACE INTO {table} VALUES(?,?,?,?)", (bucket, dimension, key, json.dumps(stats))) + db.execute(f"INSERT OR REPLACE INTO {table} VALUES(?,?,?,?)", (bucket, dimension, key, json.dumps(stats))) def _sanitize_record(self, source): result = {key: safe_label(source.get(key)) for key in @@ -346,29 +405,29 @@ def _sanitize_record(self, source): return result def record_request(self, record): - def commit(): + def commit(db): data = self._sanitize_record(record) - state = self._db.execute("SELECT * FROM state").fetchone() + state = db.execute("SELECT * FROM state").fetchone() if data["epoch"] != state["epoch"]: return {"ok": True, "recorded": False, "reason": "stale_epoch"} - if not self._db.execute("INSERT OR IGNORE INTO ingest VALUES(?, 'request', ?)", (data["id"], data["started_at"])).rowcount: + if not db.execute("INSERT OR IGNORE INTO ingest VALUES(?, 'request', ?)", (data["id"], data["started_at"])).rowcount: return {"ok": True, "recorded": False, "reason": "duplicate"} - self._aggregate(data) + self._aggregate(db, data) generation = record.get("detail_generation") keep = (data["started_at"] > state["cleared_at"] and (generation is None or generation == state["detail_generation"])) if keep: payload = json.dumps(data, separators=(",", ":")) - self._db.execute( - "INSERT INTO requests(id,started_at,model,profile,credential,outcome,status_code,payload,logical_bytes,client_ip) VALUES(?,?,?,?,?,?,?,?,?,?)", + db.execute( + "INSERT INTO requests(id,started_at,model,profile,credential,outcome,status_code,payload,logical_bytes,client_ip,duration_ms) VALUES(?,?,?,?,?,?,?,?,?,?,?)", (data["id"], data["started_at"], data["model"], data["profile"], data["credential"], data["outcome"], data["status_code"], payload, len(payload.encode()) + sum(len(json.dumps(a).encode()) + 32 for a in data["attempts"]) + 256, - data["client_ip"])) + data["client_ip"], data["duration_ms"])) for i, attempt in enumerate(data["attempts"]): - self._db.execute("INSERT INTO attempts VALUES(?,?,?)", (data["id"], i, json.dumps(attempt))) + db.execute("INSERT INTO attempts VALUES(?,?,?)", (data["id"], i, json.dumps(attempt))) self._prune() - return {"ok": True, "recorded": True, "details": bool(keep and self._db.execute("SELECT 1 FROM requests WHERE id=?", (data["id"],)).fetchone())} + return {"ok": True, "recorded": True, "details": bool(keep and db.execute("SELECT 1 FROM requests WHERE id=?", (data["id"],)).fetchone())} return self._run(commit, {"ok": False, "recorded": False, "reason": "storage_failure"}, write=True, dropped=True) def _oldest(self, cutoff=None): @@ -400,14 +459,15 @@ def _expire(self, retention_days=None, deadline=None): break self._db.execute("DELETE FROM ingest WHERE id=?", (row[0],)) - def _cleanup_pending(self, budget=None, retention_days=None): + def _cleanup_pending(self, budget=None, retention_days=None, db=None): budget = self.max_bytes if budget is None else budget cutoff = time.time() - (self.retention_days if retention_days is None else retention_days) * 86400 - accounting = self._db.execute("SELECT logical_bytes,cleanup_target FROM detail_accounting WHERE id=1").fetchone() - ingest_expired = self._db.execute( + db = self._db if db is None else db + accounting = db.execute("SELECT logical_bytes,cleanup_target FROM detail_accounting WHERE id=1").fetchone() + ingest_expired = db.execute( "SELECT 1 FROM ingest WHERE created_at budget or accounting[1] is not None or ingest_expired or any( - self._db.execute(f"SELECT 1 FROM {table} WHERE started_at limit rows = rows[:limit] return {"items": [json.loads(row["payload"]) for row in rows], "next_cursor": json.dumps([rows[-1]["started_at"], rows[-1]["id"]]) if more else None, "has_more": more} - return self._run(fetch, {"items": [], "next_cursor": None, "has_more": False, "degraded": True}, write=True) + return self._read_run(fetch, {"items": [], "next_cursor": None, "has_more": False, "degraded": True}) def get_request(self, id): - def fetch(): - self._prune() - row = self._db.execute("SELECT payload FROM requests WHERE id=? AND started_at>=?", (id, time.time() - self.retention_days * 86400)).fetchone() + def fetch(db): + row = db.execute("SELECT payload FROM requests WHERE id=? AND started_at>=?", (id, time.time() - self.retention_days * 86400)).fetchone() return json.loads(row[0]) if row else None - return self._run(fetch, write=True) + return self._read_run(fetch) def dashboard(self, days=30, granularity="auto"): days = int(days) @@ -527,10 +585,10 @@ def dashboard(self, days=30, granularity="auto"): now = time.time() start = int(now // 86400) * 86400 - (days - 1) * 86400 period = {"days": days, "start": start, "end": now, "timezone": "UTC", "granularity": grain} - def fetch(): + def fetch(db): summary = self._empty_stats() series, models, profiles = [], {}, {} - rows = self._db.execute("SELECT * FROM stats_daily WHERE bucket>=? AND bucket<=? ORDER BY bucket", (start, now)).fetchall() + rows = db.execute("SELECT * FROM stats_daily WHERE bucket>=? AND bucket<=? ORDER BY bucket", (start, now)).fetchall() for row in rows: stats = json.loads(row["payload"]) if row["dimension"] == "global": @@ -545,7 +603,7 @@ def fetch(): target = models if row["dimension"] == "model" else profiles self._merge(target.setdefault(row["dimension_key"], self._empty_stats()), stats) if grain == "hour": - rows = self._db.execute("SELECT bucket,payload FROM stats_hourly WHERE dimension='global' AND bucket>=? AND bucket<=? ORDER BY bucket", (start, now)).fetchall() + rows = db.execute("SELECT bucket,payload FROM stats_hourly WHERE dimension='global' AND bucket>=? AND bucket<=? ORDER BY bucket", (start, now)).fetchall() series = [{"bucket": row["bucket"], **json.loads(row["payload"])} for row in rows] partial = grain == "hour" and sum(row["requests"] for row in series) != summary["requests"] step = 3600 if grain == "hour" else 86400 @@ -557,32 +615,35 @@ def fetch(): row["date"] = time.strftime("%Y-%m-%d %H:00" if grain == "hour" else "%Y-%m-%d", time.gmtime(row["bucket"])) summary["success_rate"] = summary["success"] / summary["requests"] if summary["requests"] else None previous = self._empty_stats() - for row in self._db.execute( + for row in db.execute( "SELECT payload FROM stats_daily WHERE dimension='global' AND bucket>=? AND bucket=? AND started_at<=?", - (start, now)).fetchall() if r[0] is not None) + known = db.execute( + "SELECT COUNT(*) FROM requests WHERE started_at>=? AND started_at<=? AND duration_ms IS NOT NULL", + (start, now)).fetchone()[0] # Detail rows can be pruned, archived, or cleared independently of daily # aggregates; only report P95 when the detail sample covers the period. summary["p95_duration_ms"] = ( - durations[max(0, math.ceil(len(durations) * 0.95) - 1)] - if durations and len(durations) >= summary["duration_ms_known"] + db.execute( + "SELECT duration_ms FROM requests WHERE started_at>=? AND started_at<=? AND duration_ms IS NOT NULL" + " ORDER BY duration_ms LIMIT 1 OFFSET ?", + (start, now, max(0, math.ceil(known * 0.95) - 1))).fetchone()[0] + if known and known >= summary["duration_ms_known"] else None) return {"summary": summary, "previous": previous, "series": series, "models": [{"model": key, **value} for key, value in models.items()], "profiles": [{"profile": key, **value} for key, value in profiles.items()], "generated_at": now, "range": {**period, "partial": partial}} - return self._run(fetch, {"summary": {**self._empty_stats(), "p95_duration_ms": None}, + return self._read_run(fetch, {"summary": {**self._empty_stats(), "p95_duration_ms": None}, "previous": None, "series": [], "models": [], "profiles": [], "generated_at": now, "range": period, "degraded": True}) def credential_stats(self): - def fetch(): + def fetch(db): out = {} - for row in self._db.execute( + for row in db.execute( "SELECT dimension_key,payload FROM stats_totals WHERE dimension='credential' AND dimension_key!=''").fetchall(): stats = json.loads(row["payload"]) reqs = stats.get("requests") or 0 @@ -599,7 +660,7 @@ def fetch(): "credit": stats.get("credit"), "avg_duration_ms": round((stats.get("duration_ms_sum") or 0) / dur_known, 1) if dur_known else None, } - for row in self._db.execute( + for row in db.execute( "SELECT credential,MAX(started_at) AS last_at,model,outcome,status_code " "FROM requests WHERE credential IS NOT NULL AND credential!='' GROUP BY credential").fetchall(): entry = out.setdefault(row["credential"], { @@ -612,19 +673,23 @@ def fetch(): entry["last_outcome"] = row["outcome"] entry["last_status_code"] = row["status_code"] return out - return self._run(fetch, {}) + return self._read_run(fetch, {}) + def storage(self): - def fetch(): - self._prune() - self._epoch = self._db.execute("SELECT epoch FROM state").fetchone()[0] - page_size = self._db.execute("PRAGMA page_size").fetchone()[0] - freelist = self._db.execute("PRAGMA freelist_count").fetchone()[0] - return {**dict(self._db.execute("SELECT logical_bytes,request_count,event_count,ingest_count FROM detail_accounting WHERE id=1").fetchone()), - "pending_cleanup": self._cleanup_pending(), - "page_count": self._db.execute("PRAGMA page_count").fetchone()[0], + # Opening the storage/dashboard page keeps incremental cleanup moving + # with the same bounded write slice; heavy scans stay read-only. + self._run(lambda db: self._prune(), write=True) + + def fetch(db): + self._epoch = db.execute("SELECT epoch FROM state").fetchone()[0] + page_size = db.execute("PRAGMA page_size").fetchone()[0] + freelist = db.execute("PRAGMA freelist_count").fetchone()[0] + return {**dict(db.execute("SELECT logical_bytes,request_count,event_count,ingest_count FROM detail_accounting WHERE id=1").fetchone()), + "pending_cleanup": self._cleanup_pending(db=db), + "page_count": db.execute("PRAGMA page_count").fetchone()[0], "freelist_pages": freelist, "reclaimable_bytes": freelist * page_size} - result = self._run(fetch, {"logical_bytes": None, "pending_cleanup": None, - "reclaimable_bytes": None}, write=True) + result = self._read_run(fetch, {"logical_bytes": None, "pending_cleanup": None, + "reclaimable_bytes": None}) sizes = {} for key, suffix in (("db_bytes", ""), ("wal_bytes", "-wal"), ("shm_bytes", "-shm")): try: @@ -638,6 +703,7 @@ def fetch(): return {**result, **sizes, "max_bytes": self.max_bytes, "retention_days": self.retention_days, "preview_limit": self.preview_limit, "schema_version": SCHEMA_VERSION, "epoch": self._epoch, "degraded": bool(recent), "recent_failures": recent, "failure_count": self.failure_count, + "transient_contention": self._recent_transients(), "transient_count": self.transient_count, "dropped_records": self.dropped_records, "last_error": self.last_error, "last_failure_at": self.last_failure_at, "closed": self._closed, "budget_scope": "details_logical_bytes", @@ -647,16 +713,16 @@ def fetch(): def clear(self, scope="details"): if scope not in ("details", "all"): raise ValueError("invalid clear scope") - def commit(): + def commit(db): for table in ("requests", "attempts", "events"): - self._db.execute(f"DELETE FROM {table}") - self._db.execute("UPDATE detail_accounting SET cleanup_target=NULL WHERE id=1") - self._db.execute("UPDATE state SET detail_generation=detail_generation+1,cleared_at=? WHERE id=1", (time.time(),)) + db.execute(f"DELETE FROM {table}") + db.execute("UPDATE detail_accounting SET cleanup_target=NULL WHERE id=1") + db.execute("UPDATE state SET detail_generation=detail_generation+1,cleared_at=? WHERE id=1", (time.time(),)) if scope == "all": for table in ("stats_hourly", "stats_daily", "stats_totals", "ingest"): - self._db.execute(f"DELETE FROM {table}") - self._db.execute("UPDATE state SET epoch=epoch+1 WHERE id=1") - epoch = self._db.execute("SELECT epoch FROM state").fetchone()[0] + db.execute(f"DELETE FROM {table}") + db.execute("UPDATE state SET epoch=epoch+1 WHERE id=1") + epoch = db.execute("SELECT epoch FROM state").fetchone()[0] return {"ok": True, "scope": scope, "epoch": epoch, "aggregates_preserved": scope == "details"} return self._run(commit, {"ok": False, "scope": scope}, write=True, on_commit=lambda result: setattr(self, "_epoch", result["epoch"])) @@ -785,7 +851,7 @@ def configure(self, max_bytes=None, retention_days=None, preview_limit=None): self.retention_days if retention_days is None else retention_days, self.preview_limit if preview_limit is None else preview_limit) self._validate(*values) - def commit(): + def commit(db): # Resolve unspecified settings under the lock, not from a stale # snapshot taken before another configure() completed. current = (self.max_bytes if max_bytes is None else max_bytes, @@ -793,7 +859,7 @@ def commit(): self.preview_limit if preview_limit is None else preview_limit) self._prune(max_bytes=current[0], retention_days=current[1]) return {"ok": True, "max_bytes": current[0], "retention_days": current[1], "preview_limit": current[2], - "pending_cleanup": self._cleanup_pending(current[0], current[1])} + "pending_cleanup": self._cleanup_pending(current[0], current[1], db=db)} def publish(result): self.max_bytes, self.retention_days, self.preview_limit = ( result["max_bytes"], result["retention_days"], result["preview_limit"]) @@ -802,8 +868,8 @@ def publish(result): def close(self): if self._closed: return - def finish(): - self._db.execute("PRAGMA wal_checkpoint(PASSIVE)") - self._db.close() + def finish(db): + db.execute("PRAGMA wal_checkpoint(PASSIVE)") + db.close() self._closed = True self._run(finish) diff --git a/tests/test_audit_store.py b/tests/test_audit_store.py index 7980165..0494584 100644 --- a/tests/test_audit_store.py +++ b/tests/test_audit_store.py @@ -192,14 +192,19 @@ def test_sqlite_busy_has_bounded_wait_and_no_partial_ingest(self): self.assertEqual(self.store.storage()["ingest_count"], 0) self.assertTrue(self.store.storage()["degraded"]) - def test_thread_lock_timeout_is_visible(self): + def test_thread_lock_timeout_is_transient_not_degraded(self): + # Contention is expected under load; it must never latch the 24h banner. record = self.record() with self.store._lock: start = time.monotonic() result = self.store.record_request(record) self.assertLess(time.monotonic() - start, 1) self.assertFalse(result["ok"]) - self.assertEqual(self.store.storage()["last_error"], "TimeoutError") + health = self.store.storage() + self.assertFalse(health["degraded"]) + self.assertEqual(health["transient_contention"], 1) + self.assertEqual(health["dropped_records"], 1) + self.assertIsNone(health["last_error"]) def test_transient_fault_degrades_only_within_window(self): with patch.object(self.store, "_aggregate", side_effect=sqlite3.OperationalError("boom")): @@ -218,6 +223,91 @@ def test_transient_fault_degrades_only_within_window(self): self.assertEqual(health["dropped_records"], 1) self.assertEqual(health["last_error"], "OperationalError") + def test_reads_do_not_queue_behind_write_transactions(self): + # The WebUI fires parallel dashboard/log/storage reads while inference + # writes run; reads take a WAL snapshot, never the write lock. + for i in range(3): + self.assertTrue(self.store.record_request( + self.record(f"r{i}", started_at=time.time() - i, duration_ms=100 + i))["recorded"]) + with self.store._lock: # a write transaction in flight + self.assertEqual(self.store.ticket()["epoch"], 0) + self.assertEqual(self.store.dashboard(30)["summary"]["requests"], 3) + self.assertEqual(self.store.get_request("r0")["id"], "r0") + self.assertEqual(len(self.store.list_records("request", 10)["items"]), 3) + self.assertFalse(self.store.storage()["degraded"]) + # Writers and readers share no lock; a fresh record commits immediately. + self.assertTrue(self.store.record_request( + self.record("w0", started_at=time.time()))["recorded"]) + self.assertFalse(self.store.storage()["degraded"]) + + def test_dashboard_view_storm_does_not_drop_traffic_records(self): + # Regression: parallel dashboard queries used to steal the 250ms write + # lock budget of live traffic and latch a day of false storage alarms. + now = time.time() + payload = json.dumps({"id": "seed", "duration_ms": 5}) + with self.store._lock: + self.store._db.execute("PRAGMA synchronous=OFF") # bulk seed only + self.store._db.execute("BEGIN IMMEDIATE") + self.store._db.executemany( + "INSERT INTO requests VALUES(?,?,?,?,?,?,?,?,?,?,?)", + [(f"s{i}", now - i, "m", "p", "c", "success", 200, payload, 64, None, 5.0) + for i in range(20000)]) + self.store._db.execute("COMMIT") + self.store._db.execute("PRAGMA synchronous=FULL") + + def traffic(tag): + dropped = 0 + for i in range(20): + if not self.store.record_request( + self.record(f"{tag}{i}", started_at=time.time()))["ok"]: + dropped += 1 + return dropped + + with concurrent.futures.ThreadPoolExecutor(5) as pool: + futures = [pool.submit(self.store.dashboard, 30), + pool.submit(self.store.dashboard, 1, "hour"), + pool.submit(self.store.storage), pool.submit(traffic, "a"), + pool.submit(traffic, "b")] + dashboards = [futures[0].result(), futures[1].result()] + dropped = futures[3].result() + futures[4].result() + self.assertIsNone(dashboards[0].get("degraded")) + self.assertIsNone(dashboards[1].get("degraded")) + self.assertEqual(dropped, 0) + health = self.store.storage() + self.assertFalse(health["degraded"]) + self.assertEqual(health["dropped_records"], 0) + self.assertEqual(health["failure_count"], 0) + + def test_v3_duration_column_backfills_and_p95_uses_covered_index(self): + with tempfile.TemporaryDirectory() as td: + path = Path(td) / "audit.sqlite3" + store = AuditStore(path) + base = time.time() - 3 * 86400 + for i in range(5): + self.assertTrue(store.record_request( + self.record(f"r{i}", started_at=base + i, duration_ms=100 + i))["recorded"]) + store.close() + with closing(sqlite3.connect(str(path))) as db: + db.execute("DROP INDEX requests_duration") + db.execute("ALTER TABLE requests DROP COLUMN duration_ms") + db.execute("PRAGMA user_version=3") + db.commit() + migrated = AuditStore(path) + try: + with closing(sqlite3.connect(str(path))) as db: + self.assertEqual(db.execute("PRAGMA user_version").fetchone()[0], SCHEMA_VERSION) + rows = [r[0] for r in db.execute("SELECT duration_ms FROM requests ORDER BY id").fetchall()] + self.assertEqual(rows, [100.0, 101.0, 102.0, 103.0, 104.0]) + with closing(migrated._open_read()) as plan_conn: + plan = plan_conn.execute( + "EXPLAIN QUERY PLAN SELECT duration_ms FROM requests WHERE started_at>=? AND started_at<=?", + (0, time.time())).fetchall() + self.assertIn("COVERING INDEX requests_duration", plan[0][3]) + summary = migrated.dashboard(30)["summary"] + self.assertEqual(summary["requests"], 5) + self.assertEqual(summary["p95_duration_ms"], 104) + finally: + migrated.close() def test_v1_ingest_table_migrates_with_backfilled_timestamps(self): """Migrate v1 ingest timestamps and indexes while preserving retained deduplication.""" with tempfile.TemporaryDirectory() as td: diff --git a/web/src/values.tsx b/web/src/values.tsx index 70c3e70..cd4ab0a 100644 --- a/web/src/values.tsx +++ b/web/src/values.tsx @@ -149,6 +149,8 @@ const labels: Record = { failure_count: "失败次数", dropped_records: "未写入记录数", recent_failures: "近期失败次数", + transient_contention: "近期锁竞争次数", + transient_count: "锁竞争累计次数", closed: "存储已关闭", budget_scope: "容量统计范围", fault_counter_scope: "故障计数范围", From 816d9f4c402466c12ef84820eb982b3f7e6eba20 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Thu, 15 Oct 2026 16:47:07 +0800 Subject: [PATCH 2/7] Drop an over-strict failure_count assertion from the storm regression A transient prune timeout during the concurrent mix is permitted by design; the contract under test is zero dropped records and no degraded latch. --- tests/test_audit_store.py | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/test_audit_store.py b/tests/test_audit_store.py index 0494584..f00bf9c 100644 --- a/tests/test_audit_store.py +++ b/tests/test_audit_store.py @@ -276,7 +276,6 @@ def traffic(tag): health = self.store.storage() self.assertFalse(health["degraded"]) self.assertEqual(health["dropped_records"], 0) - self.assertEqual(health["failure_count"], 0) def test_v3_duration_column_backfills_and_p95_uses_covered_index(self): with tempfile.TemporaryDirectory() as td: From 34f1d9e5756e210394624ae2fe9507c7fc66ab02 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Thu, 15 Oct 2026 16:56:42 +0800 Subject: [PATCH 3/7] Translate the remaining admin log action labels Add zh/en labels and value mappings for storage_archive, credentials.uploaded/exported, session.revoke_failed, and the credential.auto_checkin/auto_travel settings events. --- web/src/i18n.tsx | 12 ++++++++++++ web/src/values.tsx | 6 ++++++ 2 files changed, 18 insertions(+) diff --git a/web/src/i18n.tsx b/web/src/i18n.tsx index 6d187fe..5f8b46e 100644 --- a/web/src/i18n.tsx +++ b/web/src/i18n.tsx @@ -138,6 +138,8 @@ const zh: Record = { "logs.action.credential_checkin": "凭证签到", "logs.action.credential_daily_chat": "凭证每日对话", "logs.action.credential_auto_daily_chat": "凭证自动每日对话", + "logs.action.credential_auto_checkin": "凭证自动签到", + "logs.action.credential_auto_travel": "凭证自动旅行", "logs.action.credential_trial": "凭证试用", "logs.action.credential_enabled": "凭证启停", "logs.action.credential_reset_cooldown": "凭证清除冷却", @@ -146,6 +148,10 @@ const zh: Record = { "logs.action.model_created": "模型已创建", "logs.action.model_updated": "模型已更新", "logs.action.model_deleted": "模型已删除", + "logs.action.session_revoke_failed": "会话吊销失败", + "logs.action.credentials_uploaded": "凭证已上传", + "logs.action.credentials_exported": "凭证已导出", + "logs.action.storage_archive": "日志归档", "dash.refresh": "刷新", "dash.runtime": "运行概览", "dash.trend": "请求趋势", @@ -409,6 +415,8 @@ const en: Record = { "logs.action.credential_checkin": "Credential check-in", "logs.action.credential_daily_chat": "Credential daily chat", "logs.action.credential_auto_daily_chat": "Credential auto daily chat", + "logs.action.credential_auto_checkin": "Credential auto check-in", + "logs.action.credential_auto_travel": "Credential auto travel", "logs.action.credential_trial": "Credential trial", "logs.action.credential_enabled": "Credential enabled", "logs.action.credential_reset_cooldown": "Credential cooldown reset", @@ -417,6 +425,10 @@ const en: Record = { "logs.action.model_created": "Model created", "logs.action.model_updated": "Model updated", "logs.action.model_deleted": "Model deleted", + "logs.action.session_revoke_failed": "Session revoke failed", + "logs.action.credentials_uploaded": "Credentials uploaded", + "logs.action.credentials_exported": "Credentials exported", + "logs.action.storage_archive": "Log archive", "dash.refresh": "Refresh", "dash.runtime": "Runtime overview", "dash.trend": "Request trend", diff --git a/web/src/values.tsx b/web/src/values.tsx index cd4ab0a..f79df63 100644 --- a/web/src/values.tsx +++ b/web/src/values.tsx @@ -276,6 +276,8 @@ const eventActions: Record = { "credential.checkin": "logs.action.credential_checkin", "credential.daily-chat": "logs.action.credential_daily_chat", "credential.auto_daily_chat": "logs.action.credential_auto_daily_chat", + "credential.auto_checkin": "logs.action.credential_auto_checkin", + "credential.auto_travel": "logs.action.credential_auto_travel", "credential.trial": "logs.action.credential_trial", "credential.enabled": "logs.action.credential_enabled", "credential.reset-cooldown": "logs.action.credential_reset_cooldown", @@ -284,6 +286,10 @@ const eventActions: Record = { "model.created": "logs.action.model_created", "model.updated": "logs.action.model_updated", "model.deleted": "logs.action.model_deleted", + "session.revoke_failed": "logs.action.session_revoke_failed", + "credentials.uploaded": "logs.action.credentials_uploaded", + "credentials.exported": "logs.action.credentials_exported", + "storage_archive": "logs.action.storage_archive", }; export function eventActionLabel(action: unknown) { return typeof action === "string" ? ownLabel(eventActions, action) : undefined; From f752f60bc8da020c47b824dcaa38b6b8332753a4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Thu, 15 Oct 2026 16:56:42 +0800 Subject: [PATCH 4/7] Move the sidebar collapse control to a bottom bar Replace the small header icon button with a full-width pill bar pinned to the sidebar bottom (label plus panel icon, icon-only when collapsed), matching the existing glass panel styling. --- web/src/App.tsx | 23 ++++++++++++--------- web/src/ui.module.scss | 46 +++++++++++++++++++++++++++--------------- 2 files changed, 43 insertions(+), 26 deletions(-) diff --git a/web/src/App.tsx b/web/src/App.tsx index 0f594d7..e7f3fc5 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -79,16 +79,6 @@ function Shell() { CodeBuddy{t("app.brandSubtitle")} -

{t("app.workspace")}

+
diff --git a/web/src/ui.module.scss b/web/src/ui.module.scss index 5c95670..98f5132 100644 --- a/web/src/ui.module.scss +++ b/web/src/ui.module.scss @@ -1898,14 +1898,30 @@ $white: var(--page); .brandText { white-space: nowrap; } -.sidebarToggle { - padding: 6px; - min-width: 32px; - width: 32px; - min-height: 32px; - border-radius: 11px; +.sidebarBar { + margin-top: auto; + display: flex; + align-items: center; + justify-content: space-between; + gap: 10px; + width: 100%; + padding: 11px 14px; + border-radius: 12px; + border: 1px solid transparent; + background: #ffffff5c; + color: #7c8a72; + font-size: 12px; + letter-spacing: 0.3px; + cursor: pointer; + transition: background 140ms ease, border-color 140ms ease, color 140ms ease; +} +.sidebarBar:hover { + background: #ffffffa0; + border-color: #ffffffd0; + color: $ink; +} +.sidebarBar svg { flex-shrink: 0; - background: #ffffff55; } .brandIcon { flex-shrink: 0; @@ -1926,16 +1942,17 @@ $white: var(--page); .collapsed .sidebar { padding-inline: 10px; } -.collapsed .sidebarHead { - flex-direction: column; - gap: 16px; -} .collapsed .brandText, .collapsed .navLabel, .collapsed .navText, -.collapsed .navArrow { +.collapsed .navArrow, +.collapsed .sidebarBarText { display: none; } +.collapsed .sidebarBar { + justify-content: center; + padding: 11px 8px; +} .collapsed .sidebar nav { margin-top: 28px; } @@ -2346,7 +2363,7 @@ $white: var(--page); background: var(--soft); } .header button, -.sidebarToggle, +.sidebarBar, .drawerHead button { background: color-mix(in srgb, var(--surface) 70%, transparent); } @@ -2749,9 +2766,6 @@ $white: var(--page); .workspace { transition: none; } - .collapsed .sidebarHead { - flex-direction: row; - } .collapsed .sidebar nav { justify-content: space-between; margin-top: 12px; From 40f257b2a971a588cd0ea25803cd56e44c84fe3b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Thu, 15 Oct 2026 17:02:26 +0800 Subject: [PATCH 5/7] Collapse the mobile sidebar to its bottom bar On narrow screens the collapsed state now hides the brand row and nav strip entirely, leaving the labeled expand bar as the only remnant instead of an icon-only nav. Also apply vp check formatting to the recent sidebar and label edits. --- web/src/ui.module.scss | 21 ++++++++++++++++----- web/src/values.tsx | 2 +- 2 files changed, 17 insertions(+), 6 deletions(-) diff --git a/web/src/ui.module.scss b/web/src/ui.module.scss index 98f5132..2d05b60 100644 --- a/web/src/ui.module.scss +++ b/web/src/ui.module.scss @@ -1913,7 +1913,10 @@ $white: var(--page); font-size: 12px; letter-spacing: 0.3px; cursor: pointer; - transition: background 140ms ease, border-color 140ms ease, color 140ms ease; + transition: + background 140ms ease, + border-color 140ms ease, + color 140ms ease; } .sidebarBar:hover { background: #ffffffa0; @@ -2766,12 +2769,20 @@ $white: var(--page); .workspace { transition: none; } + /* Mobile collapse stows the whole panel: only the bottom bar remains, + so the workspace keeps the full viewport instead of an icon strip. */ + .collapsed .sidebar { + padding: 8px 10px; + } + .collapsed .sidebarHead, .collapsed .sidebar nav { - justify-content: space-between; - margin-top: 12px; + display: none; + } + .collapsed .sidebarBar { + margin-top: 0; } - .collapsed .navItem::after { - content: none; + .collapsed .sidebarBarText { + display: inline; } .sidebar { position: relative; diff --git a/web/src/values.tsx b/web/src/values.tsx index f79df63..970535b 100644 --- a/web/src/values.tsx +++ b/web/src/values.tsx @@ -289,7 +289,7 @@ const eventActions: Record = { "session.revoke_failed": "logs.action.session_revoke_failed", "credentials.uploaded": "logs.action.credentials_uploaded", "credentials.exported": "logs.action.credentials_exported", - "storage_archive": "logs.action.storage_archive", + storage_archive: "logs.action.storage_archive", }; export function eventActionLabel(action: unknown) { return typeof action === "string" ? ownLabel(eventActions, action) : undefined; From c55e4afb0990dd0b95094ea554702fe5348bbb8c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Thu, 15 Oct 2026 17:06:53 +0800 Subject: [PATCH 6/7] Update the mobile glass e2e for the stowed sidebar The collapsed mobile sidebar now hides the nav strip behind the expand bar; the test expands it before navigating and asserts the stow. --- web/e2e/experience.spec.ts | 3 +++ 1 file changed, 3 insertions(+) diff --git a/web/e2e/experience.spec.ts b/web/e2e/experience.spec.ts index 5897b2b..5c39439 100644 --- a/web/e2e/experience.spec.ts +++ b/web/e2e/experience.spec.ts @@ -222,6 +222,9 @@ test("daily range uses hourly points, with desktop and mobile glass layouts", as await expect(page.getByRole("img", { name: "按小时用量趋势" })).toBeVisible(); expect(await page.evaluate(() => document.documentElement.scrollWidth <= innerWidth)).toBe(true); await page.screenshot({ path: "test-results/glass-overview-mobile.png" }); + // The collapsed mobile sidebar stows the nav strip; only the expand bar remains. + await expect(page.getByRole("link", { name: "模型路由", exact: true })).toBeHidden(); + await page.getByRole("button", { name: "展开侧栏" }).click(); await page.getByRole("link", { name: "模型路由", exact: true }).click(); await page.getByRole("button", { name: "新增模型" }).click(); await page.getByRole("radio", { name: /指定账号/ }).check(); From 2a066cbea12821c26a20b8fba69fb2e29396a32a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Thu, 15 Oct 2026 17:14:01 +0800 Subject: [PATCH 7/7] Show credential action results inside the detail drawer Maintenance results, preference notices, busy state and errors now render at the top of the credential drawer while it is open, instead of behind the overlay on the main page; the page-level copies return when the drawer closes. --- web/src/pages/Credentials.tsx | 103 +++++++++++++++++++--------------- 1 file changed, 57 insertions(+), 46 deletions(-) diff --git a/web/src/pages/Credentials.tsx b/web/src/pages/Credentials.tsx index 4aea4d1..090b6ba 100644 --- a/web/src/pages/Credentials.tsx +++ b/web/src/pages/Credentials.tsx @@ -121,6 +121,51 @@ function ImportDrawer({ onClose, onDone }: { onClose: () => void; onDone: () => ); } +function MaintenanceResults({ results }: { results: Record[] }) { + return ( + +
    + {results.map((r, i) => { + const complete = + r.ok === true && + !(r.travel && typeof r.travel === "object" && object(r.travel).ok === false); + return ( +
  • + {text(r.name)} + + {complete + ? "已完成" + : r.checkin_ok === true || + r.claimed === true || + r.departed === true || + r.buddy_claimed === true || + r.agreement_accepted === true || + (r.travel && + typeof r.travel === "object" && + (object(r.travel).claimed === true || object(r.travel).departed === true)) + ? "部分完成" + : r.skipped + ? "已跳过" + : "未完成"} + + {text(r.message)} + +
    • + ); + })} +
+ + ); +} + type ChannelKey = "all" | "cn" | "intl" | "enterprise"; type ViewMode = "cards" | "list"; type StatusFilter = "all" | "ready" | "issue" | "disabled"; @@ -720,14 +765,14 @@ export function Credentials() { } /> - {notice && ( + {!detail && notice && (

{notice}

)} - {busy &&

操作执行中，请勿重复提交…

} + {!detail && busy &&

操作执行中，请勿重复提交…

} - + {!detail && } {resource.data && (
{channelDefs.map(([key, label, hint]) => { @@ -783,49 +828,7 @@ export function Credentials() { })}
)} - {maintenance.length > 0 && ( - -
    - {maintenance.map((r, i) => { - const complete = - r.ok === true && - !(r.travel && typeof r.travel === "object" && object(r.travel).ok === false); - return ( -
  • - {text(r.name)} - - {complete - ? "已完成" - : r.checkin_ok === true || - r.claimed === true || - r.departed === true || - r.buddy_claimed === true || - r.agreement_accepted === true || - (r.travel && - typeof r.travel === "object" && - (object(r.travel).claimed === true || - object(r.travel).departed === true)) - ? "部分完成" - : r.skipped - ? "已跳过" - : "未完成"} - - {text(r.message)} - -
    • - ); - })} -
- - )} + {!detail && maintenance.length > 0 && } setDetail(null)} > + {notice && ( +

+ {notice} +

+ )} + {busy &&

操作执行中，请勿重复提交…

} + + {maintenance.length > 0 && }
{renderHealthBadge(liveDetail)}