From c4a6c8fbc62e3728da7444b26923410814b545d5 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 22:56:57 -0700 Subject: [PATCH] refactor(state): compact hermes_state_common comments and docstrings, keeping every invariant and failure mode --- hermes_state_common.py | 212 +++++++++++++++++------------------------ 1 file changed, 90 insertions(+), 122 deletions(-) diff --git a/hermes_state_common.py b/hermes_state_common.py index 7d23eb096b..5027bb053b 100644 --- a/hermes_state_common.py +++ b/hermes_state_common.py @@ -1,8 +1,5 @@ -"""Shared constants and helpers for the SessionDB family of modules. - -Lives outside hermes_state so the mixin modules can import it without a cycle; -hermes_state re-exports every name for backward compatibility. -""" +"""Shared constants and helpers for the SessionDB family of modules. Lives outside hermes_state so +the mixin modules can import it without a cycle; hermes_state re-exports every name.""" import contextlib import errno @@ -18,19 +15,17 @@ from agent.context_compressor import (LEGACY_SUMMARY_PREFIX, SUMMARY_PREFIX, _ME _MERGED_SUMMARY_DELIMITER, _SUMMARY_END_MARKER) -# Session preview = head of the first user message, shown wherever a session has no title. A /skill -# invocation embeds the whole skill body, so its plain head would preview the SKILL's prose; scaffolded rows -# carry a wider excerpt (whole message under budget, else head + tail where the typed instruction lands) so -# ``_shape_preview`` can recover ``/work — fix the title leak``. +# Session preview = head of the first user message (shown when a session has no title). A /skill invocation +# embeds the whole skill body, so scaffolded rows carry a wider excerpt (whole message under budget, else +# head + tail where the typed instruction lands) so ``_shape_preview`` can recover ``/work — fix ...``. _PREVIEW_HEAD_CHARS = 63 _PREVIEW_SCAFFOLD_WINDOW = 400 _PREVIEW_MAX_CHARS = 60 def escape_like(text: str) -> str: - """Escape LIKE wildcards (``%``, ``_``) so derived text matches literally; pair with - ``ESCAPE '\\'``. ``_`` is common in branch names, titles and paths, and a documented - substring/prefix match must not silently widen.""" + """Escape LIKE wildcards (``%``, ``_``) so derived text matches literally; pair with ``ESCAPE '\\'``. + ``_`` is common in branch names/titles/paths and a substring match must not silently widen.""" return text.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") @@ -61,8 +56,8 @@ def _sql_after_marker(marker: str) -> str: return f"SUBSTR(m.content, INSTR(m.content, {_sql_literal(marker)}) + {len(marker)})" -# Current and legacy long-form prefixes share this whole introduction; matching all of it keeps an ordinary -# message that merely starts with the bracketed label from counting as a compaction carrier. +# Current and legacy long-form prefixes share this introduction; matching all of it keeps an ordinary message +# that merely starts with the bracketed label from counting as a compaction carrier. _PREVIEW_LONG_FORM_PREFIX = SUMMARY_PREFIX.split("Do NOT answer", 1)[0] _PREVIEW_SUMMARY_PREFIXES = (_PREVIEW_LONG_FORM_PREFIX, LEGACY_SUMMARY_PREFIX) _PREVIEW_STANDALONE_SUMMARY_SQL = _sql_starts_with("m.content", _PREVIEW_SUMMARY_PREFIXES) @@ -78,16 +73,14 @@ _PREVIEW_MERGED_PRIOR_UNWRAPPED_SQL = (f"CASE WHEN SUBSTR({_PREVIEW_MERGED_PRIOR f" ELSE {_PREVIEW_MERGED_PRIOR_SQL} END") _PREVIEW_FORCE_USER_REMAINDER_SQL = _sql_after_marker(_SUMMARY_END_MARKER) -# Pure compaction rows are ineligible for previews; force-user-leading and merged -# carriers are eligible only when authentic content survives. +# Pure compaction rows are ineligible; force-user-leading and merged carriers only when authentic content survives. _PREVIEW_ELIGIBLE_SQL = (f"((NOT {_PREVIEW_STANDALONE_SUMMARY_SQL} AND NOT {_PREVIEW_MERGED_SUMMARY_SQL})" f" OR ({_PREVIEW_STANDALONE_SUMMARY_SQL} AND INSTR(m.content, {_sql_literal(_SUMMARY_END_MARKER)}) > 0" f" AND LENGTH({_sql_trim_whitespace(_PREVIEW_FORCE_USER_REMAINDER_SQL)}) > 0)" f" OR ({_PREVIEW_MERGED_SUMMARY_SQL}" f" AND LENGTH({_sql_trim_whitespace(_PREVIEW_MERGED_PRIOR_UNWRAPPED_SQL)}) > 0))") -# Shared ``_preview_raw`` SELECT expression for every listing query (scaffolded rows: -# head + tail spliced around SKILL_EXCERPT_JOINT when over budget). +# ``_preview_raw`` SELECT for every listing query (scaffolded rows: head + tail around SKILL_EXCERPT_JOINT). _PREVIEW_RAW_SELECT = ( f"CASE WHEN {_PREVIEW_STANDALONE_SUMMARY_SQL} THEN {_PREVIEW_FORCE_USER_REMAINDER_SQL}" f" WHEN {_PREVIEW_MERGED_SUMMARY_SQL} THEN {_PREVIEW_MERGED_PRIOR_UNWRAPPED_SQL}" @@ -116,53 +109,48 @@ _PREVIEW_RAW_SUBQUERY_SQL = (f"COALESCE((SELECT {_PREVIEW_RAW_SELECT} FROM messa # ── Session lineage predicates ({a} = sessions alias) ─────────────────────── -# A /branch child (kept visible, never cascade-deleted): stable marker OR the legacy end_reason heuristic. +# /branch child (kept visible, never cascade-deleted): stable marker OR legacy end_reason heuristic. _BRANCH_CHILD_SQL = ("json_extract(COALESCE({a}.model_config, '{{}}'), '$._branched_from') IS NOT NULL" " OR EXISTS (SELECT 1 FROM sessions p WHERE p.id = {a}.parent_session_id" " AND p.end_reason = 'branched' AND {a}.started_at >= p.ended_at)") _COMPRESSION_CHILD_SQL = ("EXISTS (SELECT 1 FROM sessions p WHERE p.id = {a}.parent_session_id" " AND p.end_reason = 'compression')") -# 'session_switch': switch_session() creates no child row, but pre-marker DBs hold legacy reset children -# whose parent later ended that way. Must stay identical to the recovery fence in -# find_latest_gateway_session_for_peer (interpolates the SQL form). +# 'session_switch' creates no child row today, but pre-marker DBs hold legacy reset children whose parent +# ended that way. Must stay identical to the recovery fence in find_latest_gateway_session_for_peer. _RESET_END_REASONS = ("session_reset", "session_switch", "idle", "daily", "suspended", "resume_pending_expired") _RESET_END_REASONS_SQL = ", ".join(f"'{reason}'" for reason in _RESET_END_REASONS) -# Accidental end reasons recovery treats as resumable (docs/session-lifecycle.md). Single source of truth: -# interpolated into recovery SQL AND exposed as SessionDB.RECOVERABLE_END_REASONS. -# superseded_by_resume: stale sentinel-parked runtime superseded by a fresh session.resume. -# startup_orphan_reap: startup sweep of rows orphaned by a dead gateway process (same accident class as -# ws_orphan_reap, kept distinct for forensics). +# Accidental end reasons recovery treats as resumable (docs/session-lifecycle.md); single source of truth for +# recovery SQL and SessionDB.RECOVERABLE_END_REASONS. superseded_by_resume = sentinel-parked runtime replaced +# by a fresh session.resume; startup_orphan_reap = dead-gateway sweep, same class as ws_orphan_reap but kept +# distinct for forensics. _RECOVERABLE_END_REASONS = ("agent_close", "ws_orphan_reap", "superseded_by_resume", "startup_orphan_reap") _RECOVERABLE_END_REASONS_SQL = ", ".join(f"'{reason}'" for reason in _RECOVERABLE_END_REASONS) -# End reasons written by AUTOMATIC cleanup (shutdown, orphan reapers, idle/LRU eviction), not by a -# deliberate conversation boundary: "some runtime went away", NOT "this conversation ended", so a writer -# that can prove liveness (e.g. a compression rotation holding the lease) may clear it. Superset of the -# recoverable set plus the TUI gateway's automatic reasons. +# End reasons written by AUTOMATIC cleanup (shutdown, orphan reapers, idle/LRU eviction), not a deliberate +# conversation boundary: "some runtime went away", so a writer that can prove liveness (e.g. a compression +# rotation holding the lease) may clear it. Recoverable set plus the TUI gateway's automatic reasons. _AUTOMATIC_END_REASONS = frozenset(_RECOVERABLE_END_REASONS) | { "tui_shutdown", "ws_disconnect", "idle_timeout", "lru_evict"} def is_automatic_end_reason(reason) -> bool: - """True when *reason* is an automatic-cleanup end stamp. Single owner of the - accidental-vs-deliberate predicate; compression-liveness sites must call this.""" + """True when *reason* is an automatic-cleanup end stamp; compression-liveness sites must call this.""" return isinstance(reason, str) and reason in _AUTOMATIC_END_REASONS def _legacy_reset_child_sql(alias: str, reasons_sql: str) -> str: """Pre-marker reset-continuation heuristic: child rides its parent's exact non-empty routing key and the - parent ended at a reset boundary. Shared by ``_RESET_CHILD_SQL`` and ``reopen_session()``'s - marker-stamping UPDATE so the two cannot drift; ``reasons_sql`` is a literal or placeholder list.""" + parent ended at a reset boundary. Shared by ``_RESET_CHILD_SQL`` and ``reopen_session()`` so the two + cannot drift; ``reasons_sql`` is a literal or placeholder list.""" return (f"EXISTS (SELECT 1 FROM sessions p WHERE p.id = {alias}.parent_session_id" f" AND p.end_reason IN ({reasons_sql}) AND {alias}.session_key IS NOT NULL" f" AND {alias}.session_key != '' AND {alias}.session_key = p.session_key)") -# A reset starts a separate user-visible conversation though rows keep parent_session_id -# for lineage. Stable marker, or the same-key fallback for pre-marker rows (the -# exact-key requirement keeps subagent children out). +# A reset starts a separate user-visible conversation though rows keep parent_session_id for lineage. +# Stable marker, or the same-key fallback for pre-marker rows (exact key keeps subagent children out). _RESET_CHILD_SQL = ("json_extract(COALESCE({a}.model_config, '{{}}'), '$._reset_from') IS NOT NULL" " OR " + _legacy_reset_child_sql("{a}", _RESET_END_REASONS_SQL)) @@ -178,9 +166,8 @@ def _ephemeral_child_sql(alias: str = "s") -> str: def _sql_freshest_of(activity: str, session_id_expr: str, started: str) -> str: - """Freshest of *activity* and the latest message timestamp for *session_id_expr*, - else *started*. Heartbeats are rate-limited (~60s) so ``last_activity_at`` can lag - a newer message; never prefer it alone.""" + """Freshest of *activity* and the latest message timestamp for *session_id_expr*, else *started*. + Heartbeats are rate-limited (~60s) so ``last_activity_at`` can lag a newer message; never use it alone.""" msg_max = f"(SELECT MAX(_act_m.timestamp) FROM messages _act_m WHERE _act_m.session_id = {session_id_expr})" return (f"COALESCE((SELECT MAX(_act_v.v) FROM (SELECT {activity} AS v UNION ALL SELECT {msg_max}) _act_v), " f"{started})") @@ -200,14 +187,12 @@ def _sql_session_last_active_by_id(session_id_expr: str) -> str: SCHEMA_VERSION = 28 -# Auto-maintenance VACUUMs only when at least this fraction of pages is on the freelist; -# below it a full rewrite costs more I/O than it returns. +# Auto-maintenance VACUUMs only above this freelist fraction; below it a rewrite costs more I/O than it returns. AUTO_VACUUM_MIN_FREELIST_RATIO = 0.25 -# FTS storage layout, tracked INDEPENDENTLY of SCHEMA_VERSION (state_meta ``fts_storage_version``): schema -# version advances freely on open, the FTS layout only changes when a DB is born fresh or explicitly -# optimized via ``hermes sessions optimize-storage``. Legacy DBs sit at 0 (marker absent) with a working -# inline index; 1 = v23 external-content layout. +# FTS layout, tracked INDEPENDENTLY of SCHEMA_VERSION (state_meta ``fts_storage_version``): it changes only +# when a DB is born fresh or via ``hermes sessions optimize-storage``. 0 (marker absent) = legacy inline +# index, still working; 1 = v23 external-content layout. FTS_STORAGE_VERSION = 1 # Cap on user-controlled FTS5 query input before sanitizer processing. @@ -215,8 +200,8 @@ MAX_FTS5_QUERY_CHARS = 2_048 def stat_db_file_identity(path) -> "tuple[int, int] | None": - """``(st_dev, st_ino)`` for *path*, or None. st_ino=0 (Windows, some network FS) - would false-positive every replaced-file check, so it counts as unknown.""" + """``(st_dev, st_ino)`` for *path*, or None. st_ino=0 (Windows, some network FS) would false-positive + every replaced-file check, so it counts as unknown.""" try: st = os.stat(path) except OSError: @@ -484,8 +469,7 @@ CREATE INDEX IF NOT EXISTS idx_async_delegations_delivery ON async_delegations(delivery_state, completed_at); """ -# Indexes on columns added in later schema versions must run AFTER -# _reconcile_columns() adds them, or executescript fails on legacy DBs. +# Indexes on later-added columns must run AFTER _reconcile_columns(), or executescript fails on legacy DBs. DEFERRED_INDEX_SQL = """ CREATE INDEX IF NOT EXISTS idx_messages_session_active ON messages(session_id, active, timestamp); @@ -501,13 +485,11 @@ CREATE INDEX IF NOT EXISTS idx_sessions_system_prompt_hash ON sessions(system_prompt_hash); """ -# ── Deferred FTS rebuild bookkeeping ── -# While a background rebuild is pending, two state_meta keys define which rows are IN the FTS indexes: H = -# fts_rebuild_high_water (MAX(messages.id) when the old indexes were dropped), P = fts_rebuild_progress -# (highest backfilled id). A row is indexed iff id <= P OR id > H (AUTOINCREMENT ids: post-drop rows are -# indexed live by the insert triggers); rows in (P, H] are not. Every trigger gates on that predicate: an -# external-content 'delete' for a row NOT in the index corrupts it, and skipping one for an indexed row -# leaves a stale entry. With no rebuild pending both keys are absent and COALESCE makes it a tautology. +# Deferred FTS rebuild bookkeeping: while a background rebuild is pending, state_meta H = fts_rebuild_high_water +# (MAX(messages.id) when the old indexes were dropped) and P = fts_rebuild_progress (highest backfilled id) +# define the indexed rows: id <= P OR id > H (post-drop rows are indexed live by the insert triggers); (P, H] +# is not. Every trigger gates on that predicate: an external-content 'delete' for an unindexed row corrupts +# the index and skipping one for an indexed row leaves a stale entry. No rebuild pending => COALESCE tautology. FTS_SQL = """ CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts USING fts5( content, @@ -557,11 +539,10 @@ BEGIN END; """ -# Trigram FTS5 table for CJK substring search (unicode61 splits CJK into single tokens, breaking phrase -# matching). The trigram index is ~2.6x the text it covers and ``role='tool'`` rows are ~90% of message -# bytes of machine noise, so it reads through the ``messages_fts_trigram_src`` view, which excludes tool -# rows; those remain searchable via ``messages_fts``, and ``search_messages`` routes CJK queries filtered on -# role='tool' to LIKE. +# Trigram FTS5 table for CJK substring search (unicode61 splits CJK into single tokens). The trigram index is +# ~2.6x the text it covers and ``role='tool'`` rows are ~90% of message bytes, so it reads through the +# ``messages_fts_trigram_src`` view excluding tool rows; those stay searchable via ``messages_fts`` and +# ``search_messages`` routes CJK queries filtered on role='tool' to LIKE. FTS_TRIGRAM_SQL = """ CREATE VIEW IF NOT EXISTS messages_fts_trigram_src AS SELECT id, role, content, tool_name, tool_calls @@ -621,23 +602,20 @@ END; _FTS_CJK_TRIGGERS = ("messages_fts_cjk_insert", "messages_fts_cjk_delete", "messages_fts_cjk_update") -# Set when a tokenizer-less process dropped the cjk triggers to keep writes alive: the cjk index is missing -# rows and must not serve reads until `hermes sessions optimize-storage` rebuilds it on a capable host. +# Set when a tokenizer-less process dropped the cjk triggers to keep writes alive: the cjk index is missing rows +# and must not serve reads until `hermes sessions optimize-storage` rebuilds it on a capable host. FTS_CJK_STALE_KEY = "fts_cjk_stale" -# Set when a base/trigram FTS index was detached after runtime corruption. -# While present, startup must rebuild the complete index before reinstalling -# sync triggers: rows written while they were absent leave an unknown gap. +# Set when a base/trigram FTS index was detached after runtime corruption. While present, startup must rebuild +# the complete index before reinstalling sync triggers: rows written while they were absent leave an unknown gap. FTS_STALE_KEY = "fts_stale" # Durable diagnostic for stale FTS recovery blocked across process restarts. FTS_REBUILD_DEFERRAL_KEY = "fts_rebuild_deferral" -# ── Legacy (v22 / inline-content) FTS DDL ────────────────────────────── -# Used ONLY to keep a pre-v23 install's search working and its triggers repairable until -# `optimize_fts_storage()` migrates it: inline copies of content || tool_name || tool_calls, trigram over -# every row. Never created on a fresh install. Handing a legacy DB the v23 DDL would create the -# external-content trigram VIEW and leave it in a mixed, broken state. +# Legacy (v22 / inline-content) FTS DDL: ONLY keeps a pre-v23 install's search working and its triggers +# repairable until `optimize_fts_storage()` migrates it (inline content || tool_name || tool_calls, trigram over +# every row). Never created fresh; the v23 DDL on a legacy DB would leave a mixed, broken state. LEGACY_FTS_SQL = """ CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts USING fts5( content @@ -691,19 +669,16 @@ AFTER UPDATE OF content, tool_name, tool_calls ON messages BEGIN END; """ -# ── Cross-process full-FTS-rebuild admission (single authority) ────────────── -# Several Hermes processes share one state.db; a full structural FTS rebuild (FTS5 'rebuild' or the -# drop/recreate in `_recover_stale_fts`) must run in ONE of them at a time — concurrent rebuilds -# structurally corrupted state.db in production. Single authority for `rebuild_fts()`, -# `_rebuild_fts_indexes()` and `_recover_stale_fts()`; the chunked backfill (`fts_rebuild_step`) is -# deliberately NOT routed through it (it claims progress under SQLite transaction authority and is -# multi-process). Semantics mirror `hermes_state._cross_process_repair_lock`: portable (msvcrt on Windows, -# flock elsewhere), bounded wait, FAIL CLOSED. flock rides the open file description, so a forked child -# that inherited the fd holds it forever after the holder dies; the holder's pid + start time are recorded -# under the lock and a provably-dead holder's lock is broken by unlinking and retaking on a fresh inode. -# Indeterminate liveness still defers. `.fts_rebuild.lock` is distinct from `.repair.lock` (schema -# surgery on an EXCLUSIVE offline connection, minutes in VACUUM). Lives here because mixins cannot import -# hermes_state (cycle). +# Cross-process full-FTS-rebuild admission (single authority). Several processes share one state.db and a +# structural rebuild (FTS5 'rebuild' or `_recover_stale_fts`'s drop/recreate) must run in ONE at a time — +# concurrent rebuilds corrupted state.db in production. Gates `rebuild_fts()`, `_rebuild_fts_indexes()`, +# `_recover_stale_fts()`; the chunked backfill (`fts_rebuild_step`) is deliberately NOT routed through it (it +# claims progress under SQLite transaction authority). Mirrors `hermes_state._cross_process_repair_lock`: +# portable (msvcrt/flock), bounded wait, FAIL CLOSED. flock rides the open file description, so a forked +# child holds it forever after the holder dies; holder pid + start time are recorded under the lock and a +# provably-dead holder's lock is broken by unlinking and retaking on a fresh inode. Indeterminate liveness +# defers. `.fts_rebuild.lock` is distinct from `.repair.lock` (offline schema surgery, minutes in +# VACUUM). Lives here because mixins cannot import hermes_state (cycle). logger = logging.getLogger("hermes_state") @@ -711,25 +686,22 @@ _FTS_REBUILD_LOCK_TIMEOUT_SECONDS = 120.0 _FTS_REBUILD_LOCK_POLL_SECONDS = 0.1 _IS_WINDOWS = sys.platform == "win32" -# Post-break re-acquire budget: the fresh inode is contended only by live -# processes, so a short wait suffices — never re-enter the full timeout. +# Post-break re-acquire budget: the fresh inode is contended only by live processes — never the full timeout. _LOCK_BREAK_REACQUIRE_SECONDS = 5.0 -# "Another process holds the lock": flock → EWOULDBLOCK/EAGAIN, msvcrt.locking → EACCES (EDEADLK when its -# retry gives up). Anything else (ESTALE, ENOTSUP, ENOLCK, EIO) is a persistent environment failure that -# polling cannot fix; treating it as contention burned the full timeout on every attempt. +# "Another process holds the lock": flock → EWOULDBLOCK/EAGAIN, msvcrt.locking → EACCES (EDEADLK when its retry +# gives up). Anything else (ESTALE, ENOTSUP, ENOLCK, EIO) is a persistent failure polling cannot fix. _LOCK_CONTENTION_ERRNOS = {errno.EAGAIN, errno.EACCES, errno.EWOULDBLOCK, errno.EDEADLK} def is_advisory_lock_contention(exc: BaseException) -> bool: - """True when *exc* means another process holds the advisory lock. For any - other ``OSError`` callers must fail closed IMMEDIATELY: retrying cannot succeed.""" + """True when *exc* means another process holds the lock; on any other ``OSError`` fail closed at once.""" return isinstance(exc, BlockingIOError) or (isinstance(exc, OSError) and exc.errno in _LOCK_CONTENTION_ERRNOS) def _proc_start_ticks(pid: int): - """Kernel start time of *pid* (field 22 of ``/proc//stat``), which with the PID uniquely identifies - a process; None off Linux or on any failure — callers must treat None as unknowable and FAIL CLOSED.""" + """Kernel start time of *pid* (field 22 of ``/proc//stat``; with the PID it identifies a process + uniquely). None off Linux or on any failure — callers must treat None as unknowable and FAIL CLOSED.""" try: with open(f"/proc/{pid}/stat", "rb") as fh: stat = fh.read() @@ -761,22 +733,20 @@ def _rewrite_lock_file(handle, payload: bytes) -> None: def _write_lock_holder_record(handle) -> None: - """Record this process as holder (best effort), written under the flock so timed-out - contenders can tell an orphaned-fd holder from a live wedged one.""" + """Record this process as holder (best effort) so timed-out contenders can tell an orphaned-fd holder + from a live wedged one.""" record = {"pid": os.getpid(), "start_ticks": _proc_start_ticks(os.getpid()), "acquired_at": time.time()} _rewrite_lock_file(handle, json.dumps(record, sort_keys=True).encode("utf-8")) def _clear_lock_holder_record(handle) -> None: - """Erase holder metadata before a normal release, so a surviving record always - means an ABNORMAL exit — the only condition allowing a break.""" + """Erase holder metadata before a normal release: a surviving record means ABNORMAL exit (break allowed).""" _rewrite_lock_file(handle, b"") def _lock_holder_provably_dead(record) -> bool: - """True ONLY when the recorded holder is provably dead or PID-recycled. Anything - indeterminate (no/malformed record, PID owned by another user, /proc unavailable) - is False — the caller must FAIL CLOSED and defer.""" + """True ONLY when the recorded holder is provably dead or PID-recycled. Anything indeterminate + (no/malformed record, PID owned by another user, /proc unavailable) is False: FAIL CLOSED and defer.""" if not isinstance(record, dict): return False try: @@ -800,15 +770,14 @@ def _lock_holder_provably_dead(record) -> bool: def _acquire_db_flock(lock_path, handle, timeout_seconds, poll_seconds, description): - """Bounded POSIX flock acquire with orphaned-holder staleness break. Returns ``(acquired, handle)``; - *handle* may have been re-opened and the caller closes whichever comes back. *acquired* is True, False - (a holder kept the lock past the deadline), or None (non-contention ``OSError``, already logged; callers - treat it as not acquired without the held-by-another-process warning). ``flock`` belongs to the open - file DESCRIPTION, which ``fork()`` duplicates: a holder that forks then dies leaves the lock held forever - by a child that never releases. When the process that ACQUIRED is provably dead yet the flock is held, - the file is unlinked and retaken on a fresh inode; the orphan's flock stays on the old inode blocking - nobody. Every successful acquire verifies its inode still names *lock_path*, so a racer that locked a - dead inode retries instead of running alongside the breaker. Indeterminate liveness defers.""" + """Bounded POSIX flock acquire with orphaned-holder break. Returns ``(acquired, handle)``; *handle* may + have been re-opened and the caller closes whichever comes back. *acquired*: True, False (a holder kept + the lock past the deadline) or None (non-contention ``OSError``, already logged; callers treat it as not + acquired without the held-by-another-process warning). ``flock`` belongs to the open file DESCRIPTION, + which ``fork()`` duplicates, so a holder that forks then dies leaves the lock held forever; when the + acquirer is provably dead the file is unlinked and retaken on a fresh inode (the orphan's flock stays on + the old inode blocking nobody). Every successful acquire verifies its inode still names *lock_path*, + so a racer that locked a dead inode retries instead of running alongside the breaker.""" import fcntl deadline = time.monotonic() + timeout_seconds broke_lock = False @@ -843,8 +812,7 @@ def _acquire_db_flock(lock_path, handle, timeout_seconds, poll_seconds, descript broke_lock = True deadline = time.monotonic() + _LOCK_BREAK_REACQUIRE_SECONDS continue - # Verify the path still names our inode: a breaker may have replaced - # the file while we waited, and a lock on a dead inode excludes nobody. + # A breaker may have replaced the file while we waited; a lock on a dead inode excludes nobody. try: fd_stat, path_stat = os.fstat(handle.fileno()), os.stat(lock_path) same_file = fd_stat.st_dev == path_stat.st_dev and fd_stat.st_ino == path_stat.st_ino @@ -894,10 +862,11 @@ def _acquire_msvcrt_lock(lock_path, handle, timeout): @contextlib.contextmanager def fts_rebuild_admission(db_path, *, timeout_seconds=None): - """Serialize full structural FTS rebuilds on *db_path* across processes. Yields True when this process holds the authority, False when the bounded acquire timed out or the lock - file could not be opened. On False the caller must NOT rebuild (fail closed); the stale breadcrumb - guarantees a retry. ``db_path`` None (in-memory DB) yields True. Opportunistic in-process retries pass - ``timeout_seconds=0`` so a live holder never stalls a long-lived writer; the orphan break still applies.""" + """Serialize full structural FTS rebuilds on *db_path* across processes. Yields True when this process + holds the authority, False when the bounded acquire timed out or the lock file could not be opened; on + False the caller must NOT rebuild (fail closed; the stale breadcrumb guarantees a retry). ``db_path`` + None (in-memory DB) yields True. Opportunistic in-process retries pass ``timeout_seconds=0`` so a live + holder never stalls a long-lived writer; the orphan break still applies.""" if db_path is None: yield True return @@ -906,11 +875,10 @@ def fts_rebuild_admission(db_path, *, timeout_seconds=None): try: handle = open(lock_path, "a+b") except OSError as exc: - # Fail closed like a timed-out acquire: an unopenable lock file means the FS is - # out of space/inodes/descriptors, and a sibling that opened its handle earlier - # may still be rebuilding — yielding True gave every process on a full disk a - # concurrent rebuild of the same DB. Deferring costs nothing (the breadcrumb - # retries, and the rebuild's own writes could not have committed either). + # Fail closed like a timed-out acquire: an unopenable lock file means the FS is out of + # space/inodes/descriptors and a sibling that opened earlier may still be rebuilding — yielding True + # gave every process on a full disk a concurrent rebuild. Deferring costs nothing (the breadcrumb + # retries; the rebuild's own writes could not have committed either). logger.warning("Could not open FTS rebuild lock %s (%s) — deferring this rebuild " "rather than running it without cross-process authority.", lock_path, exc) yield False