Merge pull request #1803 from MemPalace/codex/1800-mine-lock-cleanup
fix(palace): clean source mine locks safely
This commit is contained in:
Igor Lins e Silva
GitHub
commit
7f5bbd8be8
|
|
@ -725,38 +725,185 @@ def mine_lock(source_file: str):
|
|||
Prevents multiple agents from mining the same file simultaneously,
|
||||
which causes duplicate drawers when the delete+insert cycle interleaves.
|
||||
"""
|
||||
lock_dir = os.path.join(os.path.expanduser("~"), ".mempalace", "locks")
|
||||
os.makedirs(lock_dir, exist_ok=True)
|
||||
lock_path = os.path.join(
|
||||
lock_dir, hashlib.sha256(source_file.encode()).hexdigest()[:16] + ".lock"
|
||||
)
|
||||
|
||||
lf = open(lock_path, "w")
|
||||
lock_path = _mine_lock_path(source_file)
|
||||
lf = _acquire_mine_lock_file(lock_path)
|
||||
try:
|
||||
if os.name == "nt":
|
||||
import msvcrt
|
||||
|
||||
msvcrt.locking(lf.fileno(), msvcrt.LK_LOCK, 1)
|
||||
else:
|
||||
import fcntl
|
||||
|
||||
fcntl.flock(lf, fcntl.LOCK_EX)
|
||||
yield
|
||||
finally:
|
||||
try:
|
||||
if os.name == "nt":
|
||||
import msvcrt
|
||||
|
||||
msvcrt.locking(lf.fileno(), msvcrt.LK_UNLCK, 1)
|
||||
else:
|
||||
import fcntl
|
||||
|
||||
fcntl.flock(lf, fcntl.LOCK_UN)
|
||||
_unlock_mine_lock_file(lf)
|
||||
except Exception:
|
||||
logger.debug("Mine-lock release failed", exc_info=True)
|
||||
try:
|
||||
lf.close()
|
||||
except Exception:
|
||||
logger.debug("Mine-lock close failed", exc_info=True)
|
||||
_cleanup_mine_lock_file(lock_path)
|
||||
|
||||
|
||||
def _mine_lock_path(source_file: str) -> str:
|
||||
lock_dir = os.path.join(os.path.expanduser("~"), ".mempalace", "locks")
|
||||
os.makedirs(lock_dir, exist_ok=True)
|
||||
return os.path.join(lock_dir, hashlib.sha256(source_file.encode()).hexdigest()[:16] + ".lock")
|
||||
|
||||
|
||||
def _open_mine_lock_file(lock_path: str, *, create: bool):
|
||||
flags = os.O_RDWR
|
||||
if create:
|
||||
flags |= os.O_CREAT
|
||||
fd = os.open(lock_path, flags, 0o600)
|
||||
return os.fdopen(fd, "r+b")
|
||||
|
||||
|
||||
def _lock_mine_lock_file(lock_file, *, blocking: bool) -> bool:
|
||||
lock_file.seek(0)
|
||||
if os.name == "nt":
|
||||
import msvcrt
|
||||
|
||||
mode = msvcrt.LK_LOCK if blocking else msvcrt.LK_NBLCK
|
||||
try:
|
||||
msvcrt.locking(lock_file.fileno(), mode, 1)
|
||||
except OSError:
|
||||
if not blocking:
|
||||
return False
|
||||
raise
|
||||
return True
|
||||
|
||||
import fcntl
|
||||
|
||||
flags = fcntl.LOCK_EX
|
||||
if not blocking:
|
||||
flags |= fcntl.LOCK_NB
|
||||
try:
|
||||
fcntl.flock(lock_file, flags)
|
||||
except BlockingIOError:
|
||||
if not blocking:
|
||||
return False
|
||||
raise
|
||||
return True
|
||||
|
||||
|
||||
def _unlock_mine_lock_file(lock_file) -> None:
|
||||
lock_file.seek(0)
|
||||
if os.name == "nt":
|
||||
import msvcrt
|
||||
|
||||
msvcrt.locking(lock_file.fileno(), msvcrt.LK_UNLCK, 1)
|
||||
return
|
||||
|
||||
import fcntl
|
||||
|
||||
fcntl.flock(lock_file, fcntl.LOCK_UN)
|
||||
|
||||
|
||||
def _mine_lock_file_is_current(lock_file, lock_path: str) -> bool:
|
||||
"""Return whether ``lock_file`` is still the inode reached by ``lock_path``.
|
||||
|
||||
POSIX advisory locks attach to the opened inode, not the pathname. If a
|
||||
lock file is unlinked while a contender is waiting, that contender can later
|
||||
acquire a lock on an inode no new process will use. We reject that stale
|
||||
handle and retry on the current pathname.
|
||||
"""
|
||||
if os.name == "nt":
|
||||
return True
|
||||
try:
|
||||
path_stat = os.stat(lock_path)
|
||||
file_stat = os.fstat(lock_file.fileno())
|
||||
except OSError:
|
||||
return False
|
||||
return (path_stat.st_dev, path_stat.st_ino) == (file_stat.st_dev, file_stat.st_ino)
|
||||
|
||||
|
||||
def _acquire_open_mine_lock_file(lock_file, lock_path: str) -> bool:
|
||||
"""Acquire ``lock_file`` and return False if cleanup made it stale."""
|
||||
_lock_mine_lock_file(lock_file, blocking=True)
|
||||
if _mine_lock_file_is_current(lock_file, lock_path):
|
||||
return True
|
||||
try:
|
||||
_unlock_mine_lock_file(lock_file)
|
||||
except Exception:
|
||||
logger.debug("Mine-lock stale-handle release failed", exc_info=True)
|
||||
return False
|
||||
|
||||
|
||||
def _acquire_mine_lock_file(lock_path: str):
|
||||
while True:
|
||||
lf = _open_mine_lock_file(lock_path, create=True)
|
||||
try:
|
||||
if _acquire_open_mine_lock_file(lf, lock_path):
|
||||
return lf
|
||||
except Exception:
|
||||
lf.close()
|
||||
raise
|
||||
lf.close()
|
||||
|
||||
|
||||
def _cleanup_mine_lock_file(lock_path: str) -> None:
|
||||
"""Best-effort removal that preserves flock rendezvous semantics.
|
||||
|
||||
A plain ``os.remove(lock_path)`` after closing the critical-section lock is
|
||||
unsafe on POSIX: a waiter may already be blocked on the old inode while a
|
||||
later process creates and locks a new inode at the same pathname. Instead,
|
||||
cleanup briefly re-acquires the current file nonblocking. If it wins, it can
|
||||
unlink that inode as cleanup-only work; waiters on the old inode will detect
|
||||
the stale handle after waking and retry on the current path.
|
||||
"""
|
||||
try:
|
||||
lf = _open_mine_lock_file(lock_path, create=False)
|
||||
except FileNotFoundError:
|
||||
return
|
||||
except OSError:
|
||||
logger.debug("Mine-lock cleanup open failed for %s", lock_path, exc_info=True)
|
||||
return
|
||||
|
||||
acquired = False
|
||||
closed = False
|
||||
try:
|
||||
try:
|
||||
acquired = _lock_mine_lock_file(lf, blocking=False)
|
||||
except OSError:
|
||||
logger.debug("Mine-lock cleanup acquire failed for %s", lock_path, exc_info=True)
|
||||
return
|
||||
if not acquired:
|
||||
return
|
||||
if not _mine_lock_file_is_current(lf, lock_path):
|
||||
return
|
||||
|
||||
if os.name == "nt":
|
||||
# Windows generally cannot unlink an open locked file. Release and
|
||||
# close first; if another process opens the file in the gap,
|
||||
# os.remove should fail and we leave the rendezvous file in place.
|
||||
try:
|
||||
_unlock_mine_lock_file(lf)
|
||||
except Exception:
|
||||
logger.debug("Mine-lock cleanup release failed", exc_info=True)
|
||||
acquired = False
|
||||
return
|
||||
acquired = False
|
||||
lf.close()
|
||||
closed = True
|
||||
try:
|
||||
os.remove(lock_path)
|
||||
except OSError:
|
||||
pass
|
||||
return
|
||||
|
||||
try:
|
||||
os.remove(lock_path)
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
except OSError:
|
||||
logger.debug("Mine-lock cleanup remove failed for %s", lock_path, exc_info=True)
|
||||
finally:
|
||||
if not closed:
|
||||
if acquired:
|
||||
try:
|
||||
_unlock_mine_lock_file(lf)
|
||||
except Exception:
|
||||
logger.debug("Mine-lock cleanup release failed", exc_info=True)
|
||||
lf.close()
|
||||
|
||||
|
||||
class MineAlreadyRunning(RuntimeError):
|
||||
"""Raised when another `mempalace mine` already holds the per-palace lock."""
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,225 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import multiprocessing
|
||||
import os
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
import mempalace.palace as palace_module
|
||||
from mempalace.palace import (
|
||||
_lock_mine_lock_file,
|
||||
_mine_lock_path,
|
||||
_open_mine_lock_file,
|
||||
_unlock_mine_lock_file,
|
||||
mine_lock,
|
||||
)
|
||||
|
||||
|
||||
def _set_home(monkeypatch, tmp_path: Path) -> None:
|
||||
monkeypatch.setenv("HOME", str(tmp_path))
|
||||
monkeypatch.setenv("USERPROFILE", str(tmp_path))
|
||||
|
||||
|
||||
def _wait_for_path(path: Path, timeout: float = 10.0) -> bool:
|
||||
deadline = time.monotonic() + timeout
|
||||
while time.monotonic() < deadline:
|
||||
if path.exists():
|
||||
return True
|
||||
time.sleep(0.01)
|
||||
return path.exists()
|
||||
|
||||
|
||||
def _assert_path_absent_for(path: Path, duration: float = 0.5) -> None:
|
||||
deadline = time.monotonic() + duration
|
||||
while time.monotonic() < deadline:
|
||||
assert not path.exists(), "waiter entered while replacement lock was held"
|
||||
time.sleep(0.01)
|
||||
assert not path.exists(), "waiter entered while replacement lock was held"
|
||||
|
||||
|
||||
def _stale_waiter_target(
|
||||
lock_path: str,
|
||||
source_file: str,
|
||||
opened_flag: str,
|
||||
entered_flag: str,
|
||||
release_flag: str,
|
||||
result_q,
|
||||
) -> None:
|
||||
try:
|
||||
from mempalace.palace import (
|
||||
_acquire_open_mine_lock_file as acquire_open,
|
||||
_open_mine_lock_file as open_lock,
|
||||
_unlock_mine_lock_file as unlock_file,
|
||||
mine_lock as public_mine_lock,
|
||||
)
|
||||
|
||||
lf = open_lock(lock_path, create=True)
|
||||
Path(opened_flag).touch()
|
||||
current = acquire_open(lf, lock_path)
|
||||
result_q.put(("first-acquire-current", current))
|
||||
if current:
|
||||
Path(entered_flag).touch()
|
||||
_wait_for_path(Path(release_flag))
|
||||
unlock_file(lf)
|
||||
lf.close()
|
||||
result_q.put(("done", True))
|
||||
return
|
||||
|
||||
lf.close()
|
||||
result_q.put(("retrying", True))
|
||||
with public_mine_lock(source_file):
|
||||
Path(entered_flag).touch()
|
||||
_wait_for_path(Path(release_flag))
|
||||
result_q.put(("done", True))
|
||||
except BaseException as exc: # pragma: no cover - surfaced through queue
|
||||
result_q.put(("error", repr(exc)))
|
||||
|
||||
|
||||
def test_mine_lock_removes_uncontended_lock_file(tmp_path, monkeypatch):
|
||||
_set_home(monkeypatch, tmp_path)
|
||||
source_file = str(tmp_path / "source.txt")
|
||||
lock_path = Path(_mine_lock_path(source_file))
|
||||
|
||||
with mine_lock(source_file):
|
||||
assert lock_path.exists()
|
||||
|
||||
assert not lock_path.exists()
|
||||
|
||||
with mine_lock(source_file):
|
||||
assert lock_path.exists()
|
||||
|
||||
assert not lock_path.exists()
|
||||
|
||||
|
||||
def test_mine_lock_close_failure_still_runs_cleanup(monkeypatch):
|
||||
events = []
|
||||
|
||||
class FakeLock:
|
||||
def close(self):
|
||||
events.append("close")
|
||||
raise OSError("close failed")
|
||||
|
||||
fake_lock = FakeLock()
|
||||
monkeypatch.setattr(palace_module, "_mine_lock_path", lambda source_file: "source.lock")
|
||||
monkeypatch.setattr(palace_module, "_acquire_mine_lock_file", lambda lock_path: fake_lock)
|
||||
monkeypatch.setattr(
|
||||
palace_module, "_unlock_mine_lock_file", lambda lock_file: events.append("unlock")
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
palace_module,
|
||||
"_cleanup_mine_lock_file",
|
||||
lambda lock_path: events.append(("cleanup", lock_path)),
|
||||
)
|
||||
|
||||
with palace_module.mine_lock("source.txt"):
|
||||
events.append("body")
|
||||
|
||||
assert events == ["body", "unlock", "close", ("cleanup", "source.lock")]
|
||||
|
||||
|
||||
def test_windows_cleanup_release_failure_does_not_retry_unlock(monkeypatch):
|
||||
events = []
|
||||
|
||||
class FakeLock:
|
||||
def close(self):
|
||||
events.append("close")
|
||||
|
||||
fake_lock = FakeLock()
|
||||
|
||||
monkeypatch.setattr(palace_module.os, "name", "nt", raising=False)
|
||||
monkeypatch.setattr(
|
||||
palace_module, "_open_mine_lock_file", lambda lock_path, *, create: fake_lock
|
||||
)
|
||||
monkeypatch.setattr(palace_module, "_lock_mine_lock_file", lambda lock_file, *, blocking: True)
|
||||
monkeypatch.setattr(
|
||||
palace_module, "_mine_lock_file_is_current", lambda lock_file, lock_path: True
|
||||
)
|
||||
|
||||
def fail_unlock(lock_file):
|
||||
events.append("unlock")
|
||||
raise OSError("unlock failed")
|
||||
|
||||
monkeypatch.setattr(palace_module, "_unlock_mine_lock_file", fail_unlock)
|
||||
|
||||
palace_module._cleanup_mine_lock_file("source.lock")
|
||||
|
||||
assert events == ["unlock", "close"]
|
||||
|
||||
|
||||
@pytest.mark.skipif(os.name == "nt", reason="POSIX inode replacement regression")
|
||||
def test_mine_lock_retries_when_waiter_wakes_on_unlinked_inode(tmp_path, monkeypatch):
|
||||
"""A waiter on an unlinked lock inode must not enter the critical section.
|
||||
|
||||
This models the race from issue #1800: process A removes the path after
|
||||
release while process B was already waiting on the old inode and process C
|
||||
has locked a replacement path. B must reject the stale inode and retry.
|
||||
"""
|
||||
_set_home(monkeypatch, tmp_path)
|
||||
source_file = str(tmp_path / "source.txt")
|
||||
lock_path = Path(_mine_lock_path(source_file))
|
||||
|
||||
old_lf = _open_mine_lock_file(str(lock_path), create=True)
|
||||
replacement_lf = None
|
||||
child = None
|
||||
try:
|
||||
assert _lock_mine_lock_file(old_lf, blocking=False)
|
||||
|
||||
opened_flag = tmp_path / "opened"
|
||||
entered_flag = tmp_path / "entered"
|
||||
release_flag = tmp_path / "release"
|
||||
ctx = multiprocessing.get_context("spawn")
|
||||
result_q = ctx.Queue()
|
||||
child = ctx.Process(
|
||||
target=_stale_waiter_target,
|
||||
args=(
|
||||
str(lock_path),
|
||||
source_file,
|
||||
str(opened_flag),
|
||||
str(entered_flag),
|
||||
str(release_flag),
|
||||
result_q,
|
||||
),
|
||||
)
|
||||
child.start()
|
||||
assert _wait_for_path(opened_flag), "waiter did not open the original lock file"
|
||||
|
||||
os.remove(lock_path)
|
||||
replacement_lf = _open_mine_lock_file(str(lock_path), create=True)
|
||||
assert _lock_mine_lock_file(replacement_lf, blocking=False)
|
||||
|
||||
_unlock_mine_lock_file(old_lf)
|
||||
old_lf.close()
|
||||
old_lf = None
|
||||
|
||||
assert result_q.get(timeout=10) == ("first-acquire-current", False)
|
||||
assert result_q.get(timeout=10) == ("retrying", True)
|
||||
_assert_path_absent_for(entered_flag)
|
||||
|
||||
_unlock_mine_lock_file(replacement_lf)
|
||||
replacement_lf.close()
|
||||
replacement_lf = None
|
||||
|
||||
assert _wait_for_path(entered_flag), "waiter did not retry on the replacement path"
|
||||
release_flag.touch()
|
||||
assert result_q.get(timeout=10) == ("done", True)
|
||||
child.join(timeout=10)
|
||||
assert child.exitcode == 0
|
||||
assert not lock_path.exists()
|
||||
finally:
|
||||
if child is not None and child.is_alive():
|
||||
child.terminate()
|
||||
child.join(timeout=5)
|
||||
if replacement_lf is not None:
|
||||
try:
|
||||
_unlock_mine_lock_file(replacement_lf)
|
||||
except Exception:
|
||||
pass
|
||||
replacement_lf.close()
|
||||
if old_lf is not None:
|
||||
try:
|
||||
_unlock_mine_lock_file(old_lf)
|
||||
except Exception:
|
||||
pass
|
||||
old_lf.close()
|
||||
Loading…
Reference in New Issue