diff --git a/docs/operations/sqlite-policy.md b/docs/operations/sqlite-policy.md new file mode 100644 index 0000000..c8c642a --- /dev/null +++ b/docs/operations/sqlite-policy.md @@ -0,0 +1,88 @@ +--- +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 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` 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. 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 + +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` + +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. + +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 +``` + +## 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..d1dd4cd --- /dev/null +++ b/src/hermes_workflows/sqlite_policy.py @@ -0,0 +1,754 @@ +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 when another attempt fits the lease budget.""" + + 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 + 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: + return operation() + except sqlite3.OperationalError as error: + classification = classify_sqlite_lock(error) + 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( + 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, + ), + ) + 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 + 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..5c97ce9 --- /dev/null +++ b/tests/fixtures/sqlite_policy_v1.json @@ -0,0 +1,68 @@ +{ + "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", + "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..c8149d3 --- /dev/null +++ b/tests/probes/sqlite_lock_contention.py @@ -0,0 +1,152 @@ +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 probe timeout") + 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 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") + + 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..f8fac39 --- /dev/null +++ b/tests/test_sqlite_policy.py @@ -0,0 +1,545 @@ +from __future__ import annotations + +import json +import sqlite3 +import threading +import time +from pathlib import Path + +import pytest + +import hermes_workflows.sqlite_policy as sqlite_policy +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 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, + 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_retry_trace(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", + ] + 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 + + +@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_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_admitted_retry_success_is_authoritative_after_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) + + 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 result == "written" + assert operation_calls == 2 + 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(): + 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')") + + result = 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 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'" + ).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()