fix(kanban): add Windows init lock guard

This commit is contained in:
Squiddy 2026-05-26 15:21:45 -04:00 committed by kshitij
parent 90b6b3d18f
commit 3ba8962738
2 changed files with 46 additions and 2 deletions

View file

@ -1033,14 +1033,35 @@ def _cross_process_init_lock(path: Path):
lock_path = path.with_name(path.name + ".init.lock")
handle = lock_path.open("a+b")
try:
if not _IS_WINDOWS:
if _IS_WINDOWS:
import msvcrt
# Lock a single byte in the sidecar file. ``msvcrt.locking`` starts
# at the current file position, so seek explicitly before both
# lock and unlock. The file is opened in append/read binary mode so
# it always exists but the byte-range lock is the synchronization
# primitive; no payload needs to be written.
handle.seek(0)
locking = getattr(msvcrt, "locking")
lock_mode = getattr(msvcrt, "LK_LOCK")
locking(handle.fileno(), lock_mode, 1)
else:
import fcntl
fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
yield
finally:
try:
if not _IS_WINDOWS:
if _IS_WINDOWS:
import msvcrt
handle.seek(0)
locking = getattr(msvcrt, "locking")
unlock_mode = getattr(msvcrt, "LK_UNLCK")
locking(handle.fileno(), unlock_mode, 1)
else:
import fcntl
fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
finally:
handle.close()

View file

@ -5,7 +5,9 @@ from __future__ import annotations
import concurrent.futures
import os
import sqlite3
import sys
import time
import types
import unittest.mock
from pathlib import Path
@ -65,6 +67,27 @@ def test_connect_honors_kanban_busy_timeout_env(kanban_home, monkeypatch):
assert row[0] == 123456
def test_cross_process_init_lock_uses_windows_byte_range_lock(tmp_path, monkeypatch):
"""Windows must use a real process lock, not a no-op sidecar open."""
calls: list[tuple[int, int, int]] = []
fake_msvcrt = types.SimpleNamespace(
LK_LOCK=1,
LK_UNLCK=2,
locking=lambda fd, mode, nbytes: calls.append((fd, mode, nbytes)),
)
monkeypatch.setattr(kb, "_IS_WINDOWS", True)
monkeypatch.setitem(sys.modules, "msvcrt", fake_msvcrt)
db_path = tmp_path / "kanban.db"
with kb._cross_process_init_lock(db_path):
assert calls == [(calls[0][0], fake_msvcrt.LK_LOCK, 1)]
assert [call[1:] for call in calls] == [
(fake_msvcrt.LK_LOCK, 1),
(fake_msvcrt.LK_UNLCK, 1),
]
def test_connect_rejects_tls_record_in_sqlite_header(tmp_path, monkeypatch):
"""Kanban should classify TLS-looking page-0 clobbers before WAL setup."""
home = tmp_path / ".hermes"