cc5cf768fd
`_windows_restart_manager_holders` rebuilt two ctypes Structure classes and
called `ctypes.POINTER()` on the fresh class every scan. POINTER memoises in
the process-global `ctypes._pointer_type_cache`, so each call pinned another
class set forever: probe measured 3 -> 221 cache entries over 200 calls. The
scan runs from auto-VACUUM admission and the FTS-rebuild guard in a long-lived
gateway, so the growth was monotonic. Structures, constants and argtypes now
live at module level (ctypes.wintypes imports on every platform); the
`ctypes.WinDLL("rstrtmgr")` load stays per call so the unit test can still
inject a fake rstrtmgr. Same probe after: 22 -> 22.
While here: one `_sqlite_family(base)` helper replaces the fourth hand-rolled
(db, -wal, -shm) triple in this module and the three existing ones; the
Windows lane computes abspath once instead of per holder; the RmGetList
sizing call and the retry loop collapse into one bounded loop (pass 1 sizes
with a NULL buffer, later passes re-size), which keeps the existing fake's
`apps is None` contract.
649 lines
27 KiB
Python
649 lines
27 KiB
Python
"""Process and descriptor authority for state.db structural maintenance.
|
|
|
|
This module owns the proof that no foreign process still holds the active or
|
|
an unlinked SQLite DB/WAL/SHM generation. ``hermes_state`` supplies only the
|
|
SQLite connection factory needed by the final lock probe.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ctypes
|
|
import errno
|
|
import logging
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
from ctypes import wintypes
|
|
from pathlib import Path
|
|
from typing import Callable, List, Optional, Sequence, Set, Tuple
|
|
|
|
from hermes_state_errors import is_sqlite_lock_error
|
|
|
|
try: # Hard dependency, but tolerate scaffold-phase imports before pip install.
|
|
import psutil
|
|
except ImportError: # pragma: no cover - stripped/scaffold installs only
|
|
psutil = None # type: ignore[assignment]
|
|
|
|
|
|
def read_only_db_uri(db_path) -> str:
|
|
"""``file:`` URI for a ``mode=ro`` open. ``as_uri()`` percent-encodes ``?``/``#`` in the home
|
|
path; a raw ``f"file:{path}?mode=ro"`` truncates there and opens the wrong (empty) database."""
|
|
return Path(db_path).resolve().as_uri() + "?mode=ro"
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_IS_WINDOWS = sys.platform == "win32"
|
|
_HERMES_EXECUTABLES = frozenset({"hermes", "hermes-agent", "hermes-acp"})
|
|
_HERMES_PYTHON_MODULES = frozenset({"acp_adapter", "hermes_cli.main"})
|
|
_HERMES_PYTHON_SCRIPTS = frozenset({"hermes_cli/main.py", "run_agent.py"})
|
|
_PYTHON_SHORT_OPTIONS_WITH_OPERANDS = frozenset({"Q", "W", "X"})
|
|
_PYTHON_LONG_OPTIONS_WITH_OPERANDS = frozenset(
|
|
{"--check-hash-based-pycs", "--jit"}
|
|
)
|
|
|
|
|
|
def _read_proc_argv(pid: int) -> Optional[List[str]]:
|
|
"""Read /proc/<pid>/cmdline without losing argv boundaries."""
|
|
try:
|
|
with open(f"/proc/{pid}/cmdline", "rb") as handle:
|
|
raw = handle.read()
|
|
if not raw:
|
|
return None
|
|
argv = raw.decode("utf-8", "replace").split("\x00")
|
|
if argv[-1] == "":
|
|
argv.pop()
|
|
return argv or None
|
|
except OSError:
|
|
return None
|
|
|
|
|
|
def describe_holder_pid(pid: int) -> str:
|
|
"""``PID 123 (hermes gateway run)`` for operator-facing holder lists; /proc argv first, psutil elsewhere."""
|
|
argv = _read_proc_argv(pid)
|
|
if argv is None and psutil is not None:
|
|
try:
|
|
argv = psutil.Process(pid).cmdline() or None
|
|
except Exception:
|
|
argv = None
|
|
who = " ".join(" ".join([os.path.basename(argv[0]), *argv[1:]]).split())[:80] if argv else "command line unavailable"
|
|
return f"PID {pid} ({who})"
|
|
|
|
|
|
def _looks_like_python_executable(program: str) -> bool:
|
|
name = os.path.basename(program).lower().removesuffix(".exe")
|
|
for prefix in ("python", "pypy"):
|
|
if name.startswith(prefix):
|
|
suffix = name[len(prefix) :]
|
|
return not suffix or all(char.isdigit() or char == "." for char in suffix)
|
|
return False
|
|
|
|
|
|
def _python_execution_target(argv: Sequence[str]) -> Optional[Tuple[str, str]]:
|
|
"""Return the Python module or script selected by interpreter options."""
|
|
index = 1
|
|
while index < len(argv):
|
|
arg = argv[index]
|
|
if arg == "--":
|
|
index += 1
|
|
return ("script", argv[index]) if index < len(argv) else None
|
|
if arg in _PYTHON_LONG_OPTIONS_WITH_OPERANDS:
|
|
index += 2
|
|
continue
|
|
if arg.startswith("--check-hash-based-pycs=") or arg.startswith("--jit="):
|
|
index += 1
|
|
continue
|
|
if arg.startswith("--"):
|
|
index += 1
|
|
continue
|
|
if arg.startswith("-") and arg != "-":
|
|
options = arg[1:]
|
|
option_index = 0
|
|
consumed_next = False
|
|
while option_index < len(options):
|
|
option = options[option_index]
|
|
attached = options[option_index + 1 :]
|
|
if option == "c":
|
|
return None
|
|
if option == "m":
|
|
if attached:
|
|
return "module", attached
|
|
index += 1
|
|
return ("module", argv[index]) if index < len(argv) else None
|
|
if option in _PYTHON_SHORT_OPTIONS_WITH_OPERANDS:
|
|
consumed_next = not attached
|
|
break
|
|
option_index += 1
|
|
index += 2 if consumed_next else 1
|
|
continue
|
|
return "script", arg
|
|
return None
|
|
|
|
|
|
def _looks_like_hermes(argv: Sequence[str]) -> bool:
|
|
"""Return whether argv identifies a supported Hermes execution target."""
|
|
if not argv:
|
|
return False
|
|
program = os.path.basename(argv[0]).lower().removesuffix(".exe")
|
|
if program in _HERMES_EXECUTABLES:
|
|
return True
|
|
if not _looks_like_python_executable(program):
|
|
return False
|
|
target = _python_execution_target(argv)
|
|
if target is None:
|
|
return False
|
|
kind, value = target
|
|
normalized = value.lower().replace("\\", "/")
|
|
if kind == "module":
|
|
return normalized in _HERMES_PYTHON_MODULES
|
|
return any(
|
|
normalized == script or normalized.endswith(f"/{script}")
|
|
for script in _HERMES_PYTHON_SCRIPTS
|
|
)
|
|
|
|
|
|
def canonical_sqlite_path(path: str) -> str:
|
|
"""Normalize a /proc fd target, stripping the Linux `` (deleted)`` suffix."""
|
|
return os.path.normcase(os.path.abspath(path.removesuffix(" (deleted)")))
|
|
|
|
|
|
_HOME_FLAGS = ("--hermes-home",)
|
|
_PROFILE_FLAGS = ("--profile", "-p")
|
|
_STATE_DB_NAMES = ("state.db", "state.db-wal", "state.db-shm")
|
|
|
|
|
|
def _norm_path(value: str) -> str:
|
|
return os.path.normcase(os.path.normpath(value))
|
|
|
|
|
|
def _argv_flag_value(argv: Sequence[str], flags: Sequence[str]) -> Optional[str]:
|
|
"""Last ``--flag X`` / ``--flag=X`` value, token-exact (``--profile timothy`` is not ``tim``)."""
|
|
value: Optional[str] = None
|
|
index, count = 0, len(argv)
|
|
while index < count:
|
|
token = argv[index]
|
|
if isinstance(token, str):
|
|
if token in flags and index + 1 < count and isinstance(argv[index + 1], str):
|
|
value = argv[index + 1]
|
|
index += 2
|
|
continue
|
|
for flag in flags:
|
|
if token.startswith(flag + "="):
|
|
value = token[len(flag) + 1:]
|
|
break
|
|
index += 1
|
|
return value
|
|
|
|
|
|
def _argv_env_home(argv: Sequence[str]) -> Optional[str]:
|
|
"""``HERMES_HOME=<path>`` env-style assignment on the argv (``env HERMES_HOME=… hermes …``)."""
|
|
for token in reversed(list(argv)):
|
|
if isinstance(token, str) and token.startswith("HERMES_HOME="):
|
|
return token[len("HERMES_HOME="):]
|
|
return None
|
|
|
|
|
|
def _store_install_layout(this_home: str) -> Tuple[Optional[str], Optional[str]]:
|
|
"""``(<install root>, <our profile name>)`` for the home holding the store, else ``(None, None)``.
|
|
|
|
Derived with the canonical ``named_profile_home`` predicate, never a ``basename == "profiles"``
|
|
string test: an arbitrary ``<X>/profiles/<n>/`` tree is not a Hermes install, and promoting
|
|
``<X>`` to "ours" swallows an unrelated instance living under it — the literal two-instance
|
|
shape of #92401. The root store (``~/.hermes/state.db``) is its own root with no profile name,
|
|
so ANY named-profile selection contradicts it.
|
|
"""
|
|
try:
|
|
from hermes_constants import named_profile_home
|
|
|
|
profile_home = named_profile_home(this_home)
|
|
if profile_home is not None:
|
|
return os.path.abspath(str(profile_home.parent.parent)), profile_home.name
|
|
if os.path.basename(this_home) == ".hermes":
|
|
return os.path.abspath(this_home), None
|
|
except Exception: # constants import/resolution must never break a holder scan
|
|
logger.debug("Could not classify the install layout of %s", this_home, exc_info=True)
|
|
return None, None
|
|
|
|
|
|
def _names_other_profile(normalized: str, install_root: Optional[str], our_profile: Optional[str]) -> bool:
|
|
"""True when the token is under ``<install root>/profiles/<name>`` for a name that is not ours."""
|
|
if install_root is None:
|
|
return False
|
|
prefix = _norm_path(os.path.join(install_root, "profiles")) + os.sep
|
|
if not normalized.startswith(prefix):
|
|
return False
|
|
name = normalized[len(prefix):].split(os.sep, 1)[0]
|
|
return bool(name) and name != (os.path.normcase(our_profile) if our_profile else None)
|
|
|
|
|
|
def _argv_home_selection(
|
|
argv: Sequence[str], this_home: str, install_root: Optional[str], our_profile: Optional[str]
|
|
) -> Optional[str]:
|
|
"""``"ours"``/``"other"``/``None`` from the process's OWN profile/home selection.
|
|
|
|
Under one process per host the shared binary path proves nothing about which home a process
|
|
serves; its ``--hermes-home``/``HERMES_HOME=``/``--profile``/``-p`` selection does. Same
|
|
token-exact parsers ``gateway/run.py::_argv_contradicts_home`` uses.
|
|
"""
|
|
home_value = _argv_flag_value(argv, _HOME_FLAGS) or _argv_env_home(argv)
|
|
if home_value:
|
|
return "ours" if _norm_path(home_value) == _norm_path(this_home) else "other"
|
|
profile_value = _argv_flag_value(argv, _PROFILE_FLAGS)
|
|
if profile_value:
|
|
if our_profile is not None:
|
|
return "ours" if profile_value == our_profile else "other"
|
|
# Root/custom home: any explicit named profile selects a different home.
|
|
return "ours" if (install_root is not None and profile_value == "default") else "other"
|
|
return None
|
|
|
|
|
|
def _argv_path_tokens(argv: Sequence[str]) -> List[Tuple[int, str]]:
|
|
"""``(argv index, normalized absolute path)`` for every path-bearing token."""
|
|
tokens: List[Tuple[int, str]] = []
|
|
for index, token in enumerate(argv):
|
|
if not isinstance(token, str):
|
|
continue
|
|
if token.startswith("/"):
|
|
path_token = token
|
|
elif token.startswith("-") and "=" in token:
|
|
# ``--db=/abs/path``-style options carry a path value; anchor on
|
|
# the text after '=' so normpath does not prepend the option.
|
|
value = token.split("=", 1)[1]
|
|
path_token = value if value.startswith("/") else None
|
|
else:
|
|
path_token = None
|
|
if path_token is not None:
|
|
tokens.append((index, _norm_path(path_token)))
|
|
return tokens
|
|
|
|
|
|
def _argv_scoped_to_other_home(argv: Sequence[str], db_path: Path) -> bool:
|
|
"""Return whether argv proves the process belongs to a DIFFERENT instance.
|
|
|
|
``state.db`` lives at the HERMES_HOME root, so an absolute-path token
|
|
containing a ``/.hermes`` segment (or naming a ``state.db``/WAL/SHM under
|
|
some other parent) identifies that token's own Hermes home. When at least
|
|
one such token exists AND no token references this instance's state.db,
|
|
its sidecars, or its home directory, the process provably works on a
|
|
different generation and must not be counted as an uninspectable holder
|
|
of ours (issue #92401: a second gateway under /home/demo/.hermes deferred
|
|
this instance's stale-FTS rebuild forever despite lsof proving zero open
|
|
handles). Ambiguous argv without absolute-path tokens returns False and
|
|
keeps the fail-closed suspicion.
|
|
|
|
Evidence is ranked, because one host now runs ONE process for every profile:
|
|
|
|
1. A token naming our state.db or a sidecar exactly — definitive, ours.
|
|
2. The process's own ``--hermes-home``/``HERMES_HOME=``/``--profile``/``-p``
|
|
selection — that is what decides which home a multiplexer serves.
|
|
3. Path tokens. A token under ``<install root>/profiles/<other>`` is another
|
|
profile's store even though it sits under our root; a token that names only
|
|
the SHARED install root is NEUTRAL (it is the same binary for every profile,
|
|
so it can neither prove nor disprove a hold); ``argv[0]`` locates the INSTALL,
|
|
not the home, so it is not other-home evidence for a store whose home is not
|
|
part of an install layout (a custom ``HERMES_HOME`` is served BY the binary
|
|
under ``~/.hermes`` — dismissing on it admits maintenance under a live writer).
|
|
"""
|
|
db_path_str = os.path.abspath(os.fspath(db_path))
|
|
this_home = os.path.dirname(db_path_str)
|
|
install_root, our_profile = _store_install_layout(this_home)
|
|
sidecars = {os.path.normcase(candidate) for candidate in _sqlite_family(db_path_str)}
|
|
this_home_norm = os.path.normcase(this_home)
|
|
root_norm = os.path.normcase(install_root) if install_root else None
|
|
path_tokens = _argv_path_tokens(argv)
|
|
|
|
if any(normalized in sidecars for _, normalized in path_tokens):
|
|
return False
|
|
selection = _argv_home_selection(argv, this_home, install_root, our_profile)
|
|
if selection == "ours":
|
|
return False
|
|
# A store whose home is not itself part of an install layout cannot be identified from the
|
|
# install location, so argv[0] alone never dismisses a holder of it.
|
|
argv0_locates_home = install_root is not None
|
|
other_home_seen = selection == "other"
|
|
for index, normalized in path_tokens:
|
|
if _names_other_profile(normalized, install_root, our_profile):
|
|
other_home_seen = True
|
|
continue
|
|
if normalized == this_home_norm or normalized.startswith(this_home_norm + os.sep):
|
|
return False
|
|
if root_norm is not None and (
|
|
normalized == root_norm or normalized.startswith(root_norm + os.sep)
|
|
):
|
|
continue # shared install root: neutral, every served profile lives under it
|
|
if index == 0 and not argv0_locates_home:
|
|
continue
|
|
if "/.hermes" in normalized or normalized.endswith("/.hermes"):
|
|
other_home_seen = True
|
|
elif os.path.basename(normalized) in _STATE_DB_NAMES:
|
|
other_home_seen = True
|
|
return other_home_seen
|
|
|
|
|
|
# Restart Manager ctypes metadata lives at module level: ``ctypes.POINTER()`` on a freshly defined Structure
|
|
# pins it in ``ctypes._pointer_type_cache`` forever, so rebuilding these per scan leaked one class set per call.
|
|
_RM_ERROR_MORE_DATA = 234
|
|
_RM_SESSION_KEY_LEN = 33 # CCH_RM_SESSION_KEY + 1
|
|
|
|
|
|
class _RmUniqueProcess(ctypes.Structure):
|
|
_fields_ = [("pid", wintypes.DWORD), ("started", wintypes.FILETIME)]
|
|
|
|
|
|
class _RmProcessInfo(ctypes.Structure):
|
|
_fields_ = [
|
|
("process", _RmUniqueProcess),
|
|
("app_name", wintypes.WCHAR * 256), # CCH_RM_MAX_APP_NAME + 1
|
|
("service_name", wintypes.WCHAR * 64), # CCH_RM_MAX_SVC_NAME + 1
|
|
("app_type", wintypes.DWORD),
|
|
("app_status", wintypes.ULONG),
|
|
("ts_session_id", wintypes.DWORD),
|
|
("restartable", wintypes.BOOL),
|
|
]
|
|
|
|
|
|
_RM_ARGTYPES = {
|
|
"RmStartSession": [ctypes.POINTER(wintypes.DWORD), wintypes.DWORD, wintypes.LPWSTR],
|
|
"RmRegisterResources": [
|
|
wintypes.DWORD, wintypes.UINT, ctypes.POINTER(wintypes.LPCWSTR),
|
|
wintypes.UINT, ctypes.c_void_p, wintypes.UINT, ctypes.c_void_p,
|
|
],
|
|
"RmGetList": [
|
|
wintypes.DWORD, ctypes.POINTER(wintypes.UINT), ctypes.POINTER(wintypes.UINT),
|
|
ctypes.POINTER(_RmProcessInfo), ctypes.POINTER(wintypes.DWORD),
|
|
],
|
|
"RmEndSession": [wintypes.DWORD],
|
|
}
|
|
|
|
|
|
def _sqlite_family(base: str) -> Tuple[str, str, str]:
|
|
"""The main database file plus the ``-wal``/``-shm`` sidecars SQLite may hold alongside it."""
|
|
return (base, base + "-wal", base + "-shm")
|
|
|
|
|
|
def _windows_restart_manager_holders(db_path: Path) -> List[Tuple[int, str]]:
|
|
"""Return foreign processes using state.db or a WAL sidecar via Windows Restart Manager."""
|
|
# WinDLL per call (not memoised) so the unit test can inject a fake rstrtmgr through ctypes.WinDLL.
|
|
api = ctypes.WinDLL("rstrtmgr", use_last_error=True)
|
|
start, register, get_list, end = (
|
|
api.RmStartSession, api.RmRegisterResources, api.RmGetList, api.RmEndSession
|
|
)
|
|
for name, fn in zip(_RM_ARGTYPES, (start, register, get_list, end)):
|
|
fn.argtypes = _RM_ARGTYPES[name]
|
|
fn.restype = wintypes.DWORD
|
|
|
|
db_abspath = os.path.abspath(os.fspath(db_path))
|
|
resources = [path for path in _sqlite_family(db_abspath) if os.path.exists(path)]
|
|
if not resources:
|
|
return []
|
|
|
|
session = wintypes.DWORD()
|
|
key = ctypes.create_unicode_buffer(_RM_SESSION_KEY_LEN)
|
|
rc = start(ctypes.byref(session), 0, key)
|
|
if rc:
|
|
raise OSError(rc, "RmStartSession failed")
|
|
try:
|
|
filenames = (wintypes.LPCWSTR * len(resources))(*resources)
|
|
rc = register(session, len(resources), filenames, 0, None, 0, None)
|
|
if rc:
|
|
raise OSError(rc, "RmRegisterResources failed")
|
|
|
|
# Pass 1 sizes (ERROR_MORE_DATA); the process set can change before the data pass, so retry the
|
|
# bounded race with a re-sized buffer.
|
|
needed, reasons = wintypes.UINT(), wintypes.DWORD()
|
|
for _ in range(4):
|
|
apps = (_RmProcessInfo * needed.value)()
|
|
count = wintypes.UINT(needed.value)
|
|
rc = get_list(
|
|
session, ctypes.byref(needed), ctypes.byref(count),
|
|
apps if needed.value else None, ctypes.byref(reasons),
|
|
)
|
|
if rc == 0:
|
|
own_pid = os.getpid()
|
|
return [
|
|
(int(apps[index].process.pid), db_abspath)
|
|
for index in range(count.value)
|
|
if int(apps[index].process.pid) != own_pid
|
|
]
|
|
if rc != _RM_ERROR_MORE_DATA:
|
|
raise OSError(rc, "RmGetList failed")
|
|
raise RuntimeError("Restart Manager holder set kept changing")
|
|
finally:
|
|
end(session)
|
|
|
|
|
|
def foreign_state_db_holders(db_path: Path) -> List[Tuple[int, str]]:
|
|
"""Return foreign holders of the DB or one of its WAL sidecars.
|
|
|
|
A scan failure is represented as an unknown holder. Structural maintenance
|
|
must not assume quiescence when an old, unlinked SQLite generation may
|
|
still be open by another process.
|
|
"""
|
|
if _IS_WINDOWS:
|
|
try:
|
|
return _windows_restart_manager_holders(db_path)
|
|
except Exception as exc:
|
|
logger.warning(
|
|
"Could not prove state.db has no Windows holders; deferring structural maintenance: %s",
|
|
exc,
|
|
)
|
|
return [(-1, f"Windows Restart Manager scan failed: {exc}")]
|
|
|
|
# realpath, not abspath: psutil/libproc report the kernel-resolved pathname, so a symlinked
|
|
# HERMES_HOME would otherwise make every holder invisible and let maintenance proceed.
|
|
db_path_str = os.path.realpath(os.fspath(db_path))
|
|
watched = {canonical_sqlite_path(candidate) for candidate in _sqlite_family(db_path_str)}
|
|
holders: List[Tuple[int, str]] = []
|
|
watched_ids: Set[Tuple[int, int]] = set()
|
|
db_dev: Optional[int] = None
|
|
for candidate in _sqlite_family(db_path_str):
|
|
try:
|
|
stat_result = os.stat(candidate)
|
|
except OSError as exc:
|
|
if exc.errno not in (errno.ENOENT, errno.ESRCH):
|
|
holders.append(
|
|
(-1, f"watched-file stat failed: {candidate}: {exc}")
|
|
)
|
|
continue
|
|
watched_ids.add((stat_result.st_dev, stat_result.st_ino))
|
|
if candidate == db_path_str:
|
|
db_dev = stat_result.st_dev
|
|
|
|
if sys.platform.startswith("linux"):
|
|
try:
|
|
own_pid = os.getpid()
|
|
for pid_str in os.listdir("/proc"):
|
|
if not pid_str.isdigit():
|
|
continue
|
|
pid = int(pid_str)
|
|
if pid == own_pid:
|
|
continue
|
|
fd_dir = f"/proc/{pid}/fd"
|
|
try:
|
|
fds = os.listdir(fd_dir)
|
|
except OSError:
|
|
argv = _read_proc_argv(pid)
|
|
if (
|
|
argv is not None
|
|
and _looks_like_hermes(argv)
|
|
and not _argv_scoped_to_other_home(argv, db_path)
|
|
):
|
|
cmdline = " ".join(argv)
|
|
holders.append((pid, f"uninspectable holder: {cmdline[:80]}"))
|
|
continue
|
|
for fd in fds:
|
|
fd_path = f"{fd_dir}/{fd}"
|
|
try:
|
|
target = os.readlink(fd_path)
|
|
except OSError as exc:
|
|
if exc.errno in (errno.ENOENT, errno.ESRCH):
|
|
continue
|
|
argv = _read_proc_argv(pid)
|
|
if (
|
|
argv is not None
|
|
and _looks_like_hermes(argv)
|
|
and not _argv_scoped_to_other_home(argv, db_path)
|
|
):
|
|
holders.append(
|
|
(
|
|
pid,
|
|
f"uninspectable descriptor: {fd_path}: {exc}",
|
|
)
|
|
)
|
|
continue
|
|
target_is_watched = canonical_sqlite_path(target) in watched
|
|
try:
|
|
fd_stat = os.stat(fd_path)
|
|
except OSError as exc:
|
|
if exc.errno in (errno.ENOENT, errno.ESRCH):
|
|
continue
|
|
if target_is_watched:
|
|
holders.append(
|
|
(pid, f"uninspectable descriptor: {target}: {exc}")
|
|
)
|
|
else:
|
|
argv = _read_proc_argv(pid)
|
|
if (
|
|
argv is not None
|
|
and _looks_like_hermes(argv)
|
|
and not _argv_scoped_to_other_home(argv, db_path)
|
|
):
|
|
holders.append(
|
|
(
|
|
pid,
|
|
"uninspectable descriptor: "
|
|
f"{target}: {exc}",
|
|
)
|
|
)
|
|
continue
|
|
if (fd_stat.st_dev, fd_stat.st_ino) in watched_ids or (
|
|
target_is_watched
|
|
and target.endswith(" (deleted)")
|
|
and db_dev is not None
|
|
and fd_stat.st_dev == db_dev
|
|
):
|
|
holders.append((pid, target))
|
|
except Exception as exc:
|
|
logger.warning(
|
|
"Could not prove state.db has no foreign holders; "
|
|
"deferring structural maintenance: %s",
|
|
exc,
|
|
)
|
|
holders.append((-1, f"open-file scan failed: {exc}"))
|
|
return holders
|
|
|
|
if psutil is None:
|
|
return [(-1, "open-file scan unavailable")]
|
|
try:
|
|
for process in psutil.process_iter(["pid", "open_files"]):
|
|
info = process.info
|
|
pid = int(info["pid"])
|
|
if pid == os.getpid():
|
|
continue
|
|
for opened in info.get("open_files") or ():
|
|
path = getattr(opened, "path", "")
|
|
if path and canonical_sqlite_path(os.path.realpath(path)) in watched:
|
|
holders.append((pid, path))
|
|
except Exception as exc:
|
|
logger.warning(
|
|
"Could not prove state.db has no foreign holders; "
|
|
"deferring structural maintenance: %s",
|
|
exc,
|
|
)
|
|
holders.append((-1, f"open-file scan failed: {exc}"))
|
|
return holders
|
|
|
|
|
|
def in_process_state_db_holders(
|
|
db_path: Path, *, exclude=None
|
|
) -> List[Tuple[int, str]]:
|
|
"""Return holders of ``db_path`` inside THIS process, other than *exclude*.
|
|
|
|
:func:`foreign_state_db_holders` skips ``os.getpid()`` by design, so it answers a
|
|
cross-PROCESS question only. Consumers that read "no holders" as "the store is quiet"
|
|
(auto-VACUUM admission) need this arm too: a VACUUM plus its TRUNCATE checkpoint retires
|
|
the generation a sibling SessionDB in this very process still holds.
|
|
"""
|
|
from hermes_state_registry import other_generations_for_path
|
|
|
|
return [
|
|
(os.getpid(), description)
|
|
for description in other_generations_for_path(db_path, exclude=exclude)
|
|
]
|
|
|
|
|
|
def held_store_refusal(db_path: Path, *, command: str, force_hint: Optional[str] = "--force") -> Optional[str]:
|
|
"""Operator-facing refusal for structural maintenance (VACUUM, index rebuild, bulk delete) while another
|
|
process holds ``db_path`` or a WAL sidecar; ``None`` when the store is provably quiet.
|
|
|
|
Running ``hermes sessions optimize-storage`` underneath a fleet of live gateways put every agent into
|
|
the retired-WAL refusal until all writers were stopped (#110054). Same fail-closed scan doctor and
|
|
repair use: an incomplete scan refuses too, it never reads as an all-clear.
|
|
"""
|
|
holders = foreign_state_db_holders(db_path)
|
|
if not holders:
|
|
return None
|
|
from hermes_constants import profile_cli_selector
|
|
from hermes_state_errors import STORAGE_RECOVERY_DOCS_URL
|
|
|
|
by_pid: dict[int, Set[str]] = {}
|
|
unknown: List[str] = []
|
|
for pid, target in holders:
|
|
if pid <= 0 or target.startswith("uninspectable"):
|
|
unknown.append(target)
|
|
else:
|
|
by_pid.setdefault(pid, set()).add(Path(target.removesuffix(" (deleted)")).name)
|
|
lines = [f"Refusing `hermes sessions {command}`: another process is using {db_path}."]
|
|
lines += [f" {describe_holder_pid(pid)}: {', '.join(sorted(by_pid[pid]))}" for pid in sorted(by_pid)]
|
|
if unknown:
|
|
lines.append(f" cannot prove the database is quiet (holder scan incomplete: {unknown[0][:120]})")
|
|
profile_arg = profile_cli_selector()
|
|
lines += [
|
|
"Rewriting the database under a live writer is how every agent ends up refusing turns with the "
|
|
"retired state.db-wal error. Nothing is lost.",
|
|
f"Stop them first (`hermes {profile_arg}gateway stop`, quit the Desktop app, pause cron), then re-run.",
|
|
]
|
|
if force_hint:
|
|
lines.append(f"Override with {force_hint} if you accept the risk.")
|
|
lines.append(f"Recovery guide: {STORAGE_RECOVERY_DOCS_URL}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def live_writer_holds_db(
|
|
db_path: Path,
|
|
*,
|
|
connect_repair_durable: Callable[..., sqlite3.Connection],
|
|
) -> bool:
|
|
"""Return whether repair lacks proven exclusive ownership of ``db_path``.
|
|
|
|
ANY foreign process holding the DB or a sidecar is a live holder (#103339): the lock probe below
|
|
cannot see a DELETE-mode reader (SHARED only) and cannot run at all on a malformed file, and those
|
|
are exactly the states repair/VACUUM/checkpoint get invoked in. The holder scan is the authority and
|
|
fails closed on its own failures (unknown/uninspectable sentinels); the probe only adds a positive
|
|
lock signal on top."""
|
|
if foreign_state_db_holders(db_path):
|
|
return True
|
|
|
|
probe = None
|
|
try:
|
|
probe = connect_repair_durable(db_path, timeout=0.0)
|
|
probe.execute("PRAGMA locking_mode=EXCLUSIVE")
|
|
probe.execute("BEGIN IMMEDIATE")
|
|
probe.execute("ROLLBACK")
|
|
return False
|
|
except sqlite3.OperationalError as exc:
|
|
return is_sqlite_lock_error(exc)
|
|
except sqlite3.DatabaseError:
|
|
# Malformed/unreadable with no holder on the scan: nobody else has it open, so repair may run.
|
|
return False
|
|
finally:
|
|
if probe is not None:
|
|
try:
|
|
probe.execute("PRAGMA locking_mode=NORMAL")
|
|
except Exception:
|
|
pass
|
|
try:
|
|
probe.close()
|
|
except Exception:
|
|
pass
|