From eda37a6ac38e7bd654eeaeb9055f1f203f0ba82a Mon Sep 17 00:00:00 2001
From: cat <1716967236@qq.com>
Date: Fri, 18 Sep 2026 22:40:45 +0800
Subject: [PATCH] =?UTF-8?q?perf:=20SQLite=20=E5=BC=80=20WAL=20=E4=B8=8E?=
 =?UTF-8?q?=E5=86=99=E9=94=81=E7=AD=89=E5=BE=85=EF=BC=8C=E6=94=AF=E6=8C=81?=
 =?UTF-8?q?=E6=B5=81=E6=B0=B4=E7=BA=BF=E5=B9=B6=E5=8F=91=E5=86=99?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

批量流水线会在多个线程里同时写产物与进度，默认的回滚日志模式与 5 秒
busy timeout 下容易出现 database is locked。

- 连接统一带 `WOV_DB_BUSY_TIMEOUT_SECONDS`（默认 30 秒）等待写锁。
- 打开 WAL 与 `synchronous=NORMAL`：读写并行，落盘开销更低；不支持 WAL 的
  文件系统会保持原模式（SQLite 自身降级）。
- 测试：断言 WAL 生效，并用 4 线程 × 25 条并发写产物验证不丢数据、不报错。
---
 src/wov_app/config.py              |  4 ++
 src/wov_app/db.py                  | 13 ++++++-
 tests/app/test_db/test_database.py | 59 ++++++++++++++++++++++++++++++
 3 files changed, 74 insertions(+), 2 deletions(-)

diff --git a/src/wov_app/config.py b/src/wov_app/config.py
index 4d5fe95..8591dce 100644
--- a/src/wov_app/config.py
+++ b/src/wov_app/config.py
@@ -17,6 +17,10 @@ DATA_DIR = Path(os.getenv("WOV_DATA_DIR", str(WORKSPACE_ROOT / "data")))
 DB_PATH = Path(os.getenv("WOV_DB_PATH", str(DATA_DIR / "wov.db")))
 STORAGE_DIR = Path(os.getenv("WOV_STORAGE_DIR", str(DATA_DIR / "storage")))
 
+# SQLite 写锁等待时长（秒）：批量流水线多线程并发写产物/进度，短时竞争要排队
+# 而不是立刻抛 database is locked。
+DB_BUSY_TIMEOUT_SECONDS = float(os.getenv("WOV_DB_BUSY_TIMEOUT_SECONDS", "30"))
+
 # 调度器轮询排队任务的间隔（秒）。
 SCHEDULER_INTERVAL_SECONDS = float(os.getenv("WOV_SCHEDULER_INTERVAL_SECONDS", "1.0"))
 
diff --git a/src/wov_app/db.py b/src/wov_app/db.py
index 357997f..2f1a01d 100644
--- a/src/wov_app/db.py
+++ b/src/wov_app/db.py
@@ -13,6 +13,8 @@ from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Iterator
 
+from wov_app.config import DB_BUSY_TIMEOUT_SECONDS
+
 
 def _now_iso() -> str:
     """返回当前 UTC 时间的 ISO 格式字符串（与批量引擎的时间戳一致）。"""
@@ -30,12 +32,19 @@ class Database:
 
     @contextmanager
     def _connect(self) -> Iterator[sqlite3.Connection]:
-        """提供带事务提交的数据库连接上下文。"""
-        conn = sqlite3.connect(self.path)
+        """提供带事务提交的数据库连接上下文。
+
+        批量流水线会在多个线程里并发写产物与进度：开 WAL 让读写并行，busy
+        timeout 让短时写锁竞争排队等待，而不是直接抛 database is locked。
+        """
+        conn = sqlite3.connect(self.path, timeout=DB_BUSY_TIMEOUT_SECONDS)
         # 按列名读取结果，返回 dict 更直观。
         conn.row_factory = sqlite3.Row
         # 开启外键约束，保证子表记录引用有效。
         conn.execute("PRAGMA foreign_keys = ON")
+        # WAL 是持久设置，重复执行无副作用；不支持 WAL 的文件系统会保持原模式。
+        conn.execute("PRAGMA journal_mode = WAL")
+        conn.execute("PRAGMA synchronous = NORMAL")
         try:
             yield conn
             conn.commit()
diff --git a/tests/app/test_db/test_database.py b/tests/app/test_db/test_database.py
index 280725b..5b9be2c 100644
--- a/tests/app/test_db/test_database.py
+++ b/tests/app/test_db/test_database.py
@@ -381,3 +381,62 @@ def test_recover_interrupted_batch_jobs_requeues_zombie_completed(db_with_workfl
     assert count == 1
     assert db_with_workflow.get_batch_job("zombie")["status"] == "QUEUED"
     assert db_with_workflow.get_batch_job("done")["status"] == "COMPLETED"
+
+
+def test_wal_mode_enabled_for_concurrent_pipeline_writes(tmp_path: Path) -> None:
+    """数据：临时库文件。
+
+    过程：打开一次连接并读取日志模式。
+
+    验证：处于 WAL 模式——批量流水线多线程并发写产物/进度时，读写可并行，
+    不会互相阻塞成 database is locked。
+    """
+    db = Database(tmp_path / "wov.db")
+
+    with db._connect() as conn:
+        mode = conn.execute("PRAGMA journal_mode").fetchone()[0]
+
+    assert mode.lower() == "wal"
+
+
+def test_concurrent_artifact_writes_survive(tmp_path: Path) -> None:
+    """数据：四个线程各自往同一任务写入 25 条产物记录。
+
+    过程：并发调用 create_artifact（批量流水线里多个视频同时落产物的场景）。
+
+    验证：全部写入成功且记录数正确，没有因写锁竞争丢数据或抛异常。
+    """
+    import threading
+
+    db = Database(tmp_path / "wov.db")
+    db.upsert_workflow({"id": "wf", "name": "流程", "description": "", "published": 1, "latest_version": 1})
+    db.create_workflow_version("wf", 1, {"nodes": []})
+    db.create_run({
+        "id": "run-concurrent", "workflow_id": "wf", "workflow_version": 1,
+        "status": "RUNNING", "source": "batch", "created_at": "t1", "updated_at": "t1",
+    })
+    errors: list[Exception] = []
+
+    def writer(worker: int) -> None:
+        try:
+            for index in range(25):
+                db.create_artifact({
+                    "run_id": "run-concurrent", "node_id": "asr",
+                    "name": f"w{worker}-{index}", "uri": f"/tmp/w{worker}-{index}.srt",
+                    "mime_type": "text/plain", "size": index,
+                })
+        except Exception as exc:  # noqa: BLE001 - 用例要报告任意写失败
+            errors.append(exc)
+
+    threads = [threading.Thread(target=writer, args=(n,)) for n in range(4)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+
+    assert errors == []
+    with db._connect() as conn:
+        count = conn.execute(
+            "SELECT COUNT(*) FROM artifacts WHERE run_id = ?", ("run-concurrent",),
+        ).fetchone()[0]
+    assert count == 100