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..f00bf9c 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,90 @@ 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) + + 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/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(); 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/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/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)} diff --git a/web/src/ui.module.scss b/web/src/ui.module.scss index 5c95670..2d05b60 100644 --- a/web/src/ui.module.scss +++ b/web/src/ui.module.scss @@ -1898,14 +1898,33 @@ $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 +1945,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 +2366,7 @@ $white: var(--page); background: var(--soft); } .header button, -.sidebarToggle, +.sidebarBar, .drawerHead button { background: color-mix(in srgb, var(--surface) 70%, transparent); } @@ -2749,15 +2769,20 @@ $white: var(--page); .workspace { transition: none; } - .collapsed .sidebarHead { - flex-direction: row; + /* 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 70c3e70..970535b 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: "故障计数范围", @@ -274,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", @@ -282,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;