From 8e372bc5a91f38d5e6a5bf17c6716a3104e7f24d Mon Sep 17 00:00:00 2001 From: Tom Softreck Date: Fri, 30 Oct 2026 09:08:36 +0100 Subject: [PATCH 1/2] fix(index): close connections and preserve busy rebuild owners (ticket-260) --- planfile/core/sqlite_index.py | 73 +++++++++++------ planfile/core/store.py | 19 ++--- project/TICKETS.md | 1 + project/ticket-260/README.md | 21 +++++ project/ticket-260/intent.json | 92 +++++++++++++++++++++ tests/test_sqlite_ticket_index.py | 132 ++++++++++++++++++++++++++++++ 6 files changed, 302 insertions(+), 36 deletions(-) create mode 100644 project/ticket-260/README.md create mode 100644 project/ticket-260/intent.json diff --git a/planfile/core/sqlite_index.py b/planfile/core/sqlite_index.py index af83209..31da19f 100644 --- a/planfile/core/sqlite_index.py +++ b/planfile/core/sqlite_index.py @@ -5,7 +5,8 @@ import json import sqlite3 import time -from collections.abc import Iterable +from collections.abc import Iterable, Iterator +from contextlib import closing, contextmanager from itertools import islice from pathlib import Path from typing import Any @@ -18,6 +19,17 @@ _SOURCE_INDEXED_AT_NS_KEY = "source_indexed_at_ns" +def _is_sqlite_contention(error: sqlite3.DatabaseError) -> bool: + code = getattr(error, "sqlite_errorcode", None) + if code is not None: + return isinstance(code, int) and code & 0xFF in (5, 6) + # Python 3.10 lacks sqlite_errorcode; do not classify arbitrary messages. + message = str(error).lower() + return message in { + "database is locked", "database table is locked", "database schema is locked" + } or message.startswith(("database table is locked: ", "database schema is locked: ")) + + class SQLiteTicketIndex: """Indexed projection of YAML/JSONL sources. @@ -35,20 +47,31 @@ def serialize_signature(signature: tuple) -> str: def _connect(self) -> sqlite3.Connection: self.path.parent.mkdir(parents=True, exist_ok=True) connection = sqlite3.connect(self.path, timeout=10) - connection.row_factory = sqlite3.Row - journal_mode = connection.execute("PRAGMA journal_mode").fetchone()[0] - if str(journal_mode).lower() != "wal": - connection.execute("PRAGMA journal_mode=WAL") - connection.execute("PRAGMA synchronous=NORMAL") - connection.execute("PRAGMA foreign_keys=ON") - initialized = connection.execute( - "SELECT 1 FROM sqlite_master WHERE type='table' AND name='meta'" - ).fetchone() - if initialized is None: - self._initialize(connection) - self._ensure_routes(connection) + try: + connection.row_factory = sqlite3.Row + journal_mode = connection.execute("PRAGMA journal_mode").fetchone()[0] + if str(journal_mode).lower() != "wal": + connection.execute("PRAGMA journal_mode=WAL") + connection.execute("PRAGMA synchronous=NORMAL") + connection.execute("PRAGMA foreign_keys=ON") + initialized = connection.execute( + "SELECT 1 FROM sqlite_master WHERE type='table' AND name='meta'" + ).fetchone() + if initialized is None: + self._initialize(connection) + self._ensure_routes(connection) + except BaseException: + connection.close() + raise return connection + @contextmanager + def _connection(self) -> Iterator[sqlite3.Connection]: + # SQLite's own context manager commits/rolls back but does not close. + with closing(self._connect()) as connection: + with connection: + yield connection + @staticmethod def _initialize(connection: sqlite3.Connection) -> None: connection.executescript( @@ -204,7 +227,7 @@ def _route_page(self, where: str, values: list, limit: int, if (not isinstance(after, tuple) or len(after) != 2 or type(after[0]) is not int or not isinstance(after[1], str)): raise ValueError("ticket_index_cursor_invalid") - with self._connect() as connection: + with self._connection() as connection: rows = connection.execute( f"SELECT id, position FROM ticket_routes WHERE {where} " "AND (position, id) > (?, ?) ORDER BY position, id LIMIT ?", @@ -220,7 +243,7 @@ def is_current(self, signature: tuple) -> bool: return False expected = self.serialize_signature(signature) try: - with self._connect() as connection: + with self._connection() as connection: row = connection.execute( "SELECT value FROM meta WHERE key='source_signature'" ).fetchone() @@ -272,7 +295,7 @@ def rebuild( raise ValueError("ticket_index_rebuild_batch_size_invalid") rows = iter(records) count = 0 - with self._connect() as connection: + with self._connection() as connection: connection.execute("BEGIN IMMEDIATE") connection.execute("DELETE FROM dependencies") connection.execute("DELETE FROM tickets") @@ -303,7 +326,7 @@ def rebuild( return count def get_ticket(self, ticket_id: str) -> dict[str, Any] | None: - with self._connect() as connection: + with self._connection() as connection: row = connection.execute( "SELECT ticket_json FROM tickets WHERE id=?", (ticket_id,), @@ -320,7 +343,7 @@ def apply( """Apply a known-complete store mutation and advance its source signature.""" rows = list(upserts) deleted_ids = list(deletes) - with self._connect() as connection: + with self._connection() as connection: connection.execute("BEGIN IMMEDIATE") if deleted_ids: connection.executemany( @@ -415,7 +438,7 @@ def has_fresh_snapshot(self, max_age_seconds: float) -> bool: if max_age_seconds < 0 or not self.path.exists(): return False try: - with self._connect() as connection: + with self._connection() as connection: rows = connection.execute( "SELECT key, value FROM meta WHERE key IN ('source_signature', ?)", (_SOURCE_INDEXED_AT_NS_KEY,), @@ -449,7 +472,7 @@ def list_summaries( conditions.append(f"{key}=?") parameters.append(str(getattr(value, "value", value))) where = f" WHERE {' AND '.join(conditions)}" if conditions else "" - with self._connect() as connection: + with self._connection() as connection: total = int( connection.execute( f"SELECT COUNT(*) AS count FROM tickets{where}", @@ -488,7 +511,7 @@ def list_payloads( conditions.append(f"{key}=?") parameters.append(str(getattr(value, "value", value))) where = f" WHERE {' AND '.join(conditions)}" if conditions else "" - with self._connect() as connection: + with self._connection() as connection: total = int( connection.execute( f"SELECT COUNT(*) AS count FROM tickets{where}", @@ -527,7 +550,7 @@ def render_payloads( conditions.append(f"{key}=?") parameters.append(str(getattr(value, "value", value))) where = f" WHERE {' AND '.join(conditions)}" if conditions else "" - with self._connect() as connection: + with self._connection() as connection: total = int( connection.execute( f"SELECT COUNT(*) AS count FROM tickets{where}", @@ -582,7 +605,7 @@ def render_operational_payloads( elif offset: page_sql += " LIMIT -1 OFFSET ?" page_parameters.append(offset) - with self._connect() as connection: + with self._connection() as connection: total = int( connection.execute( f"SELECT COUNT(*) AS count FROM tickets{where}", @@ -647,7 +670,7 @@ def payload_page_metrics( conditions.append(f"{key}=?") parameters.append(str(getattr(value, "value", value))) where = f" WHERE {' AND '.join(conditions)}" if conditions else "" - with self._connect() as connection: + with self._connection() as connection: total = int( connection.execute( f"SELECT COUNT(*) AS count FROM tickets{where}", @@ -683,7 +706,7 @@ def status(self, signature: tuple | None = None) -> dict[str, Any]: "path": str(self.path), } try: - with self._connect() as connection: + with self._connection() as connection: count = int( connection.execute("SELECT COUNT(*) AS count FROM tickets").fetchone()[ "count" diff --git a/planfile/core/store.py b/planfile/core/store.py index 403217e..820b2d5 100644 --- a/planfile/core/store.py +++ b/planfile/core/store.py @@ -2046,17 +2046,9 @@ def _finish_index_mutation( # All callers have already committed authoritative files. Only # BUSY/LOCKED in this disposable projection may be deferred; other # errors and failures before the source write still propagate. - code = getattr(exc, "sqlite_errorcode", None) - if code is not None: - contention = code & 0xFF in (5, 6) # SQLITE_BUSY / SQLITE_LOCKED - else: - # Python 3.10 lacks sqlite_errorcode. Match SQLite's own - # messages exactly (or its table/schema-qualified variants). - message = str(exc).lower() - contention = message in { - "database is locked", "database table is locked", "database schema is locked" - } or message.startswith(("database table is locked: ", "database schema is locked: ")) - if not contention: + from planfile.core.sqlite_index import _is_sqlite_contention + + if not _is_sqlite_contention(exc): raise from planfile.core.fastio import _atomic_write_text @@ -2119,6 +2111,11 @@ def _rebuild_ticket_index_unlocked(self, index, *, force: bool) -> dict: if not isinstance(exc, sqlite3.DatabaseError): raise + from planfile.core.sqlite_index import _is_sqlite_contention + + if _is_sqlite_contention(exc): + self._ticket_index_rebuild_deferred_until = time.monotonic() + 5.0 + raise TicketIndexContentionError("ticket_index_rebuild_busy") from exc index.reset() count = index.rebuild(self._ticket_index_records(), signature_before) signature_after = self._ticket_index_signature() diff --git a/project/TICKETS.md b/project/TICKETS.md index fc877f9..b0f5448 100644 --- a/project/TICKETS.md +++ b/project/TICKETS.md @@ -161,4 +161,5 @@ This file indexes governance tickets without taking ownership of | **ticket-257** | [`README.md`](./ticket-257/README.md) | - | - | - | - | - | | **ticket-258** | [`README.md`](./ticket-258/README.md) | - | - | - | - | - | | **ticket-259** | [`README.md`](./ticket-259/README.md) | - | - | - | - | - | +| **ticket-260** | [`README.md`](./ticket-260/README.md) | - | - | - | - | - | diff --git a/project/ticket-260/README.md b/project/ticket-260/README.md new file mode 100644 index 0000000..4cb9017 --- /dev/null +++ b/project/ticket-260/README.md @@ -0,0 +1,21 @@ +# Ticket 260: Close SQLite index connections and preserve busy rebuild owners + +- **ID**: ticket-260 +- **Owner**: agent:codex +- **Status**: IN_PROGRESS +- **Workflow state**: PUBLICATION +- **Created**: 2026-10-30 + +## Goal and scope + +SESSION_EXECUTION_AUTHORIZATION: continue the user's autonomous repair, testing and independent publication work. Existing owner semcod/planfile#313, parent PLF-20347 / Willman430. Prior isolated diagnostics proved100status reads retained201additional descriptors while cyclic GC was disabled, and a forced rebuild replaced the index inode while an independent SQLite writer was alive. Fix the component lifetime and contention classification, preserving real corruption recovery and durable authoritative state. Current internal health/summary reads work while archived operational reads exceed deadlines; source changes alone do not prove native service recovery. + +## Acceptance criteria + +- [x] AC-01: Internal index operations close connections deterministically, including transaction and setup failures, without requiring cyclic GC. +- [x] AC-02: BUSY/LOCKED rebuilds preserve the active database and owner; independent-process and corruption-recovery regressions pass. +- [ ] AC-03: Affected checks, protected exact-head CI and independent publication pass; an installed immutable candidate has separate canary evidence and no implicit live rollout. + +## Delivery boundary + +Never remove a live lock/index, kill a writer, forge a historical publication receipt, hotpatch the live package or equate source tests with deployment. Preserve existing313/316/317 and Validator593 evidence. Allocation uses the installed ID-only allocator; delivery uses the adopted v5 planner and canonical relative worktree. diff --git a/project/ticket-260/intent.json b/project/ticket-260/intent.json new file mode 100644 index 0000000..da95ff6 --- /dev/null +++ b/project/ticket-260/intent.json @@ -0,0 +1,92 @@ +{ + "schema": "new-project.intent/v3", + "ticket": "ticket-260", + "summary": "Close SQLite index connections and preserve busy rebuild owners", + "workstream": "application", + "classification": { + "kind": "BUG", + "priority": "P1", + "origin": "health" + }, + "allowedPaths": [ + "project/ticket-260/**", + "TODO.md", + "project/TICKETS.md", + "planfile/core/sqlite_index.py", + "planfile/core/store.py", + "tests/test_sqlite_ticket_index.py" + ], + "forbiddenPaths": [ + "project/ticket-*/user-*.md" + ], + "stacks": [], + "dependsOn": [], + "conflictsWith": [], + "integrationTicket": null, + "delivery": { + "acceptedBaseSha": "9a221c92dc42a104c2347ad5b507128ded696648", + "targetBranch": "main", + "outcome": "Close every internal SQLite index connection deterministically on success, failure and connection setup exceptions; treat SQLite BUSY/LOCKED rebuild failures as bounded contention without resetting another writer's index, while retaining genuine corruption recovery.", + "nonGoals": [ + "No live database/lock deletion, live package hotpatch, owner termination, lease takeover or automatic reservation release.", + "No claim that a source fix or isolated canary is native runtime activation; prior publication exceptions remain separate." + ], + "complexity": "S", + "estimatedMinutes": 45, + "budgets": { + "maxImplementationFiles": 3, + "maxAffectedComponents": 1, + "maxPublicInterfaceChanges": 0, + "maxRuntimeDependencies": 0 + }, + "architecture": { + "status": "accepted", + "decision": "Give the existing SQLiteTicketIndex internal operation context deterministic transaction exit and close, close failed setup connections, and share exact SQLite contention classification with the store rebuild/post-commit paths.", + "components": [ + { + "name": "sqlite-index-projection", + "paths": [ + "planfile/core/sqlite_index.py", + "planfile/core/store.py", + "tests/test_sqlite_ticket_index.py" + ] + } + ], + "responsibilityChanges": false, + "interfaceChanges": [], + "dataChanges": [], + "ui": { + "impact": "none", + "states": [], + "evidence": [] + }, + "rollback": "Revert the source commit; native artifacts/activation require separate exact-target operator receipts." + }, + "runtimeDependencies": [], + "validation": [ + { + "criterion": "AC-01", + "commands": [ + "testless scan -s planfile tests/test_sqlite_ticket_index.py" + ], + "evidence": "Repeated reads release connections without cyclic GC; successful transaction/rollback/setup failures release resources." + }, + { + "criterion": "AC-02", + "commands": [ + "testless scan -s planfile tests/test_sqlite_ticket_index.py" + ], + "evidence": "Independent SQLite writer retains PID/inode/data on BUSY/LOCKED; extended SQLite codes and Python3.10 fallback handled; real corruption recovery retained." + }, + { + "criterion": "AC-03", + "commands": [ + "project/governance-check.sh", + "protected OneDev exact-head/current-base verification and independent Validator publication", + "isolated installed-wheel canary on live native base image without live volume/network" + ], + "evidence": "Genuine source publication and content-bound runtime candidate; native rollout remains separate." + } + ] + } +} diff --git a/tests/test_sqlite_ticket_index.py b/tests/test_sqlite_ticket_index.py index a17f1ec..cb8ed94 100644 --- a/tests/test_sqlite_ticket_index.py +++ b/tests/test_sqlite_ticket_index.py @@ -2,6 +2,7 @@ import gc import json +import os import selectors import sqlite3 import subprocess @@ -31,6 +32,137 @@ def _disable_archive(pf: Planfile) -> None: pf.store._write_config(config) +def test_index_reads_release_descriptors_without_cyclic_gc(tmp_path): + if not Path("/proc/self/fd").is_dir(): + pytest.skip("Descriptor observation requires procfs") + index = SQLiteTicketIndex(tmp_path / "index.sqlite") + signature = ("source", 1) + index.rebuild([], signature) + gc.collect() + was_enabled = gc.isenabled() + gc.disable() + try: + before = len(os.listdir("/proc/self/fd")) + for _ in range(100): + assert index.status(signature)["current"] + assert index.is_current(signature) + assert len(os.listdir("/proc/self/fd")) <= before + 2 + finally: + if was_enabled: + gc.enable() + + +@pytest.mark.parametrize("fail", [False, True]) +def test_index_operation_commits_or_rolls_back_then_closes(tmp_path, fail): + index = SQLiteTicketIndex(tmp_path / "index.sqlite") + index.rebuild([], ("source", 1)) + connection = None + try: + with index._connection() as connection: + connection.execute("INSERT INTO meta VALUES('lifetime-test', 'present')") + if fail: + raise RuntimeError("abort transaction") + except RuntimeError: + assert fail + with pytest.raises(sqlite3.ProgrammingError, match="closed"): + connection.execute("SELECT 1") + with index._connection() as observer: + row = observer.execute("SELECT value FROM meta WHERE key='lifetime-test'").fetchone() + assert (row is None) == fail + + +@pytest.mark.parametrize("hook", ["_initialize", "_ensure_routes"]) +def test_failed_index_setup_closes_its_connection(tmp_path, monkeypatch, hook): + connect = sqlite3.connect + opened = [] + + def capture(*args, **kwargs): + connection = connect(*args, **kwargs) + opened.append(connection) + return connection + + def fail(*args): + raise RuntimeError("setup refused") + + monkeypatch.setattr(sqlite3, "connect", capture) + monkeypatch.setattr(SQLiteTicketIndex, hook, staticmethod(fail)) + with pytest.raises(RuntimeError, match="setup refused"): + SQLiteTicketIndex(tmp_path / "index.sqlite")._connect() + assert len(opened) == 1 + with pytest.raises(sqlite3.ProgrammingError, match="closed"): + opened[0].execute("SELECT 1") + + +def test_busy_rebuild_preserves_independent_writer_and_index(tmp_path, monkeypatch): + pf = Planfile(str(tmp_path)) + _disable_archive(pf) + ticket = pf.create_ticket(name="Committed before rebuild") + pf.store.configure_ticket_index(True) + index = pf.store._sqlite_ticket_index() + inode = index.path.stat().st_ino + code = ( + "import sqlite3,sys; c=sqlite3.connect(sys.argv[1]); " + "c.execute('BEGIN IMMEDIATE'); print('held',flush=True); " + "sys.stdin.readline(); c.rollback(); c.close()" + ) + holder = subprocess.Popen([sys.executable, "-u", "-c", code, str(index.path)], + stdin=subprocess.PIPE, stdout=subprocess.PIPE, + stderr=subprocess.PIPE, text=True) + connect = SQLiteTicketIndex._connect + + def short_wait(self): + connection = connect(self) + connection.execute("PRAGMA busy_timeout=10") + return connection + + try: + with selectors.DefaultSelector() as ready: + ready.register(holder.stdout, selectors.EVENT_READ) + assert ready.select(10), "Independent writer did not start" + assert holder.stdout.readline().strip() == "held" + monkeypatch.setattr(SQLiteTicketIndex, "_connect", short_wait) + with pytest.raises(TicketIndexContentionError, match="ticket_index_rebuild_busy"): + pf.store.ensure_ticket_index(force=True) + assert holder.poll() is None + assert index.path.stat().st_ino == inode + assert index.get_ticket(ticket.id)["name"] == "Committed before rebuild" + assert pf.store._locate_ticket_source(ticket.id)[1]["name"] == "Committed before rebuild" + finally: + _, error = holder.communicate("release\n", timeout=10) + assert holder.returncode == 0, error + assert pf.store.ensure_ticket_index(force=True)["current"] + + +@pytest.mark.parametrize("code,message", [ + (5, "localized"), (6, "localized"), (5 | (2 << 8), "localized"), + (6 | (1 << 8), "localized"), (None, "database is locked"), + (None, "database table is locked: tickets"), +]) +def test_busy_rebuild_never_resets_or_replays(tmp_path, monkeypatch, code, message): + pf = Planfile(str(tmp_path)) + _disable_archive(pf) + pf.create_ticket(name="Retained") + pf.store.configure_ticket_index(True) + error = sqlite3.OperationalError(message) + if code is not None: + error.sqlite_errorcode = code + attempts = [] + + def busy(*args, **kwargs): + attempts.append(True) + raise error + + def forbidden(*args): + raise AssertionError("A busy index must not be reset") + + monkeypatch.setattr(SQLiteTicketIndex, "rebuild", busy) + monkeypatch.setattr(SQLiteTicketIndex, "reset", forbidden) + with pytest.raises(TicketIndexContentionError, match="ticket_index_rebuild_busy") as raised: + pf.store.ensure_ticket_index(force=True) + assert raised.value.__cause__ is error + assert len(attempts) == 1 + + def test_sqlite_rebuild_streams_records_in_bounded_batches(tmp_path): class TrackedRecord(dict): active = 0 From af96971d5b3782e1d8e6ca77a1fb48d7afaa3d40 Mon Sep 17 00:00:00 2001 From: Tom Softreck Date: Fri, 30 Oct 2026 09:34:00 +0100 Subject: [PATCH 2/2] test(index): verify connection closure on every platform --- project/ticket-260/README.md | 2 ++ tests/test_sqlite_ticket_index.py | 23 ++++++++++++++++++----- 2 files changed, 20 insertions(+), 5 deletions(-) diff --git a/project/ticket-260/README.md b/project/ticket-260/README.md index 4cb9017..fdcb554 100644 --- a/project/ticket-260/README.md +++ b/project/ticket-260/README.md @@ -19,3 +19,5 @@ SESSION_EXECUTION_AUTHORIZATION: continue the user's autonomous repair, testing ## Delivery boundary Never remove a live lock/index, kill a writer, forge a historical publication receipt, hotpatch the live package or equate source tests with deployment. Preserve existing313/316/317 and Validator593 evidence. Allocation uses the installed ID-only allocator; delivery uses the adopted v5 planner and canonical relative worktree. + +Independent Validator review correction: the repeated-read regression now retains all 200 real SQLite connections and proves each is closed on every platform. Linux additionally measures `/proc/self/fd`; no regression is skipped. The same 78 affected tests pass with the updated test, and pinned Ruff 0.15.11 reports no findings. Source behavior and the installed artifact bytes are unchanged; fresh exact-head CI and independent approval remain required. diff --git a/tests/test_sqlite_ticket_index.py b/tests/test_sqlite_ticket_index.py index cb8ed94..dc2e919 100644 --- a/tests/test_sqlite_ticket_index.py +++ b/tests/test_sqlite_ticket_index.py @@ -32,21 +32,34 @@ def _disable_archive(pf: Planfile) -> None: pf.store._write_config(config) -def test_index_reads_release_descriptors_without_cyclic_gc(tmp_path): - if not Path("/proc/self/fd").is_dir(): - pytest.skip("Descriptor observation requires procfs") +def test_index_reads_release_descriptors_without_cyclic_gc(tmp_path, monkeypatch): index = SQLiteTicketIndex(tmp_path / "index.sqlite") signature = ("source", 1) index.rebuild([], signature) + connect = sqlite3.connect + opened = [] + + def capture(*args, **kwargs): + connection = connect(*args, **kwargs) + opened.append(connection) + return connection + + monkeypatch.setattr(sqlite3, "connect", capture) gc.collect() was_enabled = gc.isenabled() gc.disable() + descriptor_root = Path("/proc/self/fd") try: - before = len(os.listdir("/proc/self/fd")) + before = len(os.listdir(descriptor_root)) if descriptor_root.is_dir() else None for _ in range(100): assert index.status(signature)["current"] assert index.is_current(signature) - assert len(os.listdir("/proc/self/fd")) <= before + 2 + assert len(opened) == 200 + for connection in opened: + with pytest.raises(sqlite3.ProgrammingError, match="closed"): + connection.execute("SELECT 1") + if before is not None: + assert len(os.listdir(descriptor_root)) <= before + 2 finally: if was_enabled: gc.enable()