From 41c063bee8c70d628b8dab2d5bf3b1876f821a3d Mon Sep 17 00:00:00 2001 From: WenhuaXia Date: Tue, 2 Jun 2026 17:03:05 +0800 Subject: [PATCH 1/2] =?UTF-8?q?fix(kanban):=20=E7=9C=8B=E6=9D=BF=E8=B7=A8?= =?UTF-8?q?=E8=BF=9B=E7=A8=8B=E5=86=99=E9=94=81=E9=98=B2=E6=AD=A2=20SQLite?= =?UTF-8?q?=20WAL=20=E6=8D=9F=E5=9D=8F=20/=20add=20inter-process=20flock?= =?UTF-8?q?=20to=20prevent=20DB=20corruption?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 问题 / Problem: 多个 kanban worker 进程并发写入同一 SQLite 数据库导致频繁 WAL 损坏（33 分钟内 7 次）。SQLite WAL + busy_timeout 只能保护同一进程内的 并发线程，跨进程仍会竞争 BEGIN IMMEDIATE 导致页面损坏。 Problem: Multiple kanban worker processes writing concurrently to the same SQLite database caused frequent WAL corruption (7 times in 33 minutes). SQLite WAL + busy_timeout only protects concurrent threads within the same process; cross-process races through BEGIN IMMEDIATE can corrupt WAL pages. 修复 / Fix: 1. connect() 后打开 .db.lock 文件，write_txn() 用 fcntl.flock(LOCK_EX) 包裹整个事务，确保同一时间只有一个进程在写。 2. schema init 也获取 .db.lock（不只是 .init.lock），防止 DDL/DML 竞争。 3. Windows 或 fcntl 不可用时优雅降级为 no-op。 1. After connect(), open a .db.lock file. Every write_txn() acquires fcntl.flock(LOCK_EX) around the entire transaction, ensuring only one process writes at a time. 2. Schema init also acquires .db.lock (not just .init.lock) to prevent DDL/DML races. 3. Graceful fallback on Windows or when fcntl is unavailable. Changes: - hermes_cli/kanban_db.py: fcntl flock for inter-process write safety Related: #35787 (closed), #37292 (closed, split) --- hermes_cli/kanban_db.py | 59 ++++++++++++++++++++++++++++++++++++++--- 1 file changed, 56 insertions(+), 3 deletions(-) diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index c8c53dba7ecb6..938b593c89ab3 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -86,7 +86,7 @@ from contextvars import ContextVar, Token from dataclasses import dataclass, field from pathlib import Path -from typing import Any, Iterable, Optional +from typing import Any, Iterable, Optional, TextIO from toolsets import get_toolset_names @@ -103,6 +103,19 @@ KNOWN_TOOLSET_NAMES = frozenset(name.casefold() for name in get_toolset_names()) _IS_WINDOWS = sys.platform == "win32" +# Optional fcntl import — Windows does not ship fcntl. +# On non-Windows platforms, used for per-DB inter-process write locking. +# On Windows, _KANBAN_WRITE_LOCKS entries are None and flock is a no-op. +try: + import fcntl # type: ignore[import-not-found,import-untyped] +except ImportError: + fcntl = None # type: ignore[assignment] # Windows / other platforms without fcntl + +# Maps id(conn) → open file handle for per-DB write locking. +# sqlite3.Connection is a C-extension type: no attributes, no weakref. +# We use id(conn) as key and accept the dict grows lazily. +_KANBAN_WRITE_LOCKS: dict[int, "TextIO | None"] = {} + # A running task's claim is valid for 15 minutes by default; after that the # next dispatcher tick reclaims it. Workers that outlive this window should # call ``heartbeat_claim(task_id)`` periodically. In practice most kanban @@ -1450,6 +1463,19 @@ def connect( _guard_existing_db_is_healthy(path) resolved = str(path.resolve()) conn = _sqlite_connect(path) + # Per-DB file-level write lock — prevents multiple worker processes from + # racing through BEGIN IMMEDIATE transactions simultaneously, which can + # corrupt the WAL under heavy concurrent write pressure (task_events + + # task_runs + status transitions all within the same tick). Windows + # (no fcntl) and network filesystems gracefully fall back to no-op. + if fcntl is not None: + lock_path = path.with_suffix('.db.lock') + try: + _KANBAN_WRITE_LOCKS[id(conn)] = open(lock_path, 'w') + except OSError: + _KANBAN_WRITE_LOCKS[id(conn)] = None # best-effort: don't fail + else: + _KANBAN_WRITE_LOCKS[id(conn)] = None # Windows / no-fcntl: skip file lock entirely try: conn.row_factory = sqlite3.Row with _INIT_LOCK: @@ -1480,8 +1506,22 @@ def connect( # process are cheap. The lock prevents same-process dispatcher # threads from racing through the additive ALTER TABLE pass with # stale PRAGMA snapshots during gateway startup. - conn.executescript(SCHEMA_SQL) - _migrate_add_optional_columns(conn) + # + # CRITICAL: DDL writes (executescript + migrations) must be under + # the per-DB .db.lock to prevent racing with other processes' + # write_txn() calls. Without this, a new worker's schema init can + # collide with a dispatching gateway's DML writes, corrupting + # the WAL. .init.lock alone is NOT sufficient because it is a + # different file from .db.lock. + lock_handle = _KANBAN_WRITE_LOCKS.get(id(conn)) + if lock_handle is not None: + fcntl.flock(lock_handle.fileno(), fcntl.LOCK_EX) + try: + conn.executescript(SCHEMA_SQL) + _migrate_add_optional_columns(conn) + finally: + if lock_handle is not None: + fcntl.flock(lock_handle.fileno(), fcntl.LOCK_UN) _INITIALIZED_PATHS.add(resolved) except Exception: conn.close() @@ -1986,10 +2026,20 @@ def write_txn(conn: sqlite3.Connection): task + recording an event, etc.). A claim CAS inside this context is atomic -- at most one concurrent writer can succeed. + Additionally, if conn's corresponding lock fd is set (by connect()), + an inter-process exclusive file lock is acquired around the entire + transaction. This prevents multiple worker processes from racing + through BEGIN IMMEDIATE simultaneously, which can corrupt the WAL + under heavy concurrent write pressure. On platforms without fcntl + or when the lock fd is unavailable, this is a graceful no-op. + The explicit ROLLBACK on exception is wrapped in try/except so that a SQLite auto-rollback (which leaves no active transaction) does not shadow the original exception with a spurious rollback error. """ + lock_handle = _KANBAN_WRITE_LOCKS.get(id(conn)) + if lock_handle is not None: + fcntl.flock(lock_handle.fileno(), fcntl.LOCK_EX) conn.execute("BEGIN IMMEDIATE") try: yield conn @@ -2007,6 +2057,9 @@ def write_txn(conn: sqlite3.Connection): # Post-commit file-length check: header page_count must match actual file pages. # A discrepancy means a torn-extend — raise now rather than silently corrupt. _check_file_length_invariant(conn) + finally: + if lock_handle is not None: + fcntl.flock(lock_handle.fileno(), fcntl.LOCK_UN) # --------------------------------------------------------------------------- From 822c1d7055dfafdd4b46c9c89e2a92f7b9199224 Mon Sep 17 00:00:00 2001 From: WenhuaXia Date: Tue, 2 Jun 2026 20:50:55 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix(kanban):=20address=20PR=20review=20feed?= =?UTF-8?q?back=20=E2=80=94=20FD=20leak=20and=20id(conn)=20key=20reuse?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes from @liuhao1024 review on #37344: 1. FD leak: _KANBAN_WRITE_LOCKS entries were never cleaned up. Added _kanban_release_write_lock() called from connect() exception handler and connect_closing() finally block. 2. id(conn) as dict key: if a connection is GC'd and a new sqlite3.Connection reuses the same memory address, it would get the stale lock handle. Replaced id(conn) key with resolved DB path (str) for _KANBAN_WRITE_LOCKS. Added _KANBAN_CONN_PATHS reverse mapping so write_txn() can still look up the per-DB lock from a sqlite3.Connection alone. Thanks to @liuhao1024 for the thorough review. --- hermes_cli/kanban_db.py | 52 ++++++++++++++++++++++++++++++++++------- 1 file changed, 43 insertions(+), 9 deletions(-) diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index 938b593c89ab3..de17ad42ee351 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -111,10 +111,34 @@ except ImportError: fcntl = None # type: ignore[assignment] # Windows / other platforms without fcntl -# Maps id(conn) → open file handle for per-DB write locking. -# sqlite3.Connection is a C-extension type: no attributes, no weakref. -# We use id(conn) as key and accept the dict grows lazily. -_KANBAN_WRITE_LOCKS: dict[int, "TextIO | None"] = {} +# Maps resolved DB path → open file handle for per-DB write locking. +# Using the resolved path (str) as key instead of id(conn) avoids stale +# handle reuse when a connection is GC'd and a new one reuses the same +# memory address in long-lived processes (#37344 review feedback). +_KANBAN_WRITE_LOCKS: dict[str, "TextIO | None"] = {} + +# Reverse mapping: id(conn) → resolved DB path, so write_txn() can look up +# the per-DB lock handle from a sqlite3.Connection without needing the path +# passed in explicitly. Entries are cleaned up alongside _KANBAN_WRITE_LOCKS. +_KANBAN_CONN_PATHS: dict[int, str] = {} + + +def _kanban_release_write_lock(db_path: str, conn: Optional[sqlite3.Connection] = None) -> None: + """Close the .db.lock handle and remove the entry for a given DB path. + + Also removes the reverse id(conn)→path mapping if conn is provided. + Called from connect()'s exception handler, connect_closing()'s finally + block, and any code path that closes a kanban connection. Prevents FD + leaks (#37344 review feedback — re-introducing #33159 class of bug). + """ + lock_handle = _KANBAN_WRITE_LOCKS.pop(db_path, None) + if lock_handle is not None: + try: + lock_handle.close() + except OSError: + pass + if conn is not None: + _KANBAN_CONN_PATHS.pop(id(conn), None) # A running task's claim is valid for 15 minutes by default; after that the # next dispatcher tick reclaims it. Workers that outlive this window should @@ -1471,11 +1495,14 @@ def connect( if fcntl is not None: lock_path = path.with_suffix('.db.lock') try: - _KANBAN_WRITE_LOCKS[id(conn)] = open(lock_path, 'w') + _KANBAN_WRITE_LOCKS[resolved] = open(lock_path, 'w') except OSError: - _KANBAN_WRITE_LOCKS[id(conn)] = None # best-effort: don't fail + _KANBAN_WRITE_LOCKS[resolved] = None # best-effort: don't fail else: - _KANBAN_WRITE_LOCKS[id(conn)] = None # Windows / no-fcntl: skip file lock entirely + _KANBAN_WRITE_LOCKS[resolved] = None # Windows / no-fcntl: skip file lock entirely + # Reverse mapping so write_txn() can resolve the per-DB lock handle + # from the sqlite3.Connection alone (no path arg needed). + _KANBAN_CONN_PATHS[id(conn)] = resolved try: conn.row_factory = sqlite3.Row with _INIT_LOCK: @@ -1513,7 +1540,7 @@ def connect( # collide with a dispatching gateway's DML writes, corrupting # the WAL. .init.lock alone is NOT sufficient because it is a # different file from .db.lock. - lock_handle = _KANBAN_WRITE_LOCKS.get(id(conn)) + lock_handle = _KANBAN_WRITE_LOCKS.get(resolved) if lock_handle is not None: fcntl.flock(lock_handle.fileno(), fcntl.LOCK_EX) try: @@ -1524,6 +1551,7 @@ def connect( fcntl.flock(lock_handle.fileno(), fcntl.LOCK_UN) _INITIALIZED_PATHS.add(resolved) except Exception: + _kanban_release_write_lock(resolved) conn.close() raise return conn @@ -1558,6 +1586,11 @@ def connect_closing( try: yield conn finally: + # Release the per-DB write lock handle before closing the connection + # to prevent FD leaks (#37344 review feedback). + resolved_path = _KANBAN_CONN_PATHS.get(id(conn)) + if resolved_path is not None: + _kanban_release_write_lock(resolved_path, conn) try: conn.close() except Exception: @@ -2037,7 +2070,8 @@ def write_txn(conn: sqlite3.Connection): a SQLite auto-rollback (which leaves no active transaction) does not shadow the original exception with a spurious rollback error. """ - lock_handle = _KANBAN_WRITE_LOCKS.get(id(conn)) + db_path = _KANBAN_CONN_PATHS.get(id(conn)) + lock_handle = _KANBAN_WRITE_LOCKS.get(db_path) if db_path else None if lock_handle is not None: fcntl.flock(lock_handle.fileno(), fcntl.LOCK_EX) conn.execute("BEGIN IMMEDIATE")