From 36f3d71b79212dc490514a3a7d08396e042bcd53 Mon Sep 17 00:00:00 2001 From: Tom Softreck Date: Thu, 29 Oct 2026 19:19:03 +0100 Subject: [PATCH] fix: preserve committed writes during index contention (ticket-258) --- planfile/core/store.py | 52 ++++- project/TICKETS.md | 1 + project/ticket-258/README.md | 21 ++ project/ticket-258/intent.json | 96 ++++++++++ tests/test_sqlite_ticket_index.py | 309 ++++++++++++++++++++++++++++++ 5 files changed, 474 insertions(+), 5 deletions(-) create mode 100644 project/ticket-258/README.md create mode 100644 project/ticket-258/intent.json diff --git a/planfile/core/store.py b/planfile/core/store.py index 6def6b9..403217e 100644 --- a/planfile/core/store.py +++ b/planfile/core/store.py @@ -1,10 +1,12 @@ from __future__ import annotations import json +import logging import math import os import re import shutil +import sqlite3 import tempfile import time from contextlib import contextmanager @@ -117,6 +119,7 @@ def _bind_base_dir(self, base_dir: Path) -> None: self._forensic_log_receipt_path = self.base_dir / "events" / ".logs.dsl.v1" self._evidence_dir = self.base_dir / "evidence" self._ticket_index_path = self.base_dir / "index" / "tickets.sqlite3" + self._ticket_index_invalidation_path = self.base_dir / "index" / "tickets.invalidated" self._ticket_index_rebuild_lock_path = self.base_dir / "index" / ".rebuild.lock" self._history_locations_path = self.base_dir / "index" / "history-locations.yaml" self._operational_receipt_path = self.base_dir / OPERATIONAL_RECEIPT_FILENAME @@ -1854,8 +1857,16 @@ def _ticket_index_signature(self) -> tuple: self.sprint_signature("all"), self._evidence_revision(), self._history_locations_signature(), + self._ticket_index_invalidation_revision(), ) + def _ticket_index_invalidation_revision(self) -> str | None: + """Read the cross-process token without opening the contended database.""" + try: + return self._ticket_index_invalidation_path.read_text(encoding="ascii") + except FileNotFoundError: + return None + @staticmethod def _ticket_index_signature_cache_seconds() -> float: """Bound source-signature reuse during a concurrent API read burst.""" @@ -1901,6 +1912,7 @@ def _ticket_index_signature_cache_probe(self) -> tuple: self.sprint_signature("current"), evidence_dir_mtime, self._history_locations_signature(), + self._ticket_index_invalidation_revision(), ) def _invalidate_ticket_index_signature_cache(self) -> None: @@ -2024,11 +2036,41 @@ def _finish_index_mutation( self._ticket_index_record(ticket, sprint) for ticket, sprint in (upserts or []) ] - self._sqlite_ticket_index().apply( - upserts=records, - deletes=deletes or [], - signature=self._ticket_index_signature(), - ) + try: + self._sqlite_ticket_index().apply( + upserts=records, + deletes=deletes or [], + signature=self._ticket_index_signature(), + ) + except sqlite3.OperationalError as exc: + # 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: + raise + from planfile.core.fastio import _atomic_write_text + + # A process-local cache reset alone leaves another worker's cached + # signature looking current until its TTL expires. This atomic + # token invalidates every worker, even for historical/evidence + # edits that do not change the cheap current-sprint probe. Never + # advance SQLite metadata or delete another writer's lock. + _atomic_write_text(self._ticket_index_invalidation_path, os.urandom(16).hex()) + self._invalidate_ticket_index_signature_cache() + logging.getLogger(__name__).warning( + "ticket_index_update_deferred: durable mutation committed; " + "SQLite projection stale (%s)", exc, + ) def ensure_ticket_index(self, *, force: bool = False) -> dict: """Ensure the disposable SQLite projection matches durable sources.""" diff --git a/project/TICKETS.md b/project/TICKETS.md index 7e5f668..fc877f9 100644 --- a/project/TICKETS.md +++ b/project/TICKETS.md @@ -159,5 +159,6 @@ This file indexes governance tickets without taking ownership of | **ticket-254** | [`README.md`](./ticket-254/README.md) | - | - | - | - | - | | **ticket-256** | [`README.md`](./ticket-256/README.md) | - | - | - | - | - | | **ticket-257** | [`README.md`](./ticket-257/README.md) | - | - | - | - | - | +| **ticket-258** | [`README.md`](./ticket-258/README.md) | - | - | - | - | - | | **ticket-259** | [`README.md`](./ticket-259/README.md) | - | - | - | - | - | diff --git a/project/ticket-258/README.md b/project/ticket-258/README.md new file mode 100644 index 0000000..a29c261 --- /dev/null +++ b/project/ticket-258/README.md @@ -0,0 +1,21 @@ +# Ticket 258: Avoid ambiguous post-commit SQLite index contention + +- **ID**: ticket-258 +- **Owner**: agent:codex +- **Status**: IN_PROGRESS +- **Workflow state**: PUBLICATION +- **Created**: 2026-10-29 + +## Goal and scope + +SESSION_EXECUTION_AUTHORIZATION: user requested continued autonomous repair, tests and protected publication ("kontynuuj, scalaj, testuj"). Continue parent PLF-20347 / Willman PLF-20340 and existing Planfile PLF-20365 / https://github.com/semcod/planfile/issues/316. + +A durable ticket write must not report HTTP500 solely because the disposable SQLite index is busy/locked afterward. Record an atomic invalidation token visible to other workers, retain typed stale reads and normal rebuild recovery. Preserve CAS, source serialization and idempotent evidence. Propagate non-contention and pre-write errors. Existing issue313 owns rebuild liveness; issue317 owns qualified native runtime adoption. Preserve foreign governance worktree255. + +## Acceptance criteria + +- [x] AC-01: Real SQLite contention after authoritative upsert/delete/evidence write acknowledges success, another process rejects stale projection and a later rebuild recovers. +- [x] AC-02: Evidence replay, stale CAS, pre-write errors and unrelated SQLite errors retain their semantics. +- [ ] AC-03: Scoped checks and governance pass; protected current-base/head OneDev and independent publication are verified. + +Validation: 142 scoped tests passed with Testless outcome accounting and coverage, including28 new regression cases. Ruff and managed governance passed. Real baseline/candidate canary: HTTP500 versus200 with the same committed value. Independent publication remains pending. diff --git a/project/ticket-258/intent.json b/project/ticket-258/intent.json new file mode 100644 index 0000000..48bf798 --- /dev/null +++ b/project/ticket-258/intent.json @@ -0,0 +1,96 @@ +{ + "schema": "new-project.intent/v3", + "ticket": "ticket-258", + "summary": "Avoid ambiguous post-commit SQLite index contention (issue316)", + "workstream": "application", + "classification": { + "kind": "BUG", + "priority": "P1", + "origin": "health" + }, + "allowedPaths": [ + "project/ticket-258/**", + "TODO.md", + "project/TICKETS.md", + "planfile/core/store.py", + "tests/test_sqlite_ticket_index.py" + ], + "forbiddenPaths": [ + "project/ticket-*/user-*.md" + ], + "stacks": [ + "python" + ], + "dependsOn": [], + "conflictsWith": [], + "integrationTicket": null, + "delivery": { + "acceptedBaseSha": "ee8d113322ffa8cf58f96cd676aec14c38258cdc", + "targetBranch": "main", + "outcome": "A durable ticket mutation succeeds despite SQLite BUSY/LOCKED during projection update; a durable invalidation token prevents stale caches being current in another process. Preserve CAS and idempotency and propagate unrelated failures.", + "nonGoals": [ + "No pre-write lock retry, live lock deletion, runtime hotpatch, managed adoption changes, or other retained worktree takeover.", + "Index rebuild liveness and native deployment remain with existing owner issues313/317." + ], + "complexity": "S", + "estimatedMinutes": 60, + "budgets": { + "maxImplementationFiles": 2, + "maxAffectedComponents": 1, + "maxPublicInterfaceChanges": 0, + "maxRuntimeDependencies": 0 + }, + "architecture": { + "status": "accepted", + "decision": "Handle contention at the existing post-commit projection boundary. Use a SQLite-independent atomic invalidation token in full and cached-signature probes; the normal rebuild reconciles it.", + "components": [ + { + "name": "ticket-store-index", + "paths": [ + "planfile/core/store.py", + "tests/test_sqlite_ticket_index.py" + ] + } + ], + "responsibilityChanges": false, + "interfaceChanges": [], + "dataChanges": [ + { + "kind": "component-local-state", + "component": "ticket-store-index", + "description": "Disposable index signature includes an atomic invalidation token; authoritative ticket format remains compatible." + } + ], + "ui": { + "impact": "none", + "states": [], + "evidence": [] + }, + "rollback": "Revert the source slice through protected publication; index is disposable and rebuilt from authoritative tickets. Reinstate the prior immutable runtime if a qualified rollout fails." + }, + "runtimeDependencies": [], + "validation": [ + { + "criterion": "AC-01", + "commands": [ + "python -m pytest -q tests/test_sqlite_ticket_index.py" + ], + "evidence": "Real SQLite writer contention, API acknowledgement, cross-process stale guard, eventual rebuild and upsert/delete paths." + }, + { + "criterion": "AC-02", + "commands": [ + "python -m pytest -q tests/test_store_lock_timeout.py tests/test_store_concurrency.py tests/test_ticket_api_events.py" + ], + "evidence": "Pre-write failures, unrelated index errors, CAS and evidence replay remain correct." + }, + { + "criterion": "AC-03", + "commands": [ + "./project/governance-check.sh" + ], + "evidence": "Scoped Testless, protected exact-head/current-base OneDev and independent Validator publication; runtime deployment is a separate receipt." + } + ] + } +} diff --git a/tests/test_sqlite_ticket_index.py b/tests/test_sqlite_ticket_index.py index 27744a8..a17f1ec 100644 --- a/tests/test_sqlite_ticket_index.py +++ b/tests/test_sqlite_ticket_index.py @@ -2,11 +2,17 @@ import gc import json +import selectors import sqlite3 +import subprocess +import sys import threading import time from concurrent.futures import ThreadPoolExecutor +from contextlib import contextmanager +from pathlib import Path +import pytest import yaml from fastapi.testclient import TestClient from typer.testing import CliRunner @@ -16,6 +22,7 @@ from planfile.cli.commands import app from planfile.core.fastio import mirror_path from planfile.core.sqlite_index import SQLiteTicketIndex +from planfile.core.store import TicketIndexContentionError, TicketUpdatedAtConflictError def _disable_archive(pf: Planfile) -> None: @@ -1102,3 +1109,305 @@ def test_conflicting_issue_identity_is_not_routed(tmp_path): assert index.lookup_issue("semcod/planfile", 244) == [] assert index.lookup_issue("other/repository", 244) == [] assert index.get_ticket("PLF-1") == payload + + +@contextmanager +def _held_index_writer(pf, monkeypatch): + """Hold a real WAL writer while keeping test contention waits bounded.""" + original_connect = SQLiteTicketIndex._connect + + def short_wait(index): + connection = original_connect(index) + connection.execute("PRAGMA busy_timeout=10") + return connection + + with monkeypatch.context() as patch: + patch.setattr(SQLiteTicketIndex, "_connect", short_wait) + connection = sqlite3.connect(pf.store._ticket_index_path) + connection.execute("BEGIN IMMEDIATE") + try: + yield + finally: + connection.rollback() + connection.close() + + +def _evidence(pf, ticket_id, key="receipt-1"): + return pf.append_ticket_evidence( + ticket_id, idempotency_key=key, collection="receipts", evidence={"verified": True}, + actor="bot:test", reason="Observed external result", + ) + + +@pytest.mark.parametrize("sharded", [False, True]) +@pytest.mark.parametrize("operation", ["create", "bulk-create", "update", "delete", "bulk-delete", "move", "evidence"]) +def test_committed_mutation_survives_real_index_writer_contention( + tmp_path, monkeypatch, caplog, sharded, operation, +): + pf = Planfile(str(tmp_path)) + _disable_archive(pf) + ticket = pf.create_ticket(name="Before") + if sharded: + pf.store.migrate_to_sharded_yaml(shard_size=5) + pf.store.configure_ticket_index(True) + before = pf.store._ticket_index_signature() + + with _held_index_writer(pf, monkeypatch): + if operation == "create": + ticket = pf.create_ticket(name="After") + elif operation == "bulk-create": + ticket = pf.create_tickets_bulk([{"name": "After"}])[0] + elif operation == "update": + assert pf.update_ticket(ticket.id, name="After").name == "After" + elif operation == "delete": + assert pf.store.delete_ticket(ticket.id) + elif operation == "bulk-delete": + assert pf.delete_tickets([ticket.id]) == ([ticket.id], []) + elif operation == "move": + assert pf.store.move_ticket(ticket.id, "backlog") + else: + updated, recorded = _evidence(pf, ticket.id) + assert recorded and updated.outputs.result["receipts"][0]["verified"] + assert "ticket_index_update_deferred" in caplog.text + assert not pf.store._sqlite_ticket_index().is_current(pf.store._ticket_index_signature()) + assert pf.store._ticket_index_signature() != before + with pytest.raises(TicketIndexContentionError, match="ticket_index_stale"): + pf.store.require_current_ticket_index() + # An exact read must observe authoritative files, without repairing + # the locked disposable index or claiming its snapshot is current. + current = pf.get_ticket(ticket.id, repair_index=False) + if operation in ("delete", "bulk-delete"): + assert current is None + elif operation == "move": + assert current.sprint == "backlog" + elif operation == "evidence": + assert current.outputs.result["receipts"][0]["verified"] + else: + assert current.name == "After" + + assert pf.store.ensure_ticket_index()["current"] is True + assert pf.get_ticket(ticket.id, repair_index=False) == current + # A later incremental edit must not bless a projection missing the first. + pf.create_ticket(name="Unrelated later mutation") + assert pf.get_ticket(ticket.id, repair_index=False) == current + + +@pytest.mark.parametrize("operation", ["update", "evidence", "input-required"]) +def test_api_acknowledges_committed_write_during_index_contention( + tmp_path, monkeypatch, operation, +): + pf = Planfile(str(tmp_path)) + _disable_archive(pf) + ticket = pf.create_ticket(name="Before") + pf.store.configure_ticket_index(True) + monkeypatch.setattr(server, "get_planfile", lambda: pf) + client = TestClient(server.app) + with _held_index_writer(pf, monkeypatch): + if operation == "update": + response = client.post(f"/tickets/{ticket.id}/update-if-current", json={ + "name": "After", "expected_updated_at": str(ticket.updated_at), + }) + elif operation == "evidence": + payload = {"idempotency_key": "api-receipt", "collection": "receipts", + "evidence": {"verified": True}, "actor": "bot:test", "reason": "Observed"} + response = client.post(f"/tickets/{ticket.id}/evidence", json=payload) + replay = client.post(f"/tickets/{ticket.id}/evidence", json=payload) + assert replay.status_code == 200 and replay.json()["deduplicated"] + payload["evidence"] = {"verified": False} + assert client.post(f"/tickets/{ticket.id}/evidence", json=payload).status_code == 409 + else: + response = client.post(f"/tickets/{ticket.id}/input-required", json={ + "prompt": "Which runtime is admitted?", "actor": "bot:test", "reason": "Missing binding", + }) + assert response.status_code == 200, response.text + current = pf.get_ticket(ticket.id, repair_index=False) + if operation == "update": + assert current.name == "After" + # A committed write's old CAS must not be replayed as fresh work. + assert client.post(f"/tickets/{ticket.id}/update-if-current", json={ + "name": "Replay", "expected_updated_at": str(ticket.updated_at), + }).status_code == 409 + elif operation == "evidence": + assert response.json()["recorded"] + assert len(current.outputs.result["receipts"]) == 1 + else: + assert current.execution.state == "waiting_input" + + +@pytest.mark.parametrize("edit", ["history", "existing-evidence"]) +def test_failed_index_update_invalidates_another_process_cached_signature( + tmp_path, monkeypatch, edit, +): + pf = Planfile(str(tmp_path)) + _disable_archive(pf) + ticket = pf.create_ticket(name="Before", sprint="backlog") + _evidence(pf, ticket.id, "existing") + pf.store.configure_ticket_index(True) + code = ''' +import json, sys +from pathlib import Path +import planfile +from planfile import Planfile +from planfile.core.store import TicketIndexContentionError +assert Path(planfile.__file__).resolve() == Path(sys.argv[3]).resolve() +store = Planfile(sys.argv[1]).store +store._ticket_index_signature_cache_seconds = lambda: 3600 +store.require_current_ticket_index() +print("ready", flush=True) +sys.stdin.readline() +try: + store.require_current_ticket_index() +except TicketIndexContentionError as exc: + print(json.dumps({"state": str(exc), "name": store.get_ticket(sys.argv[2], repair_index=False).name}), flush=True) +else: + raise AssertionError("another process served stale projection as current") +''' + source_root = Path(__file__).resolve().parents[1] + process = subprocess.Popen([sys.executable, "-u", "-c", code, str(tmp_path), ticket.id, + str(source_root / "planfile/__init__.py")], + cwd=source_root, + stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, + text=True) + try: + with selectors.DefaultSelector() as ready: + ready.register(process.stdout, selectors.EVENT_READ) + assert ready.select(timeout=30), "reader failed to warm cache" + assert process.stdout.readline().strip() == "ready", process.stderr.read() + probe_before = pf.store._ticket_index_signature_cache_probe() + with _held_index_writer(pf, monkeypatch): + if edit == "history": + pf.update_ticket(ticket.id, name="After") + else: + _evidence(pf, ticket.id, "second") + probe_after = pf.store._ticket_index_signature_cache_probe() + # Neither edit changes the old cheap probe; only the new token + # prevents a worker with a warmed signature from returning stale. + assert probe_before[:-1] == probe_after[:-1] + assert probe_before[-1] != probe_after[-1] + output, error = process.communicate("check\n", timeout=15) + assert process.returncode == 0, error + assert json.loads(output)["state"] == "ticket_index_stale" + finally: + if process.poll() is None: + process.kill() + process.communicate(timeout=15) + assert pf.store.ensure_ticket_index()["current"] is True + + +@pytest.mark.parametrize("message", ["disk I/O error", "no such table: tickets", "database disk image is malformed"]) +def test_non_contention_index_errors_still_propagate(tmp_path, monkeypatch, message): + pf = Planfile(str(tmp_path)) + _disable_archive(pf) + ticket = pf.create_ticket(name="Before") + pf.store.configure_ticket_index(True) + + def broken_apply(*args, **kwargs): + raise sqlite3.OperationalError(message) + + monkeypatch.setattr(SQLiteTicketIndex, "apply", broken_apply) + with pytest.raises(sqlite3.OperationalError, match=message): + pf.update_ticket(ticket.id, name="After") + assert pf.store._locate_ticket_source(ticket.id)[1]["name"] == "After" + assert not pf.store._ticket_index_invalidation_path.exists() + + +def test_pre_write_sqlite_failure_preserves_authoritative_ticket(tmp_path, monkeypatch): + pf = Planfile(str(tmp_path)) + _disable_archive(pf) + ticket = pf.create_ticket(name="Before") + pf.store.configure_ticket_index(True) + + def unavailable_source(*args, **kwargs): + raise sqlite3.OperationalError("database is locked") + + monkeypatch.setattr(pf.store, "_write_yaml_atomic", unavailable_source) + with pytest.raises(sqlite3.OperationalError, match="database is locked"): + pf.update_ticket(ticket.id, name="After") + assert pf.store._locate_ticket_source(ticket.id)[1]["name"] == "Before" + assert pf.store.require_current_ticket_index()["current"] is True + assert not pf.store._ticket_index_invalidation_path.exists() + with pytest.raises(TicketUpdatedAtConflictError): + pf.update_ticket(ticket.id, name="Stale", expected_updated_at="2000-01-01T00:00:00Z") + + +def test_actual_sqlite_locked_error_defers_only_post_commit_index(tmp_path, monkeypatch): + # SQLITE_LOCKED differs from SQLITE_BUSY: produce it with an unfinished + # statement in SQLite itself, rather than relying on exception text. + connection = sqlite3.connect(":memory:") + connection.execute("CREATE TABLE held(value)") + connection.execute("INSERT INTO held VALUES(1)") + cursor = connection.execute("SELECT * FROM held") + with pytest.raises(sqlite3.OperationalError) as captured: + connection.execute("DROP TABLE held") + cursor.close() + connection.close() + error = captured.value + if hasattr(error, "sqlite_errorcode"): + assert error.sqlite_errorcode & 0xFF == 6 + pf = Planfile(str(tmp_path)) + _disable_archive(pf) + ticket = pf.create_ticket(name="Before") + pf.store.configure_ticket_index(True) + + def locked_apply(*args, **kwargs): + raise error + + monkeypatch.setattr(SQLiteTicketIndex, "apply", locked_apply) + assert pf.update_ticket(ticket.id, name="After").name == "After" + assert pf.get_ticket(ticket.id, repair_index=False).name == "After" + assert pf.store.ensure_ticket_index()["current"] is True + + +@pytest.mark.parametrize("code", [5 | (2 << 8), 6 | (1 << 8)]) +def test_extended_contention_codes_are_classified_by_primary_code(tmp_path, monkeypatch, code): + pf = Planfile(str(tmp_path)) + _disable_archive(pf) + ticket = pf.create_ticket(name="Before") + pf.store.configure_ticket_index(True) + error = sqlite3.OperationalError("localized or version-specific description") + error.sqlite_errorcode = code + + def fail_apply(*args, **kwargs): + raise error + + monkeypatch.setattr(SQLiteTicketIndex, "apply", fail_apply) + assert pf.update_ticket(ticket.id, name="After").name == "After" + assert pf.store._ticket_index_invalidation_path.exists() + + +def test_non_contention_error_code_overrides_lock_like_message(tmp_path, monkeypatch): + pf = Planfile(str(tmp_path)) + _disable_archive(pf) + ticket = pf.create_ticket(name="Before") + pf.store.configure_ticket_index(True) + error = sqlite3.OperationalError("database is locked") + error.sqlite_errorcode = 10 # SQLITE_IOERR must never be suppressed. + + def fail_apply(*args, **kwargs): + raise error + + monkeypatch.setattr(SQLiteTicketIndex, "apply", fail_apply) + with pytest.raises(sqlite3.OperationalError): + pf.update_ticket(ticket.id, name="After") + assert not pf.store._ticket_index_invalidation_path.exists() + + +def test_failure_to_persist_invalidation_is_not_hidden(tmp_path, monkeypatch): + from planfile.core import fastio + + pf = Planfile(str(tmp_path)) + _disable_archive(pf) + ticket = pf.create_ticket(name="Before") + pf.store.configure_ticket_index(True) + original = fastio._atomic_write_text + + def no_marker(path, content): + if path == pf.store._ticket_index_invalidation_path: + raise OSError("invalidation filesystem unavailable") + return original(path, content) + + monkeypatch.setattr(fastio, "_atomic_write_text", no_marker) + with _held_index_writer(pf, monkeypatch): + with pytest.raises(OSError, match="invalidation filesystem unavailable"): + pf.update_ticket(ticket.id, name="After") + assert pf.store._locate_ticket_source(ticket.id)[1]["name"] == "After"