From ddb071ec616715c96606cde38bcb8c372f858465 Mon Sep 17 00:00:00 2001 From: Skylar Payne <4830598+skylarbpayne@users.noreply.github.com> Date: Mon, 13 Jul 2026 17:24:04 -0700 Subject: [PATCH 1/4] HW-13: make SQLite contention policy explicit --- docs/operations/sqlite-policy.md | 84 +++ src/hermes_workflows/sqlite_policy.py | 714 +++++++++++++++++++++++++ tests/fixtures/sqlite_policy_v1.json | 61 +++ tests/probes/sqlite_lock_contention.py | 150 ++++++ tests/test_sqlite_policy.py | 344 ++++++++++++ 5 files changed, 1353 insertions(+) create mode 100644 docs/operations/sqlite-policy.md create mode 100644 src/hermes_workflows/sqlite_policy.py create mode 100644 tests/fixtures/sqlite_policy_v1.json create mode 100644 tests/probes/sqlite_lock_contention.py create mode 100644 tests/test_sqlite_policy.py diff --git a/docs/operations/sqlite-policy.md b/docs/operations/sqlite-policy.md new file mode 100644 index 0000000..521bae1 --- /dev/null +++ b/docs/operations/sqlite-policy.md @@ -0,0 +1,84 @@ +--- +layout: page +title: SQLite contention policy +--- + +# SQLite contention policy + +Hermes Workflows supports SQLite on a local filesystem with WAL. The policy is explicit in `hermes_workflows.sqlite_policy.SQLitePolicyV1`; integration code must not create an independent timeout, PRAGMA, lock classifier, or retry loop. + +## Contract + +Every writable connection must apply and verify: + +- `PRAGMA journal_mode=WAL` +- `PRAGMA foreign_keys=ON` +- `PRAGMA busy_timeout=` + +`apply_writable_pragmas()` is idempotent after WAL is active. It refuses to switch an existing database from another journal mode unless the caller passes the explicit offline migration gate. A refusal is `JournalModeChangeRequired`, not an invitation to change the mode while workers are running. + +`SQLitePolicyV1.to_dict()` and `lease_retry_plan(...).to_dict()` are the queryable policy/status payloads. The version-1 compatibility fixture is `tests/fixtures/sqlite_policy_v1.json`. + +## Lock classification and retry + +Only `sqlite3.OperationalError` values identified as SQLite `BUSY`/`LOCKED`, or SQLite's known `database ... is locked` messages, are retryable. Integrity errors, full disks, missing tables, I/O errors, malformed SQL, and every unknown database error escape unchanged. Do not wrap broad database failures in this retry helper. + +`run_with_lock_retry()` emits one durable diagnostic per lock decision through a supplied sink. `JsonlSQLiteDiagnosticSink` appends canonical JSON and calls `fsync` for every record. Integrations may provide a database/event sink, but they must preserve the version-1 fields: + +- event: `sqlite.lock_retry`, `sqlite.lock_recovered`, or `sqlite.lock_exhausted` +- operation and attempt/max-attempt counts +- lock count and classification +- busy timeout, lease, renewal interval, and lease-safe budget +- elapsed time and the selected delay when retrying + +Exhaustion raises `SQLiteLockExhausted`; it is never reported as success and never silently discarded. + +## Lease-safe bound + +Let: + +- `L` be the remaining lease duration +- `R` be the renewal interval +- `M` be the configured safety margin +- `B` be the SQLite busy timeout for each attempt +- `D_i` be each exponential delay at maximum positive jitter + +The retry budget is: + +`budget_ms = (L - R) * 1000 - M` + +The plan admits only the largest prefix of attempts for which: + +`attempts * B + sum(D_i) <= budget_ms` + +If even one busy timeout cannot fit, `lease_retry_plan()` raises `LeaseUnsafePolicy`. Runtime elapsed time is checked again before each sleep, so a slow host fails explicitly before consuming the lease window. Jitter is bounded by `retry_jitter_ratio`; it cannot expand beyond the calculated worst case. + +The contention probe holds `BEGIN IMMEDIATE` longer than one renewal interval, then verifies recovery before lease exhaustion, durable lock/recovery diagnostics, and exactly one target write: + +```console +python tests/probes/sqlite_lock_contention.py +``` + +## Storage doctor + +`doctor_sqlite_storage()` is read-only with respect to the target database. It reads the current journal mode, identifies the host filesystem, and runs WAL/write/checkpoint verification against a disposable sibling database. The sibling database and sidecars are removed after the probe. + +Known network/distributed filesystems (`nfs`, `nfs4`, `cifs`, `smbfs`, `afpfs`, `sshfs`, `fuse.sshfs`, `9p`, `davfs`, `glusterfs`, and `lustre`) fail before the WAL probe. This is intentional: SQLite WAL requires shared-memory and locking semantics that these deployments cannot safely promise. `require_supported_sqlite_storage()` converts a failed report into `UnsupportedSQLiteStorage` for a doctor command's nonzero exit path. + +An existing non-WAL database can be storage-compatible while still reporting `journal_mode_change_required=true`. That result is not healthy-for-start; execute the offline plan first. + +## Stop / checkpoint / switch / start + +Never switch journal mode while the service is active. + +1. Stop every process that can write the database. Confirm there are no active writer connections. +2. Open one dedicated offline connection. Run `PRAGMA wal_checkpoint(TRUNCATE)` and require the returned busy count to be zero. +3. On that same controlled connection, run `PRAGMA journal_mode=WAL` and verify SQLite returns `wal`. +4. Close the migration connection. Start exactly one writer first and make it open through `SQLitePolicyV1`; then start the remaining services. +5. Run storage doctor plus the contention probe. Verify WAL, foreign keys, busy timeout, and no new lock-exhaustion diagnostics before restoring traffic. + +`stop_checkpoint_start_plan(db_path)` returns the same five phases as structured data. It does not stop services or mutate a database. + +## Rollback + +Rollback may stop services and restore the previous application version, but it must keep lock diagnostics and unsupported-storage refusal. Do not switch away from WAL merely to roll application code back. If a journal change is genuinely required, use the same offline stop/checkpoint/switch/start sequence and preserve the database plus `-wal`/`-shm` files until the checkpoint is verified. diff --git a/src/hermes_workflows/sqlite_policy.py b/src/hermes_workflows/sqlite_policy.py new file mode 100644 index 0000000..b9a8c9c --- /dev/null +++ b/src/hermes_workflows/sqlite_policy.py @@ -0,0 +1,714 @@ +from __future__ import annotations + +import json +import os +import random +import sqlite3 +import subprocess +import tempfile +import threading +import time +from collections.abc import Callable, Mapping +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Optional, Protocol, Tuple, Union +from urllib.parse import quote + + +SQLITE_POLICY_SCHEMA_VERSION = 1 +SQLITE_POLICY_SERVICE_ID = "sqlite.policy" +SQLITE_POLICY_CONTRACT_VERSION = 1 +_SQLITE_BUSY_CODE = 5 +_SQLITE_LOCKED_CODE = 6 + +UNSUPPORTED_FILESYSTEMS = frozenset( + { + "9p", + "afpfs", + "cifs", + "davfs", + "fuse.sshfs", + "glusterfs", + "lustre", + "nfs", + "nfs4", + "smbfs", + "sshfs", + } +) + + +class SQLitePolicyError(RuntimeError): + """Base class for explicit SQLite policy failures.""" + + +class LeaseUnsafePolicy(SQLitePolicyError): + """The configured SQLite wait cannot fit inside the active lease window.""" + + +class WalCompatibilityError(SQLitePolicyError): + """SQLite or the backing storage refused the required WAL mode.""" + + +class JournalModeChangeRequired(SQLitePolicyError): + def __init__(self, current_mode: str, requested_mode: str, database: Union[str, Path]): + self.current_mode = current_mode + self.requested_mode = requested_mode + self.plan = stop_checkpoint_start_plan(database) + super().__init__( + f"journal mode change requires an offline stop/checkpoint/switch/start plan: " + f"{current_mode} -> {requested_mode}" + ) + + +class SQLiteLockExhausted(SQLitePolicyError): + def __init__(self, operation: str, attempts: int, budget_ms: float): + self.operation = operation + self.attempts = attempts + self.budget_ms = budget_ms + super().__init__( + f"known SQLite lock exhausted bounded retry for {operation} " + f"after {attempts} attempts within {budget_ms:g}ms lease budget" + ) + + +class UnsupportedSQLiteStorage(SQLitePolicyError): + def __init__(self, failures: Tuple[str, ...]): + self.failures = failures + super().__init__("SQLite storage doctor failed: " + ", ".join(failures)) + + +class SQLiteDiagnosticSink(Protocol): + def emit(self, record: Mapping[str, Any]) -> None: ... + + +@dataclass(frozen=True) +class SQLitePolicyV1: + schema_version: int = SQLITE_POLICY_SCHEMA_VERSION + journal_mode: str = "wal" + foreign_keys: bool = True + busy_timeout_ms: int = 250 + retry_initial_delay_ms: int = 25 + retry_multiplier: float = 2.0 + retry_max_delay_ms: int = 250 + retry_max_attempts: int = 8 + retry_jitter_ratio: float = 0.2 + lease_safety_margin_ms: int = 100 + + def __post_init__(self) -> None: + if type(self.schema_version) is not int or self.schema_version != SQLITE_POLICY_SCHEMA_VERSION: + raise ValueError("schema_version must equal 1") + if not isinstance(self.journal_mode, str) or self.journal_mode.lower() != "wal": + raise ValueError("journal_mode must equal 'wal'") + object.__setattr__(self, "journal_mode", self.journal_mode.lower()) + if type(self.foreign_keys) is not bool or not self.foreign_keys: + raise ValueError("foreign_keys must be true") + for name in ( + "busy_timeout_ms", + "retry_initial_delay_ms", + "retry_max_delay_ms", + "retry_max_attempts", + "lease_safety_margin_ms", + ): + value = getattr(self, name) + if type(value) is not int or value <= 0: + raise ValueError(f"{name} must be an integer > 0") + if type(self.retry_multiplier) not in (int, float) or self.retry_multiplier < 1.0: + raise ValueError("retry_multiplier must be a number >= 1") + object.__setattr__(self, "retry_multiplier", float(self.retry_multiplier)) + if type(self.retry_jitter_ratio) not in (int, float): + raise ValueError("retry_jitter_ratio must be a number") + if not 0.0 <= float(self.retry_jitter_ratio) <= 1.0: + raise ValueError("retry_jitter_ratio must be between 0 and 1") + object.__setattr__(self, "retry_jitter_ratio", float(self.retry_jitter_ratio)) + if self.retry_initial_delay_ms > self.retry_max_delay_ms: + raise ValueError("retry_initial_delay_ms must not exceed retry_max_delay_ms") + + def to_dict(self) -> dict[str, object]: + return { + "schema_version": self.schema_version, + "journal_mode": self.journal_mode, + "foreign_keys": self.foreign_keys, + "busy_timeout_ms": self.busy_timeout_ms, + "retry_initial_delay_ms": self.retry_initial_delay_ms, + "retry_multiplier": self.retry_multiplier, + "retry_max_delay_ms": self.retry_max_delay_ms, + "retry_max_attempts": self.retry_max_attempts, + "retry_jitter_ratio": self.retry_jitter_ratio, + "lease_safety_margin_ms": self.lease_safety_margin_ms, + } + + @classmethod + def from_dict(cls, value: Mapping[str, Any]) -> "SQLitePolicyV1": + if not isinstance(value, Mapping): + raise TypeError("SQLite policy must be a mapping") + expected = { + "schema_version", + "journal_mode", + "foreign_keys", + "busy_timeout_ms", + "retry_initial_delay_ms", + "retry_multiplier", + "retry_max_delay_ms", + "retry_max_attempts", + "retry_jitter_ratio", + "lease_safety_margin_ms", + } + if set(value) != expected: + missing = sorted(expected - set(value)) + extra = sorted(set(value) - expected) + raise ValueError(f"SQLite policy fields mismatch: missing={missing}, extra={extra}") + return cls(**dict(value)) + + +DEFAULT_SQLITE_POLICY = SQLitePolicyV1() + + +@dataclass(frozen=True) +class SQLiteRetryPlanV1: + lease_seconds: float + renewal_interval_seconds: float + budget_ms: float + max_attempts: int + max_elapsed_ms: float + delays_ms: Tuple[float, ...] + worst_case_delays_ms: Tuple[float, ...] + + def to_dict(self) -> dict[str, object]: + return { + "lease_seconds": self.lease_seconds, + "renewal_interval_seconds": self.renewal_interval_seconds, + "budget_ms": self.budget_ms, + "max_attempts": self.max_attempts, + "max_elapsed_ms": self.max_elapsed_ms, + "delays_ms": list(self.delays_ms), + "worst_case_delays_ms": list(self.worst_case_delays_ms), + } + + +@dataclass(frozen=True) +class SQLitePragmaReportV1: + journal_mode: str + foreign_keys: int + busy_timeout_ms: int + + def to_dict(self) -> dict[str, object]: + return { + "journal_mode": self.journal_mode, + "foreign_keys": self.foreign_keys, + "busy_timeout_ms": self.busy_timeout_ms, + } + + +@dataclass(frozen=True) +class SQLiteStorageDoctorReportV1: + database_path: str + filesystem_type: str + supported: bool + failures: Tuple[str, ...] + unsupported_filesystems: Tuple[str, ...] + wal_probe_performed: bool + wal_compatible: bool + current_journal_mode: Optional[str] + journal_mode_change_required: bool + plan: Tuple[dict[str, str], ...] + + def to_dict(self) -> dict[str, object]: + return { + "schema_version": SQLITE_POLICY_SCHEMA_VERSION, + "database_path": self.database_path, + "filesystem_type": self.filesystem_type, + "supported": self.supported, + "failures": list(self.failures), + "unsupported_filesystems": list(self.unsupported_filesystems), + "wal_probe_performed": self.wal_probe_performed, + "wal_compatible": self.wal_compatible, + "current_journal_mode": self.current_journal_mode, + "journal_mode_change_required": self.journal_mode_change_required, + "plan": list(self.plan), + } + + +class JsonlSQLiteDiagnosticSink: + """Append and fsync each bounded lock diagnostic as one canonical JSON line.""" + + def __init__(self, path: Union[str, Path]): + self.path = Path(path) + self._lock = threading.Lock() + + def emit(self, record: Mapping[str, Any]) -> None: + payload = json.dumps(dict(record), sort_keys=True, separators=(",", ":")) + self.path.parent.mkdir(parents=True, exist_ok=True) + with self._lock: + with self.path.open("a", encoding="utf-8") as stream: + stream.write(payload + "\n") + stream.flush() + os.fsync(stream.fileno()) + + +def lease_retry_plan( + policy: SQLitePolicyV1, + *, + lease_seconds: float, + renewal_interval_seconds: float, +) -> SQLiteRetryPlanV1: + if not isinstance(policy, SQLitePolicyV1): + raise TypeError("policy must be SQLitePolicyV1") + if type(lease_seconds) not in (int, float) or lease_seconds <= 0: + raise ValueError("lease_seconds must be a number > 0") + if type(renewal_interval_seconds) not in (int, float) or renewal_interval_seconds <= 0: + raise ValueError("renewal_interval_seconds must be a number > 0") + lease_seconds = float(lease_seconds) + renewal_interval_seconds = float(renewal_interval_seconds) + if renewal_interval_seconds >= lease_seconds: + raise LeaseUnsafePolicy("renewal interval must be shorter than the lease") + + budget_ms = _rounded( + (lease_seconds - renewal_interval_seconds) * 1000.0 - policy.lease_safety_margin_ms + ) + if budget_ms <= 0: + raise LeaseUnsafePolicy("lease safety margin leaves no SQLite retry budget") + if policy.busy_timeout_ms > budget_ms: + raise LeaseUnsafePolicy( + f"busy timeout {policy.busy_timeout_ms}ms exceeds lease-safe budget {budget_ms:g}ms" + ) + + delays: list[float] = [] + worst_delays: list[float] = [] + attempts = 0 + worst_elapsed = 0.0 + for configured_attempt in range(1, policy.retry_max_attempts + 1): + candidate = worst_elapsed + policy.busy_timeout_ms + if candidate > budget_ms: + break + worst_elapsed = candidate + attempts += 1 + if configured_attempt == policy.retry_max_attempts: + continue + delay = min( + policy.retry_initial_delay_ms * (policy.retry_multiplier ** (configured_attempt - 1)), + policy.retry_max_delay_ms, + ) + worst_delay = delay * (1.0 + policy.retry_jitter_ratio) + next_attempt_bound = worst_elapsed + worst_delay + policy.busy_timeout_ms + if next_attempt_bound > budget_ms: + break + delays.append(_rounded(delay)) + worst_delays.append(_rounded(worst_delay)) + worst_elapsed += worst_delay + + if attempts < 1: + raise LeaseUnsafePolicy("busy timeout cannot fit one attempt inside the lease-safe budget") + return SQLiteRetryPlanV1( + lease_seconds=lease_seconds, + renewal_interval_seconds=renewal_interval_seconds, + budget_ms=budget_ms, + max_attempts=attempts, + max_elapsed_ms=_rounded(worst_elapsed), + delays_ms=tuple(delays), + worst_case_delays_ms=tuple(worst_delays), + ) + + +def classify_sqlite_lock(error: BaseException) -> Optional[str]: + """Classify only SQLite BUSY/LOCKED failures; every other DB error is non-retryable.""" + + if not isinstance(error, sqlite3.OperationalError): + return None + code = getattr(error, "sqlite_errorcode", None) + if isinstance(code, int) and (code & 0xFF) in (_SQLITE_BUSY_CODE, _SQLITE_LOCKED_CODE): + return "locked" + message = str(error).strip().lower() + known_prefixes = ( + "database is locked", + "database table is locked", + "database schema is locked", + ) + if message.startswith(known_prefixes): + return "locked" + return None + + +def apply_writable_pragmas( + connection: sqlite3.Connection, + policy: SQLitePolicyV1 = DEFAULT_SQLITE_POLICY, + *, + allow_journal_mode_change: bool = False, + database: Optional[Union[str, Path]] = None, +) -> SQLitePragmaReportV1: + """Apply the canonical writable policy without silently changing journal mode.""" + + if not isinstance(connection, sqlite3.Connection): + raise TypeError("connection must be sqlite3.Connection") + if not isinstance(policy, SQLitePolicyV1): + raise TypeError("policy must be SQLitePolicyV1") + if connection.in_transaction: + raise SQLitePolicyError("writable PRAGMAs must be applied outside a transaction") + + connection.execute("PRAGMA foreign_keys = ON") + connection.execute(f"PRAGMA busy_timeout = {policy.busy_timeout_ms}") + current_mode = str(connection.execute("PRAGMA journal_mode").fetchone()[0]).lower() + target = _connection_database(connection, database) + if current_mode != policy.journal_mode: + if not allow_journal_mode_change: + raise JournalModeChangeRequired(current_mode, policy.journal_mode, target) + returned_mode = str( + connection.execute(f"PRAGMA journal_mode = {policy.journal_mode.upper()}").fetchone()[0] + ).lower() + if returned_mode != policy.journal_mode: + raise WalCompatibilityError( + f"SQLite returned journal_mode={returned_mode}; required {policy.journal_mode}" + ) + + report = SQLitePragmaReportV1( + journal_mode=str(connection.execute("PRAGMA journal_mode").fetchone()[0]).lower(), + foreign_keys=int(connection.execute("PRAGMA foreign_keys").fetchone()[0]), + busy_timeout_ms=int(connection.execute("PRAGMA busy_timeout").fetchone()[0]), + ) + if report.journal_mode != policy.journal_mode: + raise WalCompatibilityError( + f"SQLite returned journal_mode={report.journal_mode}; required {policy.journal_mode}" + ) + if report.foreign_keys != 1: + raise SQLitePolicyError("SQLite refused PRAGMA foreign_keys=ON") + if report.busy_timeout_ms != policy.busy_timeout_ms: + raise SQLitePolicyError( + f"SQLite busy_timeout mismatch: {report.busy_timeout_ms} != {policy.busy_timeout_ms}" + ) + return report + + +def run_with_lock_retry( + operation: Callable[[], Any], + *, + policy: SQLitePolicyV1 = DEFAULT_SQLITE_POLICY, + lease_seconds: float, + renewal_interval_seconds: float, + operation_name: str = "sqlite_operation", + diagnostic_sink: Union[SQLiteDiagnosticSink, Callable[[Mapping[str, Any]], None]], + sleep: Callable[[float], None] = time.sleep, + random_value: Callable[[], float] = random.random, +) -> Any: + """Retry known lock failures only, bounded by the active lease safety window.""" + + if not callable(operation): + raise TypeError("operation must be callable") + if not isinstance(operation_name, str) or not operation_name.strip(): + raise ValueError("operation_name must be a nonblank string") + plan = lease_retry_plan( + policy, + lease_seconds=lease_seconds, + renewal_interval_seconds=renewal_interval_seconds, + ) + started = time.monotonic() + lock_count = 0 + for attempt in range(1, plan.max_attempts + 1): + try: + result = operation() + except sqlite3.OperationalError as error: + classification = classify_sqlite_lock(error) + if classification is None: + raise + lock_count += 1 + elapsed_ms = (time.monotonic() - started) * 1000.0 + if attempt >= plan.max_attempts: + _emit_diagnostic( + diagnostic_sink, + _diagnostic_record( + event="sqlite.lock_exhausted", + operation=operation_name, + attempt=attempt, + max_attempts=plan.max_attempts, + lock_count=lock_count, + classification=classification, + elapsed_ms=elapsed_ms, + delay_ms=None, + policy=policy, + plan=plan, + ), + ) + raise SQLiteLockExhausted(operation_name, attempt, plan.budget_ms) from error + + base_delay_ms = plan.delays_ms[attempt - 1] + jitter_sample = float(random_value()) + if not 0.0 <= jitter_sample <= 1.0: + raise ValueError("random_value must return a number between 0 and 1") + jitter_factor = 1.0 - policy.retry_jitter_ratio + 2.0 * policy.retry_jitter_ratio * jitter_sample + delay_ms = base_delay_ms * jitter_factor + if elapsed_ms + delay_ms + policy.busy_timeout_ms > plan.budget_ms: + _emit_diagnostic( + diagnostic_sink, + _diagnostic_record( + event="sqlite.lock_exhausted", + operation=operation_name, + attempt=attempt, + max_attempts=plan.max_attempts, + lock_count=lock_count, + classification=classification, + elapsed_ms=elapsed_ms, + delay_ms=None, + policy=policy, + plan=plan, + ), + ) + raise SQLiteLockExhausted(operation_name, attempt, plan.budget_ms) from error + _emit_diagnostic( + diagnostic_sink, + _diagnostic_record( + event="sqlite.lock_retry", + operation=operation_name, + attempt=attempt, + max_attempts=plan.max_attempts, + lock_count=lock_count, + classification=classification, + elapsed_ms=elapsed_ms, + delay_ms=delay_ms, + policy=policy, + plan=plan, + ), + ) + sleep(delay_ms / 1000.0) + continue + if lock_count: + _emit_diagnostic( + diagnostic_sink, + _diagnostic_record( + event="sqlite.lock_recovered", + operation=operation_name, + attempt=attempt, + max_attempts=plan.max_attempts, + lock_count=lock_count, + classification="locked", + elapsed_ms=(time.monotonic() - started) * 1000.0, + delay_ms=None, + policy=policy, + plan=plan, + ), + ) + return result + raise AssertionError("bounded SQLite retry loop terminated without result or error") + + +def detect_filesystem_type(path: Union[str, Path]) -> str: + """Return the host filesystem name for doctor output, or 'unknown' if unavailable.""" + + target = Path(path) + if not target.exists(): + target = target.parent + commands = ( + ("stat", "-f", "%T", str(target)), + ("stat", "-f", "-c", "%T", str(target)), + ) + for command in commands: + try: + output = subprocess.check_output(command, stderr=subprocess.DEVNULL, text=True).strip() + except (OSError, subprocess.CalledProcessError): + continue + if output and "%T" not in output: + return output.lower() + return "unknown" + + +def wal_compatibility_probe( + directory: Union[str, Path], + policy: SQLitePolicyV1 = DEFAULT_SQLITE_POLICY, +) -> SQLitePragmaReportV1: + """Probe WAL on a disposable sibling DB; never mutate the target workflow DB.""" + + root = Path(directory) + if not root.is_dir(): + raise SQLitePolicyError(f"SQLite database directory does not exist: {root}") + descriptor, raw_path = tempfile.mkstemp(prefix=".hermes-wal-probe-", suffix=".sqlite", dir=root) + os.close(descriptor) + probe = Path(raw_path) + try: + with sqlite3.connect(probe, isolation_level=None) as connection: + report = apply_writable_pragmas( + connection, + policy, + allow_journal_mode_change=True, + database=probe, + ) + connection.execute("CREATE TABLE probe (value INTEGER NOT NULL)") + connection.execute("INSERT INTO probe VALUES (1)") + checkpoint = connection.execute("PRAGMA wal_checkpoint(TRUNCATE)").fetchone() + if checkpoint is None or int(checkpoint[0]) != 0: + raise WalCompatibilityError(f"WAL checkpoint remained busy: {checkpoint}") + return report + finally: + for candidate in (probe, Path(str(probe) + "-wal"), Path(str(probe) + "-shm")): + try: + candidate.unlink() + except FileNotFoundError: + pass + + +def doctor_sqlite_storage( + database: Union[str, Path], + *, + policy: SQLitePolicyV1 = DEFAULT_SQLITE_POLICY, + filesystem_type: Optional[str] = None, +) -> SQLiteStorageDoctorReportV1: + """Read/probe SQLite storage compatibility without changing the target journal mode.""" + + database_path = Path(database).expanduser().resolve(strict=False) + directory = database_path.parent + fs_type = (filesystem_type or detect_filesystem_type(directory)).strip().lower() + failures: list[str] = [] + current_mode = _read_journal_mode(database_path) + probe_performed = False + wal_compatible = False + + if database_path.exists() and not database_path.is_file(): + failures.append("target_database_not_regular_file") + elif database_path.is_file() and current_mode is None: + failures.append("target_database_unreadable") + if not directory.is_dir(): + failures.append("database_directory_missing") + elif fs_type in UNSUPPORTED_FILESYSTEMS: + failures.append(f"unsupported_filesystem:{fs_type}") + else: + probe_performed = True + try: + wal_compatibility_probe(directory, policy) + except (sqlite3.Error, SQLitePolicyError, OSError) as error: + failures.append(f"wal_probe_failed:{type(error).__name__}") + else: + wal_compatible = True + + change_required = current_mode is not None and current_mode != policy.journal_mode + plan = stop_checkpoint_start_plan(database_path) if change_required else () + return SQLiteStorageDoctorReportV1( + database_path=str(database_path), + filesystem_type=fs_type, + supported=not failures and wal_compatible, + failures=tuple(failures), + unsupported_filesystems=tuple(sorted(UNSUPPORTED_FILESYSTEMS)), + wal_probe_performed=probe_performed, + wal_compatible=wal_compatible, + current_journal_mode=current_mode, + journal_mode_change_required=change_required, + plan=plan, + ) + + +def require_supported_sqlite_storage( + report: SQLiteStorageDoctorReportV1, +) -> SQLiteStorageDoctorReportV1: + if not isinstance(report, SQLiteStorageDoctorReportV1): + raise TypeError("report must be SQLiteStorageDoctorReportV1") + if not report.supported: + raise UnsupportedSQLiteStorage(report.failures) + return report + + +def stop_checkpoint_start_plan(database: Union[str, Path]) -> Tuple[dict[str, str], ...]: + target = str(Path(database).expanduser().resolve(strict=False)) + return ( + { + "phase": "stop", + "database": target, + "requirement": "stop every process that can write the database", + "action": "confirm no active writer connections remain", + }, + { + "phase": "checkpoint", + "database": target, + "requirement": "run from one dedicated offline connection and require busy=0", + "action": "PRAGMA wal_checkpoint(TRUNCATE)", + }, + { + "phase": "switch", + "database": target, + "requirement": "change journal mode only while all services are stopped", + "action": "PRAGMA journal_mode=WAL", + }, + { + "phase": "start", + "database": target, + "requirement": "start exactly one writer first, then the remaining services", + "action": "reopen every writable connection through SQLitePolicyV1", + }, + { + "phase": "verify", + "database": target, + "requirement": "verify WAL, foreign_keys, busy_timeout, and zero lock diagnostics", + "action": "run storage doctor and contention probe before restoring traffic", + }, + ) + + +def _read_journal_mode(database: Path) -> Optional[str]: + if not database.is_file(): + return None + encoded = quote(str(database), safe="/") + try: + with sqlite3.connect(f"file:{encoded}?mode=ro", uri=True) as connection: + return str(connection.execute("PRAGMA journal_mode").fetchone()[0]).lower() + except sqlite3.Error: + return None + + +def _connection_database( + connection: sqlite3.Connection, + explicit: Optional[Union[str, Path]], +) -> Union[str, Path]: + if explicit is not None: + return explicit + row = connection.execute("PRAGMA database_list").fetchone() + if row is not None and len(row) >= 3 and row[2]: + return str(row[2]) + return ":memory:" + + +def _emit_diagnostic( + sink: Optional[Union[SQLiteDiagnosticSink, Callable[[Mapping[str, Any]], None]]], + record: Mapping[str, Any], +) -> None: + if sink is None: + return + emit = getattr(sink, "emit", None) + if callable(emit): + emit(record) + return + if callable(sink): + sink(record) + return + raise TypeError("diagnostic_sink must be callable or provide emit(record)") + + +def _diagnostic_record( + *, + event: str, + operation: str, + attempt: int, + max_attempts: int, + lock_count: int, + classification: str, + elapsed_ms: float, + delay_ms: Optional[float], + policy: SQLitePolicyV1, + plan: SQLiteRetryPlanV1, +) -> dict[str, object]: + record: dict[str, object] = { + "schema_version": SQLITE_POLICY_SCHEMA_VERSION, + "event": event, + "recorded_at": time.time(), + "operation": operation, + "attempt": attempt, + "max_attempts": max_attempts, + "lock_count": lock_count, + "classification": classification, + "elapsed_ms": _rounded(elapsed_ms), + "busy_timeout_ms": policy.busy_timeout_ms, + "lease_seconds": plan.lease_seconds, + "renewal_interval_seconds": plan.renewal_interval_seconds, + "lease_budget_ms": plan.budget_ms, + } + if delay_ms is not None: + record["delay_ms"] = _rounded(delay_ms) + return record + + +def _rounded(value: float) -> float: + return round(float(value), 3) diff --git a/tests/fixtures/sqlite_policy_v1.json b/tests/fixtures/sqlite_policy_v1.json new file mode 100644 index 0000000..7f79278 --- /dev/null +++ b/tests/fixtures/sqlite_policy_v1.json @@ -0,0 +1,61 @@ +{ + "schema_version": 1, + "policy": { + "schema_version": 1, + "journal_mode": "wal", + "foreign_keys": true, + "busy_timeout_ms": 50, + "retry_initial_delay_ms": 20, + "retry_multiplier": 2.0, + "retry_max_delay_ms": 80, + "retry_max_attempts": 6, + "retry_jitter_ratio": 0.25, + "lease_safety_margin_ms": 50 + }, + "pragmas": { + "journal_mode": "wal", + "foreign_keys": 1, + "busy_timeout_ms": 50 + }, + "known_lock_cases": [ + { + "message": "database is locked", + "classification": "locked" + }, + { + "message": "database table is locked", + "classification": "locked" + }, + { + "message": "database schema is locked: main", + "classification": "locked" + } + ], + "non_lock_messages": [ + "disk I/O error", + "no such table: workflow_instances", + "database or disk is full" + ], + "lease_case": { + "lease_seconds": 1.0, + "renewal_interval_seconds": 0.2, + "budget_ms": 750.0, + "max_attempts": 6, + "max_elapsed_ms": 675.0, + "delays_ms": [20.0, 40.0, 80.0, 80.0, 80.0], + "worst_case_delays_ms": [25.0, 50.0, 100.0, 100.0, 100.0] + }, + "unsupported_filesystems": [ + "9p", + "afpfs", + "cifs", + "davfs", + "fuse.sshfs", + "glusterfs", + "lustre", + "nfs", + "nfs4", + "smbfs", + "sshfs" + ] +} diff --git a/tests/probes/sqlite_lock_contention.py b/tests/probes/sqlite_lock_contention.py new file mode 100644 index 0000000..ef1b645 --- /dev/null +++ b/tests/probes/sqlite_lock_contention.py @@ -0,0 +1,150 @@ +from __future__ import annotations + +import hashlib +import json +import sqlite3 +import sys +import tempfile +import threading +import time +from pathlib import Path + + +REPO_ROOT = Path(__file__).resolve().parents[2] +SRC_ROOT = REPO_ROOT / "src" +if str(SRC_ROOT) not in sys.path: + sys.path.insert(0, str(SRC_ROOT)) + +from hermes_workflows.sqlite_policy import ( # noqa: E402 + JsonlSQLiteDiagnosticSink, + SQLitePolicyV1, + apply_writable_pragmas, + doctor_sqlite_storage, + lease_retry_plan, + run_with_lock_retry, +) + + +LEASE_SECONDS = 1.0 +RENEWAL_INTERVAL_SECONDS = 0.1 +LOCK_HOLD_SECONDS = 0.25 + + +def run_probe(root: Path) -> dict[str, object]: + database = root / "workflow.sqlite" + diagnostics_path = root / "sqlite-diagnostics.jsonl" + policy = SQLitePolicyV1( + busy_timeout_ms=20, + retry_initial_delay_ms=20, + retry_max_delay_ms=60, + retry_max_attempts=12, + retry_jitter_ratio=0.0, + lease_safety_margin_ms=50, + ) + with sqlite3.connect(database, isolation_level=None) as setup: + pragma_report = apply_writable_pragmas( + setup, + policy, + allow_journal_mode_change=True, + ) + setup.execute("CREATE TABLE effects (operation_id TEXT PRIMARY KEY)") + + blocker = sqlite3.connect(database, isolation_level=None) + apply_writable_pragmas(blocker, policy) + blocker.execute("BEGIN IMMEDIATE") + blocker.execute("INSERT INTO effects VALUES ('blocker')") + + diagnostics = JsonlSQLiteDiagnosticSink(diagnostics_path) + worker_result: dict[str, object] = {} + worker_error: list[BaseException] = [] + + def insert_once() -> str: + with sqlite3.connect(database, isolation_level=None) as writer: + apply_writable_pragmas(writer, policy) + writer.execute("INSERT INTO effects VALUES ('target-operation')") + return "written" + + def worker() -> None: + try: + started = time.monotonic() + worker_result["result"] = run_with_lock_retry( + insert_once, + policy=policy, + lease_seconds=LEASE_SECONDS, + renewal_interval_seconds=RENEWAL_INTERVAL_SECONDS, + operation_name="renewal_contention_probe", + diagnostic_sink=diagnostics, + random_value=lambda: 0.5, + ) + worker_result["elapsed_seconds"] = round(time.monotonic() - started, 3) + except BaseException as error: + worker_error.append(error) + + thread = threading.Thread(target=worker, name="sqlite-contention-probe") + thread.start() + time.sleep(LOCK_HOLD_SECONDS) + blocker.commit() + blocker.close() + thread.join(timeout=2.0) + if thread.is_alive(): + raise RuntimeError("contention worker did not finish within the lease bound") + if worker_error: + raise worker_error[0] + + records = [json.loads(line) for line in diagnostics_path.read_text().splitlines()] + events = [str(record["event"]) for record in records] + with sqlite3.connect(database) as connection: + target_count = int( + connection.execute( + "SELECT COUNT(*) FROM effects WHERE operation_id = 'target-operation'" + ).fetchone()[0] + ) + journal_mode = str(connection.execute("PRAGMA journal_mode").fetchone()[0]).lower() + if target_count != 1: + raise RuntimeError(f"target operation was silently lost or duplicated: count={target_count}") + if "sqlite.lock_retry" not in events or events[-1] != "sqlite.lock_recovered": + raise RuntimeError(f"contention did not produce durable retry/recovery diagnostics: {events}") + if LOCK_HOLD_SECONDS <= RENEWAL_INTERVAL_SECONDS: + raise RuntimeError("probe lock must outlive one renewal interval") + + fixture = REPO_ROOT / "tests" / "fixtures" / "sqlite_policy_v1.json" + unsupported = doctor_sqlite_storage( + database, + policy=policy, + filesystem_type="nfs", + ) + return { + "schema_version": 1, + "pragma_trace": pragma_report.to_dict(), + "observed_journal_mode": journal_mode, + "lease_bound": lease_retry_plan( + policy, + lease_seconds=LEASE_SECONDS, + renewal_interval_seconds=RENEWAL_INTERVAL_SECONDS, + ).to_dict(), + "lock_hold_seconds": LOCK_HOLD_SECONDS, + "renewal_interval_seconds": RENEWAL_INTERVAL_SECONDS, + "lock_outlived_renewal_interval": LOCK_HOLD_SECONDS > RENEWAL_INTERVAL_SECONDS, + "worker_result": worker_result, + "diagnostic_events": events, + "diagnostic_records": records, + "target_operation_count": target_count, + "unsupported_storage": { + "filesystem_type": unsupported.filesystem_type, + "supported": unsupported.supported, + "failures": list(unsupported.failures), + "wal_probe_performed": unsupported.wal_probe_performed, + }, + "fixture_sha256": hashlib.sha256(fixture.read_bytes()).hexdigest(), + } + + +def main() -> int: + with tempfile.TemporaryDirectory(prefix="hw13-sqlite-contention-") as temporary: + result = run_probe(Path(temporary)) + print(json.dumps(result, indent=2, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_sqlite_policy.py b/tests/test_sqlite_policy.py new file mode 100644 index 0000000..1d971dc --- /dev/null +++ b/tests/test_sqlite_policy.py @@ -0,0 +1,344 @@ +from __future__ import annotations + +import json +import sqlite3 +import threading +import time +from pathlib import Path + +import pytest + +from hermes_workflows.sqlite_policy import ( + JsonlSQLiteDiagnosticSink, + JournalModeChangeRequired, + LeaseUnsafePolicy, + SQLiteLockExhausted, + SQLitePolicyV1, + UnsupportedSQLiteStorage, + WalCompatibilityError, + apply_writable_pragmas, + classify_sqlite_lock, + doctor_sqlite_storage, + lease_retry_plan, + require_supported_sqlite_storage, + run_with_lock_retry, + stop_checkpoint_start_plan, +) + + +FIXTURE_PATH = Path(__file__).parent / "fixtures" / "sqlite_policy_v1.json" + + +def _fixture() -> dict[str, object]: + return json.loads(FIXTURE_PATH.read_text()) + + +def _policy() -> SQLitePolicyV1: + return SQLitePolicyV1.from_dict(_fixture()["policy"]) + + +def test_versioned_policy_and_lease_plan_match_frozen_fixture(): + fixture = _fixture() + policy = _policy() + + assert policy.to_dict() == fixture["policy"] + plan = lease_retry_plan( + policy, + lease_seconds=fixture["lease_case"]["lease_seconds"], + renewal_interval_seconds=fixture["lease_case"]["renewal_interval_seconds"], + ) + + assert plan.to_dict() == fixture["lease_case"] + + +def test_writable_pragmas_are_explicit_queryable_and_idempotent(tmp_path): + db = tmp_path / "workflow.sqlite" + policy = _policy() + + with sqlite3.connect(db, isolation_level=None) as connection: + report = apply_writable_pragmas( + connection, + policy, + allow_journal_mode_change=True, + ) + assert report.to_dict() == _fixture()["pragmas"] + assert connection.execute("PRAGMA journal_mode").fetchone()[0].lower() == "wal" + assert connection.execute("PRAGMA foreign_keys").fetchone()[0] == 1 + assert connection.execute("PRAGMA busy_timeout").fetchone()[0] == policy.busy_timeout_ms + + with sqlite3.connect(db, isolation_level=None) as connection: + assert apply_writable_pragmas(connection, policy).to_dict() == _fixture()["pragmas"] + + +def test_journal_mode_change_is_never_implicit(tmp_path): + db = tmp_path / "workflow.sqlite" + policy = _policy() + with sqlite3.connect(db, isolation_level=None) as connection: + before = connection.execute("PRAGMA journal_mode").fetchone()[0].lower() + assert before == "delete" + with pytest.raises(JournalModeChangeRequired) as raised: + apply_writable_pragmas(connection, policy) + after = connection.execute("PRAGMA journal_mode").fetchone()[0].lower() + + assert after == before + assert raised.value.current_mode == "delete" + assert raised.value.requested_mode == "wal" + assert raised.value.plan == stop_checkpoint_start_plan(db) + + +def test_wal_compatibility_fails_closed_when_sqlite_refuses_wal(): + policy = _policy() + with sqlite3.connect(":memory:", isolation_level=None) as connection: + with pytest.raises(WalCompatibilityError, match="returned journal_mode=memory"): + apply_writable_pragmas(connection, policy, allow_journal_mode_change=True) + + +def test_known_lock_classification_is_narrow_and_fixture_backed(): + fixture = _fixture() + for case in fixture["known_lock_cases"]: + assert classify_sqlite_lock(sqlite3.OperationalError(case["message"])) == case["classification"] + for message in fixture["non_lock_messages"]: + assert classify_sqlite_lock(sqlite3.OperationalError(message)) is None + assert classify_sqlite_lock(sqlite3.IntegrityError("database is locked")) is None + + +def test_lease_plan_truncates_attempts_before_the_safety_window(): + policy = _policy() + + plan = lease_retry_plan(policy, lease_seconds=0.5, renewal_interval_seconds=0.2) + + assert plan.budget_ms == 250.0 + assert plan.max_attempts == 3 + assert plan.max_elapsed_ms == 225.0 + assert plan.worst_case_delays_ms == (25.0, 50.0) + + +def test_lease_plan_rejects_a_busy_timeout_that_cannot_fit(): + policy = SQLitePolicyV1( + busy_timeout_ms=500, + retry_initial_delay_ms=20, + retry_max_delay_ms=80, + retry_max_attempts=2, + retry_jitter_ratio=0.0, + lease_safety_margin_ms=50, + ) + + with pytest.raises(LeaseUnsafePolicy, match="busy timeout"): + lease_retry_plan(policy, lease_seconds=0.5, renewal_interval_seconds=0.1) + + +def test_known_lock_retries_with_bounded_jitter_and_durable_recovery_diagnostic(tmp_path): + policy = _policy() + outcomes: list[object] = [ + sqlite3.OperationalError("database is locked"), + sqlite3.OperationalError("database table is locked"), + "written", + ] + sleeps: list[float] = [] + diagnostics = JsonlSQLiteDiagnosticSink(tmp_path / "sqlite-diagnostics.jsonl") + + def operation(): + outcome = outcomes.pop(0) + if isinstance(outcome, BaseException): + raise outcome + return outcome + + result = run_with_lock_retry( + operation, + policy=policy, + lease_seconds=1.0, + renewal_interval_seconds=0.2, + operation_name="renew_command_lease", + diagnostic_sink=diagnostics, + sleep=sleeps.append, + random_value=lambda: 1.0, + ) + + assert result == "written" + assert sleeps == [0.025, 0.05] + records = [json.loads(line) for line in diagnostics.path.read_text().splitlines()] + assert [record["event"] for record in records] == [ + "sqlite.lock_retry", + "sqlite.lock_retry", + "sqlite.lock_recovered", + ] + assert records[-1]["operation"] == "renew_command_lease" + assert records[-1]["lock_count"] == 2 + assert all(record["lease_budget_ms"] == 750.0 for record in records) + + +def test_lock_exhaustion_is_explicit_and_durable(tmp_path): + policy = SQLitePolicyV1( + busy_timeout_ms=10, + retry_initial_delay_ms=10, + retry_max_delay_ms=10, + retry_max_attempts=2, + retry_jitter_ratio=0.0, + lease_safety_margin_ms=10, + ) + diagnostics = JsonlSQLiteDiagnosticSink(tmp_path / "sqlite-diagnostics.jsonl") + + def locked(): + raise sqlite3.OperationalError("database is locked") + + with pytest.raises(SQLiteLockExhausted) as raised: + run_with_lock_retry( + locked, + policy=policy, + lease_seconds=0.2, + renewal_interval_seconds=0.05, + operation_name="claim_command", + diagnostic_sink=diagnostics, + sleep=lambda _: None, + random_value=lambda: 0.5, + ) + + assert raised.value.attempts == 2 + records = [json.loads(line) for line in diagnostics.path.read_text().splitlines()] + assert [record["event"] for record in records] == [ + "sqlite.lock_retry", + "sqlite.lock_exhausted", + ] + assert records[-1]["attempt"] == 2 + + +def test_non_lock_database_errors_are_never_retried_or_swallowed(): + calls = 0 + + def broken(): + nonlocal calls + calls += 1 + raise sqlite3.OperationalError("disk I/O error") + + with pytest.raises(sqlite3.OperationalError, match="disk I/O error"): + run_with_lock_retry( + broken, + policy=_policy(), + lease_seconds=1.0, + renewal_interval_seconds=0.2, + diagnostic_sink=lambda _: pytest.fail("non-lock error must not emit a diagnostic"), + sleep=lambda _: pytest.fail("non-lock error must not sleep"), + ) + assert calls == 1 + + +def test_two_connections_recover_after_contention_without_duplicate_write(tmp_path): + db = tmp_path / "workflow.sqlite" + policy = SQLitePolicyV1( + busy_timeout_ms=10, + retry_initial_delay_ms=10, + retry_max_delay_ms=20, + retry_max_attempts=10, + retry_jitter_ratio=0.0, + lease_safety_margin_ms=20, + ) + with sqlite3.connect(db, isolation_level=None) as setup: + apply_writable_pragmas(setup, policy, allow_journal_mode_change=True) + setup.execute("CREATE TABLE effects (operation_id TEXT PRIMARY KEY)") + + blocker = sqlite3.connect(db, isolation_level=None, check_same_thread=False) + apply_writable_pragmas(blocker, policy) + blocker.execute("BEGIN IMMEDIATE") + blocker.execute("INSERT INTO effects VALUES ('blocker')") + released = threading.Event() + + def release_lock(): + time.sleep(0.08) + blocker.commit() + released.set() + + thread = threading.Thread(target=release_lock) + thread.start() + diagnostics: list[dict[str, object]] = [] + try: + def insert_once(): + with sqlite3.connect(db, isolation_level=None) as writer: + apply_writable_pragmas(writer, policy) + writer.execute("INSERT INTO effects VALUES ('target')") + + run_with_lock_retry( + insert_once, + policy=policy, + lease_seconds=0.5, + renewal_interval_seconds=0.05, + operation_name="effect_receipt", + diagnostic_sink=lambda record: diagnostics.append(dict(record)), + ) + finally: + thread.join(timeout=1.0) + blocker.close() + + assert released.is_set() + assert diagnostics[-1]["event"] == "sqlite.lock_recovered" + with sqlite3.connect(db) as connection: + assert connection.execute( + "SELECT COUNT(*) FROM effects WHERE operation_id = 'target'" + ).fetchone()[0] == 1 + + +def test_unsupported_storage_fails_doctor_before_any_wal_probe(tmp_path): + report = doctor_sqlite_storage( + tmp_path / "workflow.sqlite", + policy=_policy(), + filesystem_type="nfs", + ) + + assert report.supported is False + assert report.failures == ("unsupported_filesystem:nfs",) + assert report.wal_probe_performed is False + with pytest.raises(UnsupportedSQLiteStorage, match="unsupported_filesystem:nfs"): + require_supported_sqlite_storage(report) + assert sorted(_fixture()["unsupported_filesystems"]) == sorted(report.unsupported_filesystems) + + +def test_doctor_probes_a_sibling_database_without_switching_the_target(tmp_path): + db = tmp_path / "workflow.sqlite" + with sqlite3.connect(db) as connection: + connection.execute("CREATE TABLE existing (value TEXT)") + with sqlite3.connect(db) as connection: + assert connection.execute("PRAGMA journal_mode").fetchone()[0].lower() == "delete" + + report = doctor_sqlite_storage( + db, + policy=_policy(), + filesystem_type="apfs", + ) + + assert report.supported is True + assert report.wal_probe_performed is True + assert report.wal_compatible is True + assert report.current_journal_mode == "delete" + assert report.journal_mode_change_required is True + assert report.plan == stop_checkpoint_start_plan(db) + require_supported_sqlite_storage(report) + with sqlite3.connect(db) as connection: + assert connection.execute("PRAGMA journal_mode").fetchone()[0].lower() == "delete" + assert sorted(path.name for path in tmp_path.iterdir()) == ["workflow.sqlite"] + + +def test_doctor_fails_an_unreadable_target_even_when_sibling_wal_probe_passes(tmp_path): + db = tmp_path / "workflow.sqlite" + db.write_text("not a SQLite database") + + report = doctor_sqlite_storage(db, policy=_policy(), filesystem_type="apfs") + + assert report.supported is False + assert report.failures == ("target_database_unreadable",) + assert report.wal_probe_performed is True + assert report.wal_compatible is True + with pytest.raises(UnsupportedSQLiteStorage, match="target_database_unreadable"): + require_supported_sqlite_storage(report) + + +def test_stop_checkpoint_start_plan_is_explicit_and_non_mutating(tmp_path): + db = tmp_path / "workflow.sqlite" + + plan = stop_checkpoint_start_plan(db) + + assert [step["phase"] for step in plan] == ["stop", "checkpoint", "switch", "start", "verify"] + assert plan[0]["requirement"] == "stop every process that can write the database" + assert "PRAGMA wal_checkpoint(TRUNCATE)" in plan[1]["action"] + assert plan[2]["action"] == "PRAGMA journal_mode=WAL" + assert plan[3]["requirement"] == "start exactly one writer first, then the remaining services" + assert plan[4]["requirement"] == "verify WAL, foreign_keys, busy_timeout, and zero lock diagnostics" + assert not db.exists() From f9f3f69fd9df7e773aa9bcd980aaf4de26a7d6d8 Mon Sep 17 00:00:00 2001 From: Skylar Payne <4830598+skylarbpayne@users.noreply.github.com> Date: Mon, 13 Jul 2026 18:07:23 -0700 Subject: [PATCH 2/4] fix sqlite retry lease budget accounting --- docs/operations/sqlite-policy.md | 2 +- src/hermes_workflows/sqlite_policy.py | 36 ++++++++++++++ tests/test_sqlite_policy.py | 67 +++++++++++++++++++++++++++ 3 files changed, 104 insertions(+), 1 deletion(-) diff --git a/docs/operations/sqlite-policy.md b/docs/operations/sqlite-policy.md index 521bae1..28a6cdd 100644 --- a/docs/operations/sqlite-policy.md +++ b/docs/operations/sqlite-policy.md @@ -51,7 +51,7 @@ The plan admits only the largest prefix of attempts for which: `attempts * B + sum(D_i) <= budget_ms` -If even one busy timeout cannot fit, `lease_retry_plan()` raises `LeaseUnsafePolicy`. Runtime elapsed time is checked again before each sleep, so a slow host fails explicitly before consuming the lease window. Jitter is bounded by `retry_jitter_ratio`; it cannot expand beyond the calculated worst case. +If even one busy timeout cannot fit, `lease_retry_plan()` raises `LeaseUnsafePolicy`. Runtime elapsed time is re-read after each durable diagnostic and sleep before another operation can start, so slow diagnostics or a slow host fail explicitly instead of consuming the lease window. Jitter is bounded by `retry_jitter_ratio`; it cannot expand beyond the calculated worst case. The contention probe holds `BEGIN IMMEDIATE` longer than one renewal interval, then verifies recovery before lease exhaustion, durable lock/recovery diagnostics, and exactly one target write: diff --git a/src/hermes_workflows/sqlite_policy.py b/src/hermes_workflows/sqlite_policy.py index b9a8c9c..8b14c14 100644 --- a/src/hermes_workflows/sqlite_policy.py +++ b/src/hermes_workflows/sqlite_policy.py @@ -467,7 +467,43 @@ def run_with_lock_retry( plan=plan, ), ) + elapsed_ms = (time.monotonic() - started) * 1000.0 + if elapsed_ms + delay_ms + policy.busy_timeout_ms > plan.budget_ms: + _emit_diagnostic( + diagnostic_sink, + _diagnostic_record( + event="sqlite.lock_exhausted", + operation=operation_name, + attempt=attempt, + max_attempts=plan.max_attempts, + lock_count=lock_count, + classification=classification, + elapsed_ms=elapsed_ms, + delay_ms=None, + policy=policy, + plan=plan, + ), + ) + raise SQLiteLockExhausted(operation_name, attempt, plan.budget_ms) from error sleep(delay_ms / 1000.0) + elapsed_ms = (time.monotonic() - started) * 1000.0 + if elapsed_ms + policy.busy_timeout_ms > plan.budget_ms: + _emit_diagnostic( + diagnostic_sink, + _diagnostic_record( + event="sqlite.lock_exhausted", + operation=operation_name, + attempt=attempt, + max_attempts=plan.max_attempts, + lock_count=lock_count, + classification=classification, + elapsed_ms=elapsed_ms, + delay_ms=None, + policy=policy, + plan=plan, + ), + ) + raise SQLiteLockExhausted(operation_name, attempt, plan.budget_ms) from error continue if lock_count: _emit_diagnostic( diff --git a/tests/test_sqlite_policy.py b/tests/test_sqlite_policy.py index 1d971dc..9210c61 100644 --- a/tests/test_sqlite_policy.py +++ b/tests/test_sqlite_policy.py @@ -8,6 +8,7 @@ import pytest +import hermes_workflows.sqlite_policy as sqlite_policy from hermes_workflows.sqlite_policy import ( JsonlSQLiteDiagnosticSink, JournalModeChangeRequired, @@ -202,6 +203,72 @@ def locked(): assert records[-1]["attempt"] == 2 +@pytest.mark.parametrize( + ("diagnostic_elapsed_seconds", "sleep_elapsed_seconds", "expected_sleeps"), + ((0.2, 0.0, 0), (0.0, 0.2, 1)), +) +def test_retry_never_starts_after_diagnostics_or_sleep_exhaust_the_lease_budget( + monkeypatch, + diagnostic_elapsed_seconds, + sleep_elapsed_seconds, + expected_sleeps, +): + policy = SQLitePolicyV1( + busy_timeout_ms=10, + retry_initial_delay_ms=10, + retry_max_delay_ms=10, + retry_max_attempts=2, + retry_jitter_ratio=0.0, + lease_safety_margin_ms=10, + ) + now = 0.0 + operation_calls: list[float] = [] + sleeps: list[float] = [] + diagnostics: list[dict[str, object]] = [] + + def monotonic(): + return now + + def operation(): + operation_calls.append(now) + if len(operation_calls) == 1: + raise sqlite3.OperationalError("database is locked") + return "written" + + def emit(record): + nonlocal now + diagnostics.append(dict(record)) + now += diagnostic_elapsed_seconds + + def sleep(seconds): + nonlocal now + sleeps.append(seconds) + now += sleep_elapsed_seconds + + monkeypatch.setattr(sqlite_policy.time, "monotonic", monotonic) + + with pytest.raises(SQLiteLockExhausted) as raised: + run_with_lock_retry( + operation, + policy=policy, + lease_seconds=0.2, + renewal_interval_seconds=0.05, + operation_name="claim_command", + diagnostic_sink=emit, + sleep=sleep, + random_value=lambda: 0.5, + ) + + assert raised.value.attempts == 1 + assert operation_calls == [0.0] + assert len(sleeps) == expected_sleeps + assert [record["event"] for record in diagnostics] == [ + "sqlite.lock_retry", + "sqlite.lock_exhausted", + ] + assert diagnostics[-1]["attempt"] == 1 + + def test_non_lock_database_errors_are_never_retried_or_swallowed(): calls = 0 From b9b63cb5cca2d0d7f0ea790c4f78a2c8d3d6c6c4 Mon Sep 17 00:00:00 2001 From: Skylar Payne <4830598+skylarbpayne@users.noreply.github.com> Date: Mon, 13 Jul 2026 18:38:02 -0700 Subject: [PATCH 3/4] HW-13: fence retry recovery at lease deadline --- src/hermes_workflows/sqlite_policy.py | 41 ++++++++++++- tests/test_sqlite_policy.py | 87 +++++++++++++++++++++++++++ 2 files changed, 127 insertions(+), 1 deletion(-) diff --git a/src/hermes_workflows/sqlite_policy.py b/src/hermes_workflows/sqlite_policy.py index 8b14c14..edc6cbd 100644 --- a/src/hermes_workflows/sqlite_policy.py +++ b/src/hermes_workflows/sqlite_policy.py @@ -402,7 +402,27 @@ def run_with_lock_retry( ) started = time.monotonic() lock_count = 0 + last_classification = "locked" for attempt in range(1, plan.max_attempts + 1): + if lock_count: + elapsed_ms = (time.monotonic() - started) * 1000.0 + if elapsed_ms + policy.busy_timeout_ms > plan.budget_ms: + _emit_diagnostic( + diagnostic_sink, + _diagnostic_record( + event="sqlite.lock_exhausted", + operation=operation_name, + attempt=attempt - 1, + max_attempts=plan.max_attempts, + lock_count=lock_count, + classification=last_classification, + elapsed_ms=elapsed_ms, + delay_ms=None, + policy=policy, + plan=plan, + ), + ) + raise SQLiteLockExhausted(operation_name, attempt - 1, plan.budget_ms) try: result = operation() except sqlite3.OperationalError as error: @@ -410,6 +430,7 @@ def run_with_lock_retry( if classification is None: raise lock_count += 1 + last_classification = classification elapsed_ms = (time.monotonic() - started) * 1000.0 if attempt >= plan.max_attempts: _emit_diagnostic( @@ -506,6 +527,24 @@ def run_with_lock_retry( raise SQLiteLockExhausted(operation_name, attempt, plan.budget_ms) from error continue if lock_count: + elapsed_ms = (time.monotonic() - started) * 1000.0 + if elapsed_ms + policy.busy_timeout_ms > plan.budget_ms: + _emit_diagnostic( + diagnostic_sink, + _diagnostic_record( + event="sqlite.lock_exhausted", + operation=operation_name, + attempt=attempt, + max_attempts=plan.max_attempts, + lock_count=lock_count, + classification=last_classification, + elapsed_ms=elapsed_ms, + delay_ms=None, + policy=policy, + plan=plan, + ), + ) + raise SQLiteLockExhausted(operation_name, attempt, plan.budget_ms) _emit_diagnostic( diagnostic_sink, _diagnostic_record( @@ -515,7 +554,7 @@ def run_with_lock_retry( max_attempts=plan.max_attempts, lock_count=lock_count, classification="locked", - elapsed_ms=(time.monotonic() - started) * 1000.0, + elapsed_ms=elapsed_ms, delay_ms=None, policy=policy, plan=plan, diff --git a/tests/test_sqlite_policy.py b/tests/test_sqlite_policy.py index 9210c61..a9c247f 100644 --- a/tests/test_sqlite_policy.py +++ b/tests/test_sqlite_policy.py @@ -269,6 +269,93 @@ def sleep(seconds): assert diagnostics[-1]["attempt"] == 1 +def test_retry_rechecks_lease_budget_immediately_before_the_next_operation(monkeypatch): + policy = SQLitePolicyV1( + busy_timeout_ms=10, + retry_initial_delay_ms=10, + retry_max_delay_ms=10, + retry_max_attempts=2, + retry_jitter_ratio=0.0, + lease_safety_margin_ms=10, + ) + timeline = iter((0.0, 0.0, 0.0, 0.12, 0.2)) + operation_calls = 0 + diagnostics: list[dict[str, object]] = [] + + def operation(): + nonlocal operation_calls + operation_calls += 1 + if operation_calls == 1: + raise sqlite3.OperationalError("database is locked") + return "written" + + monkeypatch.setattr(sqlite_policy.time, "monotonic", lambda: next(timeline)) + + with pytest.raises(SQLiteLockExhausted) as raised: + run_with_lock_retry( + operation, + policy=policy, + lease_seconds=0.2, + renewal_interval_seconds=0.05, + operation_name="claim_command", + diagnostic_sink=lambda record: diagnostics.append(dict(record)), + sleep=lambda _: None, + random_value=lambda: 0.5, + ) + + assert raised.value.attempts == 1 + assert operation_calls == 1 + assert [record["event"] for record in diagnostics] == [ + "sqlite.lock_retry", + "sqlite.lock_exhausted", + ] + assert diagnostics[-1]["elapsed_ms"] == pytest.approx(200.0) + + +def test_recovered_retry_must_complete_within_the_lease_budget(monkeypatch): + policy = SQLitePolicyV1( + busy_timeout_ms=10, + retry_initial_delay_ms=10, + retry_max_delay_ms=10, + retry_max_attempts=2, + retry_jitter_ratio=0.0, + lease_safety_margin_ms=10, + ) + now = 0.0 + operation_calls = 0 + diagnostics: list[dict[str, object]] = [] + + def operation(): + nonlocal now, operation_calls + operation_calls += 1 + if operation_calls == 1: + raise sqlite3.OperationalError("database is locked") + now = 0.2 + return "written" + + monkeypatch.setattr(sqlite_policy.time, "monotonic", lambda: now) + + with pytest.raises(SQLiteLockExhausted) as raised: + run_with_lock_retry( + operation, + policy=policy, + lease_seconds=0.2, + renewal_interval_seconds=0.05, + operation_name="claim_command", + diagnostic_sink=lambda record: diagnostics.append(dict(record)), + sleep=lambda _: None, + random_value=lambda: 0.5, + ) + + assert raised.value.attempts == 2 + assert operation_calls == 2 + assert [record["event"] for record in diagnostics] == [ + "sqlite.lock_retry", + "sqlite.lock_exhausted", + ] + assert diagnostics[-1]["elapsed_ms"] == pytest.approx(200.0) + + def test_non_lock_database_errors_are_never_retried_or_swallowed(): calls = 0 From d8d7e5f4a97a672d2e1b35de4ac2a0318bf5d0df Mon Sep 17 00:00:00 2001 From: Skylar Payne <4830598+skylarbpayne@users.noreply.github.com> Date: Mon, 13 Jul 2026 19:42:21 -0700 Subject: [PATCH 4/4] Freeze SQLite retry admission contract --- docs/operations/sqlite-policy.md | 14 ++-- src/hermes_workflows/sqlite_policy.py | 39 +---------- tests/fixtures/sqlite_policy_v1.json | 7 ++ tests/probes/sqlite_lock_contention.py | 8 ++- tests/test_sqlite_policy.py | 91 +++++++++++++++++++------- 5 files changed, 92 insertions(+), 67 deletions(-) diff --git a/docs/operations/sqlite-policy.md b/docs/operations/sqlite-policy.md index 28a6cdd..c8c642a 100644 --- a/docs/operations/sqlite-policy.md +++ b/docs/operations/sqlite-policy.md @@ -23,15 +23,15 @@ Every writable connection must apply and verify: Only `sqlite3.OperationalError` values identified as SQLite `BUSY`/`LOCKED`, or SQLite's known `database ... is locked` messages, are retryable. Integrity errors, full disks, missing tables, I/O errors, malformed SQL, and every unknown database error escape unchanged. Do not wrap broad database failures in this retry helper. -`run_with_lock_retry()` emits one durable diagnostic per lock decision through a supplied sink. `JsonlSQLiteDiagnosticSink` appends canonical JSON and calls `fsync` for every record. Integrations may provide a database/event sink, but they must preserve the version-1 fields: +`run_with_lock_retry()` emits one durable diagnostic per retry or exhaustion decision through a supplied sink. `JsonlSQLiteDiagnosticSink` appends canonical JSON and calls `fsync` for every record. Integrations may provide a database/event sink, but they must preserve the version-1 fields: -- event: `sqlite.lock_retry`, `sqlite.lock_recovered`, or `sqlite.lock_exhausted` +- event: `sqlite.lock_retry` or `sqlite.lock_exhausted` - operation and attempt/max-attempt counts - lock count and classification - busy timeout, lease, renewal interval, and lease-safe budget - elapsed time and the selected delay when retrying -Exhaustion raises `SQLiteLockExhausted`; it is never reported as success and never silently discarded. +Exhaustion raises `SQLiteLockExhausted`; it is never reported as success and never silently discarded. Version 1 does not emit `sqlite.lock_recovered`: a prior durable `sqlite.lock_retry` record plus the operation's normal result is the recovery trace. ## Lease-safe bound @@ -51,9 +51,13 @@ The plan admits only the largest prefix of attempts for which: `attempts * B + sum(D_i) <= budget_ms` -If even one busy timeout cannot fit, `lease_retry_plan()` raises `LeaseUnsafePolicy`. Runtime elapsed time is re-read after each durable diagnostic and sleep before another operation can start, so slow diagnostics or a slow host fail explicitly instead of consuming the lease window. Jitter is bounded by `retry_jitter_ratio`; it cannot expand beyond the calculated worst case. +HW-13 v1 is an admission-bounded retry policy, not a completion or publication deadline. The usable lease budget controls whether a retry may begin. Before every retry after the first, `run_with_lock_retry()` durably emits `sqlite.lock_retry`, applies only bounded jitter and sleep, re-reads monotonic time immediately before invoking the operation, and invokes it only when `elapsed_ms + busy_timeout_ms <= budget_ms`. If admission fails, it durably emits `sqlite.lock_exhausted` and raises without invoking another operation. -The contention probe holds `BEGIN IMMEDIATE` longer than one renewal interval, then verifies recovery before lease exhaustion, durable lock/recovery diagnostics, and exactly one target write: +If even one busy timeout cannot fit, `lease_retry_plan()` raises `LeaseUnsafePolicy`. Runtime elapsed time is re-read after each durable diagnostic and sleep before another operation can start, so slow diagnostics or a slow host fail explicitly instead of starting another SQLite unit outside the admission window. Jitter is bounded by `retry_jitter_ratio`; it cannot expand beyond the calculated worst case. + +The operation callable must be one bounded SQLite unit using the configured busy timeout, and it must own any domain idempotency needed for retry. Once an admitted operation returns normally, that result is authoritative and is returned immediately without a subsequent diagnostic callback, elapsed-time failure, or exception translation. Version 1 does not promise wall-clock completion or publication within the lease budget. Recovered-diagnostic and publication-deadline semantics require a separately specified, transaction-aware successor. + +The contention probe holds `BEGIN IMMEDIATE` longer than one renewal interval, then verifies at least one durable retry record, normal success, and exactly one target write: ```console python tests/probes/sqlite_lock_contention.py diff --git a/src/hermes_workflows/sqlite_policy.py b/src/hermes_workflows/sqlite_policy.py index edc6cbd..d1dd4cd 100644 --- a/src/hermes_workflows/sqlite_policy.py +++ b/src/hermes_workflows/sqlite_policy.py @@ -389,7 +389,7 @@ def run_with_lock_retry( sleep: Callable[[float], None] = time.sleep, random_value: Callable[[], float] = random.random, ) -> Any: - """Retry known lock failures only, bounded by the active lease safety window.""" + """Retry known lock failures only when another attempt fits the lease budget.""" if not callable(operation): raise TypeError("operation must be callable") @@ -424,7 +424,7 @@ def run_with_lock_retry( ) raise SQLiteLockExhausted(operation_name, attempt - 1, plan.budget_ms) try: - result = operation() + return operation() except sqlite3.OperationalError as error: classification = classify_sqlite_lock(error) if classification is None: @@ -526,41 +526,6 @@ def run_with_lock_retry( ) raise SQLiteLockExhausted(operation_name, attempt, plan.budget_ms) from error continue - if lock_count: - elapsed_ms = (time.monotonic() - started) * 1000.0 - if elapsed_ms + policy.busy_timeout_ms > plan.budget_ms: - _emit_diagnostic( - diagnostic_sink, - _diagnostic_record( - event="sqlite.lock_exhausted", - operation=operation_name, - attempt=attempt, - max_attempts=plan.max_attempts, - lock_count=lock_count, - classification=last_classification, - elapsed_ms=elapsed_ms, - delay_ms=None, - policy=policy, - plan=plan, - ), - ) - raise SQLiteLockExhausted(operation_name, attempt, plan.budget_ms) - _emit_diagnostic( - diagnostic_sink, - _diagnostic_record( - event="sqlite.lock_recovered", - operation=operation_name, - attempt=attempt, - max_attempts=plan.max_attempts, - lock_count=lock_count, - classification="locked", - elapsed_ms=elapsed_ms, - delay_ms=None, - policy=policy, - plan=plan, - ), - ) - return result raise AssertionError("bounded SQLite retry loop terminated without result or error") diff --git a/tests/fixtures/sqlite_policy_v1.json b/tests/fixtures/sqlite_policy_v1.json index 7f79278..5c97ce9 100644 --- a/tests/fixtures/sqlite_policy_v1.json +++ b/tests/fixtures/sqlite_policy_v1.json @@ -1,5 +1,12 @@ { "schema_version": 1, + "retry_contract": { + "budget_semantics": "retry_admission", + "success_semantics": "authoritative_operation_result", + "recovery_trace": "durable_sqlite.lock_retry_plus_normal_result", + "emits_lock_recovered": false, + "completion_deadline": false + }, "policy": { "schema_version": 1, "journal_mode": "wal", diff --git a/tests/probes/sqlite_lock_contention.py b/tests/probes/sqlite_lock_contention.py index ef1b645..c8149d3 100644 --- a/tests/probes/sqlite_lock_contention.py +++ b/tests/probes/sqlite_lock_contention.py @@ -87,7 +87,7 @@ def worker() -> None: blocker.close() thread.join(timeout=2.0) if thread.is_alive(): - raise RuntimeError("contention worker did not finish within the lease bound") + raise RuntimeError("contention worker did not finish within the probe timeout") if worker_error: raise worker_error[0] @@ -102,8 +102,10 @@ def worker() -> None: journal_mode = str(connection.execute("PRAGMA journal_mode").fetchone()[0]).lower() if target_count != 1: raise RuntimeError(f"target operation was silently lost or duplicated: count={target_count}") - if "sqlite.lock_retry" not in events or events[-1] != "sqlite.lock_recovered": - raise RuntimeError(f"contention did not produce durable retry/recovery diagnostics: {events}") + if "sqlite.lock_retry" not in events or any(event != "sqlite.lock_retry" for event in events): + raise RuntimeError(f"contention did not produce the durable v1 retry trace: {events}") + if worker_result.get("result") != "written": + raise RuntimeError(f"contention operation did not return normal success: {worker_result}") if LOCK_HOLD_SECONDS <= RENEWAL_INTERVAL_SECONDS: raise RuntimeError("probe lock must outlive one renewal interval") diff --git a/tests/test_sqlite_policy.py b/tests/test_sqlite_policy.py index a9c247f..f8fac39 100644 --- a/tests/test_sqlite_policy.py +++ b/tests/test_sqlite_policy.py @@ -42,6 +42,13 @@ def test_versioned_policy_and_lease_plan_match_frozen_fixture(): fixture = _fixture() policy = _policy() + assert fixture["retry_contract"] == { + "budget_semantics": "retry_admission", + "success_semantics": "authoritative_operation_result", + "recovery_trace": "durable_sqlite.lock_retry_plus_normal_result", + "emits_lock_recovered": False, + "completion_deadline": False, + } assert policy.to_dict() == fixture["policy"] plan = lease_retry_plan( policy, @@ -128,7 +135,7 @@ def test_lease_plan_rejects_a_busy_timeout_that_cannot_fit(): lease_retry_plan(policy, lease_seconds=0.5, renewal_interval_seconds=0.1) -def test_known_lock_retries_with_bounded_jitter_and_durable_recovery_diagnostic(tmp_path): +def test_known_lock_retries_with_bounded_jitter_and_durable_retry_trace(tmp_path): policy = _policy() outcomes: list[object] = [ sqlite3.OperationalError("database is locked"), @@ -161,7 +168,6 @@ def operation(): assert [record["event"] for record in records] == [ "sqlite.lock_retry", "sqlite.lock_retry", - "sqlite.lock_recovered", ] assert records[-1]["operation"] == "renew_command_lease" assert records[-1]["lock_count"] == 2 @@ -312,7 +318,7 @@ def operation(): assert diagnostics[-1]["elapsed_ms"] == pytest.approx(200.0) -def test_recovered_retry_must_complete_within_the_lease_budget(monkeypatch): +def test_admitted_retry_success_is_authoritative_after_the_lease_budget(monkeypatch): policy = SQLitePolicyV1( busy_timeout_ms=10, retry_initial_delay_ms=10, @@ -335,25 +341,64 @@ def operation(): monkeypatch.setattr(sqlite_policy.time, "monotonic", lambda: now) - with pytest.raises(SQLiteLockExhausted) as raised: - run_with_lock_retry( - operation, - policy=policy, - lease_seconds=0.2, - renewal_interval_seconds=0.05, - operation_name="claim_command", - diagnostic_sink=lambda record: diagnostics.append(dict(record)), - sleep=lambda _: None, - random_value=lambda: 0.5, - ) + result = run_with_lock_retry( + operation, + policy=policy, + lease_seconds=0.2, + renewal_interval_seconds=0.05, + operation_name="claim_command", + diagnostic_sink=lambda record: diagnostics.append(dict(record)), + sleep=lambda _: None, + random_value=lambda: 0.5, + ) - assert raised.value.attempts == 2 + assert result == "written" assert operation_calls == 2 - assert [record["event"] for record in diagnostics] == [ - "sqlite.lock_retry", - "sqlite.lock_exhausted", - ] - assert diagnostics[-1]["elapsed_ms"] == pytest.approx(200.0) + assert [record["event"] for record in diagnostics] == ["sqlite.lock_retry"] + + +def test_success_never_emits_recovered_or_reinvokes_the_effect(): + policy = SQLitePolicyV1( + busy_timeout_ms=10, + retry_initial_delay_ms=10, + retry_max_delay_ms=10, + retry_max_attempts=2, + retry_jitter_ratio=0.0, + lease_safety_margin_ms=10, + ) + attempts = 0 + committed_effects: list[str] = [] + diagnostics: list[str] = [] + + def operation(): + nonlocal attempts + attempts += 1 + if attempts == 1: + raise sqlite3.OperationalError("database is locked") + committed_effects.append("effect-1") + return "written" + + def emit(record): + event = str(record["event"]) + if event == "sqlite.lock_recovered": + raise RuntimeError("recovered diagnostics are outside the v1 contract") + diagnostics.append(event) + + result = run_with_lock_retry( + operation, + policy=policy, + lease_seconds=0.2, + renewal_interval_seconds=0.05, + operation_name="commit_effect", + diagnostic_sink=emit, + sleep=lambda _: None, + random_value=lambda: 0.5, + ) + + assert result == "written" + assert attempts == 2 + assert committed_effects == ["effect-1"] + assert diagnostics == ["sqlite.lock_retry"] def test_non_lock_database_errors_are_never_retried_or_swallowed(): @@ -410,7 +455,7 @@ def insert_once(): apply_writable_pragmas(writer, policy) writer.execute("INSERT INTO effects VALUES ('target')") - run_with_lock_retry( + result = run_with_lock_retry( insert_once, policy=policy, lease_seconds=0.5, @@ -423,7 +468,9 @@ def insert_once(): blocker.close() assert released.is_set() - assert diagnostics[-1]["event"] == "sqlite.lock_recovered" + assert result is None + assert diagnostics + assert all(record["event"] == "sqlite.lock_retry" for record in diagnostics) with sqlite3.connect(db) as connection: assert connection.execute( "SELECT COUNT(*) FROM effects WHERE operation_id = 'target'"