Complete SQLite WAL maintenance before executor final states

executor/database_maintenance.py

@@ -0,0 +1,75 @@
+# Executor 的 SQLite WAL 收尾与共享数据库使用协调。
+
+from __future__ import annotations
+
+import sqlite3
+import stat
+import threading
+from contextlib import closing
+from pathlib import Path
+
+
+def checkpoint_database(database: Path) -> dict[str, object]:
+    """对已有数据库执行有限等待的 TRUNCATE，返回维护结果及具体 SQLite 原因。"""
+    try:
+        if any(parent.is_symlink() for parent in database.absolute().parents):
+            return {"status": "failed", "code": "wal_checkpoint_path_invalid", "message": "数据库目录不能包含符号链接"}
+        for path in (database, Path(str(database) + "-wal"), Path(str(database) + "-shm")):
+            if path.is_symlink():
+                return {"status": "failed", "code": "wal_checkpoint_path_invalid", "message": "数据库及 WAL/SHM 文件不能是符号链接"}
+        if database.exists() and not stat.S_ISREG(database.stat().st_mode):
+            return {"status": "failed", "code": "wal_checkpoint_path_invalid", "message": "数据库路径必须是普通文件"}
+        # mode=rw 禁止维护操作创建数据库；连接关闭前必须读取 checkpoint 返回值。
+        with closing(sqlite3.connect(database.absolute().as_uri() + "?mode=rw", uri=True, timeout=1.0)) as connection:
+            busy, log_pages, checkpointed_pages = connection.execute("PRAGMA wal_checkpoint(TRUNCATE)").fetchone()
+        return {
+            "status": "succeeded" if busy == 0 else "failed",
+            "code": "wal_checkpoint_complete" if busy == 0 else "wal_checkpoint_busy",
+            "busy": busy,
+            "log_pages": log_pages,
+            "checkpointed_pages": checkpointed_pages,
+        }
+    except sqlite3.Error as exc:
+        # 维护故障独立于业务结果，保留 SQLite 的错误类别和具体消息。
+        return {
+            "status": "failed", "code": "wal_checkpoint_sqlite_error",
+            "sqlite_errorcode": getattr(exc, "sqlite_errorcode", None),
+            "sqlite_errorname": getattr(exc, "sqlite_errorname", None),
+            "message": str(exc),
+        }
+    except OSError as exc:
+        return {"status": "failed", "code": "wal_checkpoint_filesystem_error", "errno": exc.errno, "message": str(exc)}
+
+
+class DatabaseMaintenance:
+    """记录 Executor 内数据库使用数，最后一个已回收进程负责 WAL 收尾。"""
+
+    def __init__(self) -> None:
+        """初始化使用数和收尾集合，阻止维护期间启动同库进程。"""
+        self._condition = threading.Condition()
+        self._users: dict[Path, int] = {}
+        self._checkpointing: set[Path] = set()
+
+    def acquire(self, database: Path) -> None:
+        """在启动进程之前注册数据库使用，允许同库任务正常并发执行。"""
+        with self._condition:
+            self._condition.wait_for(lambda: database not in self._checkpointing)
+            self._users[database] = self._users.get(database, 0) + 1
+
+    def release(self, database: Path, *, process_reaped: bool) -> dict[str, object]:
+        """进程回收后释放使用数并维护空闲数据库；未知进程继续占用注册。"""
+        with self._condition:
+            if not process_reaped:
+                return {"status": "skipped", "code": "wal_checkpoint_process_unreaped"}
+            users = self._users[database] - 1
+            if users:
+                self._users[database] = users
+                return {"status": "deferred", "code": "wal_checkpoint_database_in_use"}
+            del self._users[database]
+            self._checkpointing.add(database)
+        try:
+            return checkpoint_database(database)
+        finally:
+            with self._condition:
+                self._checkpointing.remove(database)
+                self._condition.notify_all()

executor/server.py

@@ -67,6 +67,7 @@ from executor.analysis_policy import AnalysisPolicyError, parse_analysis_policy
 from executor.analysis_monitor import AnalysisMonitor, AnalysisControlError
 from executor.attempt_diagnostics import AttemptOutput, configure_executor_diagnostics, stream_sample as _stream_sample
 from executor.process_supervisor import ProcessSupervisor
+from executor.database_maintenance import DatabaseMaintenance
 from executor.result_store import ExecutorResultStore, ExecutorResultStoreError
 from executor.task_queue import ExecutionQueue
 from executor.token import ExecutorTokenError, ensure_token_file, read_token_file
@@ -361,6 +362,7 @@ class TaskState:
     observed_cost: str | None = None
     usage_checked_at: str | None = None
     reminder_sent: bool = False
+    database_maintenance: dict[str, object] | None = None
 
     def public(self) -> dict[str, object]:
         """返回 API 可见状态，不暴露进程对象和 executor 本地实现细节。"""
@@ -385,6 +387,7 @@ class TaskState:
             "observed_cost": self.observed_cost,
             "usage_checked_at": self.usage_checked_at,
             "reminder_sent": self.reminder_sent,
+            "database_maintenance": self.database_maintenance,
         }
 
 
@@ -448,6 +451,7 @@ class Executor:
         # 控制面维护，避免 HTTP 入口直接持有线程池细节。
         self.pool = ExecutionQueue(self.max_workers, thread_name_prefix="mememeow-opencode")
         self.process_supervisor = ProcessSupervisor()
+        self.database_maintenance = DatabaseMaintenance()
         self.futures: dict[str, Future[None]] = {}
         self.ready_error: str | None = None
         self._prepare_runtime()
@@ -741,6 +745,7 @@ class Executor:
             "observed_cost": task.observed_cost,
             "usage_checked_at": task.usage_checked_at,
             "reminder_sent": task.reminder_sent,
+            "database_maintenance": task.database_maintenance,
         }
         values = [item for item in existing if item.get("executor_attempt_id") != task.executor_attempt_id]
         values.append(entry)
@@ -985,7 +990,7 @@ class Executor:
         source = max(matches, key=lambda task: (task.completed_at or task.created_at, task.created_at))
         if source.status != "failed":
             raise RuntimeError("session_not_resumable")
-        if source.process_reaped is not True:
+        if source.process_reaped is not True or not source.done.is_set():
             raise RuntimeError("session_not_resumable")
         source_error = (source.error or {}).get("error")
         if source_error not in {
@@ -1309,6 +1314,9 @@ class Executor:
         analysis_socket: socket.socket | None = None
         output = AttemptOutput()
         monitor = None
+        database: Path | None = None
+        final_status = "failed"
+        final_error: dict[str, object] | None = None
         started = time.monotonic()
         diagnostic_secrets = secret_inventory_from_mapping(os.environ) + (
             self.legacy_api_key, self.token, task.callback_token or "",
@@ -1318,10 +1326,8 @@ class Executor:
         try:
             with self.lock:
                 if task.cancel_event.is_set():
-                    task.status = "cancelled"
-                    task.error = _json_error("task_interrupted", "任务已取消")
-                    task.completed_at = time.time()
-                    task.done.set()
+                    final_status = "cancelled"
+                    final_error = _json_error("task_interrupted", "任务已取消")
                     return
                 self._verify_task_workspace_capability(task)
                 task.status = "running"
@@ -1391,6 +1397,8 @@ class Executor:
             command = self._sandbox_command(task, command)
             output.phase = "process_start"
             with output.capture(LOG_ROOT, task.task_id) as (out, err):
+                database = Path(env["OPENCODE_DB"])
+                self.database_maintenance.acquire(database)
                 process = subprocess.Popen(command, cwd=process_directory, env=env, stdin=subprocess.DEVNULL, stdout=out, stderr=err, start_new_session=True)
                 output.phase = "process_running"
                 with self.lock:
@@ -1467,15 +1475,14 @@ class Executor:
             with self.lock:
                 if task.cancel_event.is_set():
                     raise RuntimeError("task_interrupted")
-                task.status = "succeeded"
-                task.completed_at = time.time()
+                final_status = "succeeded"
         except _ProcessFailure as exc:
             with self.lock:
                 # 未确认进程已回收时，取消/超时也不能伪装成可安全重试的普通失败。
                 code = exc.code
                 if task.cancel_event.is_set() and code != "unknown_execution":
                     code = "task_interrupted"
-                task.status = "cancelled" if code == "task_interrupted" else "failed"
+                final_status = "cancelled" if code == "task_interrupted" else "failed"
                 error: dict[str, object] = {
                     "error": code,
                     "message": _redact_diagnostic(exc.args[0], (self.legacy_api_key, self.token, task.callback_token or "")),
@@ -1484,8 +1491,7 @@ class Executor:
                     error["analysis_diagnostic"] = exc.analysis_diagnostic
                 if exc.http_status is not None:
                     error["http_status"] = exc.http_status
-                task.error = error
-                task.completed_at = time.time()
+                final_error = error
         except RuntimeError as exc:
             code, _, detail = str(exc).partition(":")
             reason_code = getattr(exc, "reason_code", None)
@@ -1500,34 +1506,36 @@ class Executor:
                     code, detail = "task_interrupted", ""
                 if code not in _EXECUTOR_ERROR_CODES:
                     code = "agent_process_failed"
-                task.status = "cancelled" if code == "task_interrupted" else "failed"
+                final_status = "cancelled" if code == "task_interrupted" else "failed"
                 fallback = {
                     "agent_timeout": "OpenCode 执行超时",
                     "task_interrupted": "任务已取消",
                 }.get(code, "任务执行失败")
-                task.error = _json_error(
+                final_error = _json_error(
                     code,
                     _redact_diagnostic(detail, (self.legacy_api_key, self.token, task.callback_token or "")) if detail else fallback,
                     reason_code=reason_code,
                 )
-                task.completed_at = time.time()
         except (OSError, ValueError, json.JSONDecodeError) as exc:
             with self.lock:
                 if process is not None and process.poll() is not None:
                     task.process_reaped = True
                 code = "unknown_execution" if task.process_reaped is not True else "task_interrupted" if task.cancel_event.is_set() else "agent_process_failed"
-                task.status = "cancelled" if code == "task_interrupted" else "failed"
-                task.error = _json_error(code, "任务已取消" if code == "task_interrupted" else "无法确认 OpenCode 进程已终止" if code == "unknown_execution" else _redact_diagnostic(str(exc), (self.legacy_api_key, self.token, task.callback_token or "")))
-                task.completed_at = time.time()
+                final_status = "cancelled" if code == "task_interrupted" else "failed"
+                final_error = _json_error(code, "任务已取消" if code == "task_interrupted" else "无法确认 OpenCode 进程已终止" if code == "unknown_execution" else _redact_diagnostic(str(exc), (self.legacy_api_key, self.token, task.callback_token or "")))
         except Exception as exc:  # noqa: BLE001 - 任务必须以终态收束，不能留下永久 running
             with self.lock:
                 if process is not None and process.poll() is not None:
                     task.process_reaped = True
                 code = "unknown_execution" if task.process_reaped is not True else "task_interrupted" if task.cancel_event.is_set() else "agent_process_failed"
-                task.status = "cancelled" if code == "task_interrupted" else "failed"
-                task.error = _json_error(code, "任务已取消" if code == "task_interrupted" else "无法确认 OpenCode 进程已终止" if code == "unknown_execution" else _redact_diagnostic(str(exc), (self.legacy_api_key, self.token, task.callback_token or "")))
-                task.completed_at = time.time()
+                final_status = "cancelled" if code == "task_interrupted" else "failed"
+                final_error = _json_error(code, "任务已取消" if code == "task_interrupted" else "无法确认 OpenCode 进程已终止" if code == "unknown_execution" else _redact_diagnostic(str(exc), (self.legacy_api_key, self.token, task.callback_token or "")))
         finally:
+            if process is not None:
+                task.process_reaped = self.process_supervisor.terminate(process).reaped
+                if not task.process_reaped:
+                    final_status = "failed"
+                    final_error = _json_error("unknown_execution", "无法确认 OpenCode 进程已终止")
             if analysis_socket is not None:
                 analysis_socket.close()
                 socket_path = ANALYSIS_SOCKET_ROOT / f"{task.executor_attempt_id}.sock"
@@ -1535,8 +1543,24 @@ class Executor:
                     socket_path.unlink()
                 except FileNotFoundError:
                     pass
+            if database is not None:
+                task.database_maintenance = self.database_maintenance.release(database, process_reaped=task.process_reaped)
+                diagnostic = dict(task.database_maintenance)
+                if "message" in diagnostic:
+                    diagnostic["message"] = _redact_diagnostic(str(diagnostic["message"]), diagnostic_secrets)
+                    task.database_maintenance = diagnostic
+                logger.bind(attempt_id=task.executor_attempt_id, phase="database_checkpoint", **diagnostic).log(
+                    "WARNING" if diagnostic["status"] in {"failed", "skipped"} else "INFO",
+                    "OpenCode 数据库收尾：attempt={}；结果={}", task.executor_attempt_id, diagnostic,
+                )
             with self.lock:
                 task.process = None
+                if task.cancel_event.is_set() and (final_error or {}).get("error") != "unknown_execution":
+                    final_status = "cancelled"
+                    final_error = _json_error("task_interrupted", "任务已取消")
+                task.status = final_status
+                task.error = final_error
+                task.completed_at = time.time()
                 try:
                     self._persist_attempt_metadata(task)
                 except (OSError, ValueError, TypeError):
@@ -1679,19 +1703,12 @@ class Executor:
             task.cancel_event.set()
             was_queued = task.status == "queued"
             process = task.process
-            task.status = "cancelled"
-            task.error = _json_error("task_interrupted", "任务已取消")
-            task.completed_at = time.time()
             if process is None and was_queued:
+                task.status = "cancelled"
+                task.error = _json_error("task_interrupted", "任务已取消")
+                task.completed_at = time.time()
                 task.done.set()
-        if process is not None:
-            reaped = self.process_supervisor.terminate(process).reaped
-            with self.lock:
-                task.process_reaped = reaped
-                if not reaped:
-                    task.status = "failed"
-                    task.error = _json_error("unknown_execution", "无法确认 OpenCode 进程已终止")
-                    task.completed_at = time.time()
+        # 运行任务由执行线程终止并完成数据库维护，再统一发布最终状态。
         return task
 
     def close(self) -> None:

tests/support/wal_writer.py

@@ -0,0 +1,15 @@
+# 数据库维护测试使用的真实 SQLite 写入进程，提交数据后等待父进程终止。
+
+import sqlite3
+import sys
+
+
+connection = sqlite3.connect(sys.argv[1])
+connection.execute("PRAGMA journal_mode=WAL")
+connection.execute("PRAGMA wal_autocheckpoint=0")
+connection.execute("CREATE TABLE evidence (value TEXT)")
+connection.execute("INSERT INTO evidence VALUES ('committed')")
+connection.commit()
+print("committed", flush=True)
+sys.stdin.read()
+connection.close()

tests/test_database_maintenance.py

@@ -0,0 +1,169 @@
+# 使用真实 SQLite 连接和进程验证 Executor 数据库收尾。
+
+import sqlite3
+import subprocess
+import sys
+import threading
+import time
+from contextlib import closing
+from pathlib import Path
+
+from executor.database_maintenance import DatabaseMaintenance, checkpoint_database
+from executor.process_supervisor import ProcessSupervisor
+
+
+def test_checkpoint_preserves_committed_rows_and_truncates_wal(tmp_path: Path) -> None:
+    """真实 WAL 中已提交的数据在 TRUNCATE 后仍然完整可读。"""
+    database = tmp_path / "opencode.db"
+    with closing(sqlite3.connect(database)) as writer:
+        writer.execute("PRAGMA journal_mode=WAL")
+        writer.execute("CREATE TABLE evidence (value TEXT)")
+        writer.execute("INSERT INTO evidence VALUES (?)", ("checkpoint-evidence",))
+        writer.commit()
+        wal = Path(str(database) + "-wal")
+        assert wal.stat().st_size > 0
+        result = checkpoint_database(database)
+        assert result == {
+            "status": "succeeded", "code": "wal_checkpoint_complete",
+            "busy": 0, "log_pages": 0, "checkpointed_pages": 0,
+        }
+        assert wal.stat().st_size == 0
+        assert writer.execute("SELECT value FROM evidence").fetchone() == ("checkpoint-evidence",)
+
+
+def test_busy_reader_returns_bounded_diagnostic(tmp_path: Path) -> None:
+    """读取事务占用旧快照时返回 busy 页数，释放读取事务后可以重新维护。"""
+    database = tmp_path / "opencode.db"
+    with closing(sqlite3.connect(database)) as writer, closing(sqlite3.connect(database)) as reader:
+        writer.execute("PRAGMA journal_mode=WAL")
+        writer.execute("CREATE TABLE evidence (value INTEGER)")
+        writer.commit()
+        reader.execute("BEGIN")
+        reader.execute("SELECT * FROM evidence").fetchall()
+        writer.execute("INSERT INTO evidence VALUES (1)")
+        writer.commit()
+        started = time.monotonic()
+        result = checkpoint_database(database)
+        assert time.monotonic() - started < 3
+        assert result["code"] == "wal_checkpoint_busy"
+        assert result["status"] == "failed"
+        assert result["busy"] == 1
+        assert result["log_pages"] > result["checkpointed_pages"]
+        reader.rollback()
+        assert checkpoint_database(database)["status"] == "succeeded"
+
+
+def test_missing_and_invalid_database_preserve_sqlite_reason(tmp_path: Path) -> None:
+    """缺失文件不被创建，损坏文件返回 SQLite 原始类别和具体消息。"""
+    database = tmp_path / "opencode.db"
+    result = checkpoint_database(database)
+    assert result["sqlite_errorname"] == "SQLITE_CANTOPEN"
+    assert result["message"]
+    assert not database.exists()
+    database.write_bytes(b"invalid sqlite database" * 64)
+    result = checkpoint_database(database)
+    assert result["code"] == "wal_checkpoint_sqlite_error"
+    assert result["sqlite_errorname"] == "SQLITE_NOTADB"
+    assert result["message"] == "file is not a database"
+
+
+def test_shared_database_checkpoints_only_after_last_user(tmp_path: Path) -> None:
+    """共享数据库最后一个使用者释放之前保留 WAL，未回收进程继续占用注册。"""
+    database = tmp_path / "opencode.db"
+    maintenance = DatabaseMaintenance()
+    with closing(sqlite3.connect(database)) as writer:
+        writer.execute("PRAGMA journal_mode=WAL")
+        writer.execute("CREATE TABLE evidence (value INTEGER)")
+        writer.commit()
+        maintenance.acquire(database)
+        maintenance.acquire(database)
+        assert maintenance.release(database, process_reaped=True)["status"] == "deferred"
+        assert Path(str(database) + "-wal").stat().st_size > 0
+        assert maintenance.release(database, process_reaped=True)["status"] == "succeeded"
+        assert Path(str(database) + "-wal").stat().st_size == 0
+        maintenance.acquire(database)
+        assert maintenance.release(database, process_reaped=False)["code"] == "wal_checkpoint_process_unreaped"
+        maintenance.acquire(database)
+        assert maintenance.release(database, process_reaped=True)["status"] == "deferred"
+
+
+def test_process_exit_then_checkpoint(tmp_path: Path) -> None:
+    """真实 SQLite 写入进程被终止并回收后，维护连接回收 WAL 空间且保留数据。"""
+    database = tmp_path / "opencode.db"
+    maintenance = DatabaseMaintenance()
+    maintenance.acquire(database)
+    process = subprocess.Popen(
+        [sys.executable, str(Path(__file__).parent / "support" / "wal_writer.py"), str(database)],
+        stdout=subprocess.PIPE, stdin=subprocess.PIPE, text=True, start_new_session=True,
+    )
+    try:
+        assert process.stdout.readline().strip() == "committed"
+        assert Path(str(database) + "-wal").stat().st_size > 0
+        termination = ProcessSupervisor().terminate(process)
+        assert termination.reaped
+        assert maintenance.release(database, process_reaped=termination.reaped)["status"] == "succeeded"
+        wal = Path(str(database) + "-wal")
+        assert not wal.exists() or wal.stat().st_size == 0
+        with closing(sqlite3.connect(database)) as reader:
+            assert reader.execute("SELECT value FROM evidence").fetchall() == [("committed",)]
+    finally:
+        ProcessSupervisor().terminate(process)
+        process.stdout.close()
+        process.stdin.close()
+
+
+def test_registration_waits_for_running_checkpoint(tmp_path: Path) -> None:
+    """真实写入锁使 checkpoint 等待期间，新的同库进程注册也必须等待。"""
+    database = tmp_path / "opencode.db"
+    maintenance = DatabaseMaintenance()
+    with closing(sqlite3.connect(database)) as writer:
+        writer.execute("PRAGMA journal_mode=WAL")
+        writer.execute("CREATE TABLE evidence (value INTEGER)")
+        writer.commit()
+        writer.execute("BEGIN IMMEDIATE")
+        maintenance.acquire(database)
+        results = []
+        finished = threading.Event()
+
+        def release() -> None:
+            """让真实 checkpoint 在写入锁上等待，收集实际返回结果。"""
+            results.append(maintenance.release(database, process_reaped=True))
+
+        def acquire() -> None:
+            """注册后记录完成事件，验证注册与收尾操作的顺序。"""
+            maintenance.acquire(database)
+            finished.set()
+
+        checkpoint_thread = threading.Thread(target=release)
+        checkpoint_thread.start()
+        deadline = time.monotonic() + 2
+        while database not in maintenance._checkpointing and time.monotonic() < deadline:
+            time.sleep(0.001)
+        assert database in maintenance._checkpointing
+        # 其他任务的独立数据库不应被当前数据库的 SQLite 等待阻止。
+        other_database = tmp_path / "other.db"
+        with closing(sqlite3.connect(other_database)) as other:
+            other.execute("CREATE TABLE evidence (value INTEGER)")
+        maintenance.acquire(other_database)
+        assert maintenance.release(other_database, process_reaped=True)["status"] == "succeeded"
+        start_thread = threading.Thread(target=acquire)
+        start_thread.start()
+        assert not finished.wait(0.05)
+        writer.rollback()
+        checkpoint_thread.join(3)
+        start_thread.join(3)
+        assert finished.is_set()
+        assert results[0]["status"] == "succeeded"
+        assert maintenance.release(database, process_reaped=True)["status"] == "succeeded"
+
+
+def test_checkpoint_rejects_database_and_sidecar_symlinks(tmp_path: Path) -> None:
+    """维护操作拒绝数据库及 sidecar 符号链接，保持目标文件内容不变。"""
+    target = tmp_path / "target"
+    target.write_text("protected")
+    database = tmp_path / "opencode.db"
+    for path in (database, Path(str(database) + "-wal"), Path(str(database) + "-shm")):
+        path.symlink_to(target)
+        assert checkpoint_database(database)["code"] == "wal_checkpoint_path_invalid"
+        assert target.read_text() == "protected"
+        path.unlink()