fix: move usage persistence off the inference response path

README.md

@@ -45,6 +45,29 @@ curl -X POST http://YOUR_SERVER_IP:8080/v1/systemone \
 
 Successful responses include `answers`, `model`, and actual token counts in `usage`. The published image includes the multilingual model; `model=auto` (the default) and `model=multilingual` are available. See **Documentation** in the console for more request examples.
 
+## Usage statistics and storage
+
+Inference responses enqueue usage statistics without waiting for SQLite writes. A single
+background thread per process reuses a connection, writes batches of up to 100 events,
+and coalesces `last_used_at` updates by API key. Partial batches flush after up to one
+second when the writer is idle; slow storage can delay statistics further. The `usage`
+field in each inference response is unchanged. `duration_ms` measures model prediction
+and result validation, not queue time or database writes.
+
+Statistics are best-effort, eventually consistent telemetry, not a billing or quota
+ledger. Each process buffers up to 1,000 queued events plus one in-flight batch. A full
+queue drops new statistics with rate-limited warnings. SQLite failures roll back where
+possible, log and drop the affected batch, then reconnect for later batches; failed
+batches are not retried because an uncertain commit could double-count usage. Neither
+condition turns successful inference into an HTTP 503. API-key authentication and
+revocation still use the database synchronously.
+
+Graceful shutdown attempts to drain the queue for up to five seconds. An in-flight
+SQLite call cannot be cancelled safely and may outlive that wait; forced termination,
+a shutdown timeout, or storage errors can lose unflushed statistics. Keep `/data` on
+low-latency persistent storage: background batching reduces write pressure but does
+not fix slow authentication reads or database locking.
+
 ## License and Feedback
 
 This project is licensed under [Apache License 2.0](LICENSE). For issues and suggestions, visit [GitHub Issues](https://github.com/1Panel-dev/laya-server/issues).

README.zh-CN.md

@@ -48,3 +48,21 @@ curl -X POST http://YOUR_SERVER_IP:8080/v1/systemone \
 ## 许可与反馈
 
 本项目采用 [Apache License 2.0](LICENSE) 许可证。问题和建议请提交至 [GitHub Issues](https://github.com/1Panel-dev/laya-server/issues)。
+
+## 用量统计与存储
+
+推理成功后将统计放入内存队列即可返回，不等待 SQLite 写入。每个进程使用一个后台
+线程复用连接，每批最多写入 100 条记录，并合并同一个 API key 的 `last_used_at`
+更新。写入线程空闲时，不满一批的记录最多等待 1 秒开始写入；存储缓慢时统计会进一步
+延迟。响应中的 `usage` 字段保持不变，`duration_ms` 只计算模型调用及结果校验时间，
+不包含排队和数据库写入时间。
+
+统计采用尽力记录、最终一致的方式，不适合作为计费或严格配额账本。每个进程最多缓存
+1,000 条待写记录，另有一个正在处理的批次。队列满时丢弃新统计并限频记录警告。
+SQLite 写入失败时尽可能回滚，记录错误并丢弃该批次，后续批次重新连接；不重试失败
+批次，以免提交结果不确定时重复计数。这些情况不会让成功的推理返回 HTTP 503。
+API key 验证及撤销仍同步访问数据库。
+
+正常停机时最多等待 5 秒清空队列。已经执行中的 SQLite 调用无法安全取消，可能超过
+该等待时间；强制终止、停机超时或存储错误可能导致尚未保存的统计丢失。`/data` 仍应
+使用低延迟的持久化存储，后台批量写入不能解决鉴权读取缓慢或数据库锁竞争问题。

backend/server/database.py

@@ -13,12 +13,20 @@ class Database:
     def __init__(self, path: Path):
         self.path = path
 
+    def open_connection(self) -> sqlite3.Connection:
+        connection = sqlite3.connect(self.path, timeout=10)
+        try:
+            connection.row_factory = sqlite3.Row
+            connection.execute("PRAGMA foreign_keys=ON")
+            connection.execute("PRAGMA busy_timeout=10000")
+        except sqlite3.Error:
+            connection.close()
+            raise
+        return connection
+
     @contextmanager
     def connect(self) -> Iterator[sqlite3.Connection]:
-        connection = sqlite3.connect(self.path, timeout=10)
-        connection.row_factory = sqlite3.Row
-        connection.execute("PRAGMA foreign_keys=ON")
-        connection.execute("PRAGMA busy_timeout=10000")
+        connection = self.open_connection()
         try:
             yield connection
             connection.commit()

backend/server/main.py

@@ -2,9 +2,9 @@
 import hmac
 import logging
 import secrets
-import sqlite3
 import threading
 import time
+from contextlib import asynccontextmanager
 from datetime import datetime, timedelta, timezone
 from typing import Annotated, Any
 
@@ -13,11 +13,13 @@
 from fastapi import Depends, FastAPI, Header, HTTPException, Request, Response
 from fastapi.exceptions import RequestValidationError
 from fastapi.responses import JSONResponse
+from starlette.concurrency import run_in_threadpool
 
 from .config import Settings
 from .database import Database, utc_now
 from .laya_adapter import LayaAdapter, Predictor
 from .schemas import CreateKeyRequest, InferenceRequest, LoginRequest
+from .usage_writer import UsageEvent, UsageWriter
 
 logger = logging.getLogger(__name__)
 SESSION_COOKIE = "laya_session"
@@ -36,7 +38,18 @@ def create_app(settings: Settings | None = None, predictor: Predictor | None = N
     settings = settings or Settings.from_env()
     db = Database(settings.database_path)
     db.initialize()
-    app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
+    usage_writer = UsageWriter(db)
+
+    @asynccontextmanager
+    async def lifespan(app: FastAPI):
+        usage_writer.start()
+        try:
+            yield
+        finally:
+            # Joining a slow writer must not block the event loop.
+            await run_in_threadpool(usage_writer.stop)
+
+    app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None, lifespan=lifespan)
     hasher = PasswordHasher()
     router_holder: dict[str, Predictor] = {}
     router_lock = threading.Lock()
@@ -129,17 +142,10 @@ def infer(payload: InferenceRequest, source: str, key_id: int | None) -> dict[st
             logger.error("Laya inference failed: %s", type(exc).__name__)
             raise error(503, "MODEL_UNAVAILABLE", "Model is unavailable")
         duration_ms = round((time.monotonic() - started) * 1000)
-        try:
-            with db.connect() as connection:
-                connection.execute(
-                    "INSERT INTO usage_events (created_at,source,key_id,model,input_tokens,output_tokens,duration_ms) VALUES (?,?,?,?,?,?,?)",
-                    (utc_now(), source, key_id, result["model"], usage["input_tokens"], usage["output_tokens"], duration_ms),
-                )
-                if key_id is not None:
-                    connection.execute("UPDATE api_keys SET last_used_at=? WHERE id=?", (utc_now(), key_id))
-        except sqlite3.Error:
-            logger.exception("Failed to persist usage")
-            raise error(503, "USAGE_UNAVAILABLE", "Usage could not be recorded")
+        usage_writer.submit(UsageEvent(
+            utc_now(), source, key_id, result["model"],
+            usage["input_tokens"], usage["output_tokens"], duration_ms,
+        ))
         return result
 
     @app.get("/health/live")

backend/server/usage_writer.py

@@ -0,0 +1,130 @@
+"""Bounded, best-effort usage telemetry; never part of inference success."""
+import logging
+import sqlite3
+import threading
+import time
+from collections import deque
+from dataclasses import dataclass
+
+from .database import Database
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class UsageEvent:
+    created_at: str
+    source: str
+    key_id: int | None
+    model: str
+    input_tokens: int
+    output_tokens: int
+    duration_ms: int
+
+
+class UsageWriter:
+    def __init__(self, db: Database, *, capacity: int = 1000, batch_size: int = 100,
+                 flush_interval: float = 1.0, shutdown_timeout: float = 5.0):
+        if min(capacity, batch_size, flush_interval, shutdown_timeout) <= 0:
+            raise ValueError("Usage writer limits must be positive")
+        self.db = db
+        self.capacity = capacity
+        self.batch_size = batch_size
+        self.flush_interval = flush_interval
+        self.shutdown_timeout = shutdown_timeout
+        self._condition = threading.Condition()
+        self._pending: deque[UsageEvent] = deque()
+        self._thread: threading.Thread | None = None
+        self._accepting = False
+        self._deadline: float | None = None
+        self.dropped = 0
+
+    def start(self) -> None:
+        with self._condition:
+            if self._thread is not None and self._thread.is_alive():
+                raise RuntimeError("Previous usage writer is still running")
+            self._deadline = None
+            self._accepting = True
+            self._thread = threading.Thread(target=self._run, name="laya-usage-writer", daemon=True)
+            self._thread.start()
+
+    def submit(self, event: UsageEvent) -> bool:
+        with self._condition:
+            if not self._accepting or len(self._pending) >= self.capacity:
+                self.dropped += 1
+                # Avoid flooding logs when storage cannot keep up.
+                if self.dropped == 1 or self.dropped & (self.dropped - 1) == 0:
+                    logger.warning("Usage queue unavailable/full; dropped events=%d", self.dropped)
+                return False
+            self._pending.append(event)
+            self._condition.notify()
+            return True
+
+    def stop(self) -> None:
+        with self._condition:
+            self._accepting = False
+            self._deadline = time.monotonic() + self.shutdown_timeout
+            self._condition.notify_all()
+        if self._thread is not None:
+            self._thread.join(self.shutdown_timeout)
+            if self._thread.is_alive():
+                logger.warning("Usage shutdown timed out; unflushed statistics may be lost")
+
+    def _next_batch(self) -> list[UsageEvent]:
+        with self._condition:
+            while not self._pending and self._accepting:
+                self._condition.wait()
+            flush_at = time.monotonic() + self.flush_interval
+            while self._accepting and len(self._pending) < self.batch_size:
+                remaining = flush_at - time.monotonic()
+                if remaining <= 0:
+                    break
+                self._condition.wait(remaining)
+            if self._deadline is not None and time.monotonic() >= self._deadline:
+                self.dropped += len(self._pending)
+                self._pending.clear()
+            return [self._pending.popleft() for _ in range(min(self.batch_size, len(self._pending)))]
+
+    @staticmethod
+    def _write_batch(connection: sqlite3.Connection, batch: list[UsageEvent]) -> None:
+        latest: dict[int, str] = {}
+        for event in batch:
+            if event.key_id is not None:
+                latest[event.key_id] = max(latest.get(event.key_id, ""), event.created_at)
+        with connection:
+            connection.executemany(
+                "INSERT INTO usage_events (created_at,source,key_id,model,input_tokens,output_tokens,duration_ms) "
+                "VALUES (?,?,?,?,?,?,?)",
+                [(e.created_at, e.source, e.key_id, e.model, e.input_tokens, e.output_tokens, e.duration_ms)
+                 for e in batch],
+            )
+            connection.executemany(
+                "UPDATE api_keys SET last_used_at=MAX(COALESCE(last_used_at,''),?) WHERE id=?",
+                [(created_at, key_id) for key_id, created_at in latest.items()],
+            )
+
+    def _run(self) -> None:
+        connection = None
+        try:
+            while batch := self._next_batch():
+                try:
+                    if connection is None:
+                        connection = self.db.open_connection()
+                    self._write_batch(connection, batch)
+                except sqlite3.Error:
+                    with self._condition:
+                        self.dropped += len(batch)
+                    logger.exception("Failed to persist %d usage events; dropping batch", len(batch))
+                    # Do not retry an ambiguously committed batch and double-count it.
+                    if connection is not None:
+                        connection.close()
+                        connection = None
+        finally:
+            with self._condition:
+                self._accepting = False
+                if self._pending:
+                    self.dropped += len(self._pending)
+                    logger.warning("Usage writer stopped; discarding %d pending events", len(self._pending))
+                    self._pending.clear()
+            if connection is not None:
+                connection.close()

backend/tests/conftest.py

@@ -0,0 +1,21 @@
+import time
+from contextlib import ExitStack
+
+import pytest
+from fastapi.testclient import TestClient
+
+
+@pytest.fixture
+def client_factory():
+    with ExitStack() as stack:
+        yield lambda *args, **kwargs: stack.enter_context(TestClient(*args, **kwargs))
+
+
+def wait_for_usage(client, requests):
+    deadline = time.monotonic() + 5
+    while True:
+        usage = client.get('/internal/usage').json()
+        if usage['totals']['requests'] == requests:
+            return usage
+        assert time.monotonic() < deadline, usage
+        time.sleep(0.01)