diff --git a/src/cachew/tests/test_sqlite_backends.py b/src/cachew/tests/test_sqlite_backends.py index bf6b6bf..598c1eb 100644 --- a/src/cachew/tests/test_sqlite_backends.py +++ b/src/cachew/tests/test_sqlite_backends.py @@ -1,10 +1,18 @@ +import logging +import sqlite3 from collections.abc import Generator, Iterator, Sequence +from contextlib import ExitStack, closing +from multiprocessing import get_context +from multiprocessing.process import BaseProcess from pathlib import Path -from typing import cast +from types import ModuleType +from typing import Protocol, cast import pytest -from .. import BACKENDS, Backend, cachew, get_logger +from .. import BACKENDS, Backend, cachew, get_logger, settings +from ..backend import sqlite as sqlalchemy_sqlite +from ..backend import sqlite_raw from ..backend.common import AbstractBackend from ..common import SourceHash @@ -15,11 +23,74 @@ ('sqlite_raw', 'sqlite_raw'), ] +_PROCESS_TIMEOUT_SECONDS = 10.0 +# The child timeout is longer so the parent remains responsible for diagnosing and cleaning up a stalled child. +_CHILD_RETRY_TIMEOUT_SECONDS = 30.0 + + +class _ProcessEvent(Protocol): + """The subset of a multiprocessing Event used by spawned workers.""" + + def set(self) -> None: ... + + def wait(self, timeout: float | None = None) -> bool: ... + + +class _RetryWaiter: + """Turn a production retry sleep into observable, parent-controlled synchronization.""" + + def __init__( + self, + *, + retry_entered: _ProcessEvent, + allow_retry: _ProcessEvent, + ) -> None: + self.retry_entered = retry_entered + self.allow_retry = allow_retry + + def sleep(self, _seconds: float) -> None: + # Reaching this method proves that the backend caught a lock error and entered its retry branch. + self.retry_entered.set() + # Keep the constructor inside that retry branch until the parent has observed both workers. + assert self.allow_retry.wait(timeout=_CHILD_RETRY_TIMEOUT_SECONDS) + def _backend(*, cache_path: Path, backend: Backend) -> AbstractBackend: return BACKENDS[backend](cache_path=cache_path, logger=get_logger()) +def _call_cachew_after_wal_retry( + *, + cache_path: Path, + backend: Backend, + retry_entered: _ProcessEvent, + allow_retry: _ProcessEvent, +) -> None: + """Run a strict Cachew call after exposing and pausing its first WAL retry.""" + + backend_module: ModuleType + if backend == 'sqlite': + backend_module = sqlalchemy_sqlite + else: + assert backend == 'sqlite_raw', backend + backend_module = sqlite_raw + + # Replace the backend module's binding without changing time.sleep globally in the child. + setattr( + backend_module, + 'time', + _RetryWaiter(retry_entered=retry_entered, allow_retry=allow_retry), + ) + settings.THROW_ON_ERROR = True + + @cachew(cache_path=cache_path, force_file=True, backend=backend) + def items(version: int) -> Iterator[int]: + yield version * 10 + yield version * 10 + 1 + + assert list(items(version=1)) == [10, 11] + + def _publish( *, cache_path: Path, @@ -46,6 +117,133 @@ def _assert_cache( assert list(reader.cached_blobs()) == list(blobs) +@pytest.mark.parametrize(('first_backend', 'second_backend'), _SQLITE_BACKEND_PAIRS) +def test_concurrent_first_cachew_calls_retry_initial_wal_transition( + *, + tmp_path: Path, + first_backend: Backend, + second_backend: Backend, +) -> None: + """ + Concurrent first calls may use the same new cache path, so a transient lock while enabling WAL must delay initialization rather than fail either decorated call. + Hold an empty DELETE-mode database exclusively, prove that both calls enter their backend's real retry branch, then release them and verify that they return normally and publish a reusable cache. + The parameter matrix covers both implementations in both arrival orders. + """ + cache_path = tmp_path / 'cache.sqlite' + assert cache_path.exists() is False + + # 'spawn' gives each worker a fresh interpreter instead of inheriting the parent's live SQLite connection and exclusive lock. + context = get_context('spawn') + # Each retry event identifies one worker reaching its retry branch, while allow_retry releases both workers together. + allow_retry = context.Event() + retry_events = [context.Event(), context.Event()] + backends = [first_backend, second_backend] + processes = [ + context.Process( + target=_call_cachew_after_wal_retry, + kwargs={ + 'cache_path': cache_path, + 'backend': backend, + 'retry_entered': retry_entered, + 'allow_retry': allow_retry, + }, + # Explicit reaping is still required, but daemon mode is a final safeguard against a stuck retry loop blocking interpreter shutdown. + daemon=True, + ) + for backend, retry_entered in zip(backends, retry_events, strict=True) + ] + # Only successfully started processes need joining or termination. + started_processes: list[BaseProcess] = [] + # A forced process failed to exit during the initial graceful join and required terminate(). + forced_processes: list[str] = [] + # A surviving process remained alive even after the terminate/kill cleanup sequence. + surviving_processes: list[str] = [] + # Exit codes prove that both planned workers completed successfully instead of merely disappearing. + exit_codes: list[int | None] = [] + + def reap_processes() -> None: + """Join every started worker, escalate bounded cleanup when needed, and retain failure diagnostics.""" + + # Give released workers a bounded opportunity to complete normally. + for process in started_processes: + process.join(timeout=_PROCESS_TIMEOUT_SECONDS) + # Any timeout is a test failure, but terminate the worker so pytest cannot hang during shutdown. + for process in started_processes: + if process.is_alive(): + forced_processes.append(process.name) + process.terminate() + for process in started_processes: + process.join(timeout=_PROCESS_TIMEOUT_SECONDS) + # Escalate after the graceful termination period. + for process in started_processes: + process.kill() + for process in started_processes: + process.join(timeout=_PROCESS_TIMEOUT_SECONDS) + + surviving_processes.extend(process.name for process in started_processes if process.is_alive()) + exit_codes.extend(process.exitcode for process in processes) + for process in processes: + if process.is_alive() is False: + process.close() + + with ExitStack() as cleanup: + # ExitStack is LIFO, so it rolls back and closes the blocker, releases both retry loops, and reaps the workers last. + cleanup.callback(reap_processes) + cleanup.callback(allow_retry.set) + # 'blocker' because its exclusive transaction deliberately prevents both constructors from changing journal mode. + blocker = sqlite3.connect( + cache_path, + timeout=0.0, + autocommit=cast(bool, sqlite3.LEGACY_TRANSACTION_CONTROL), + ) + cleanup.callback(blocker.close) + cleanup.callback(blocker.rollback) + + with closing(blocker.execute('PRAGMA journal_mode')) as cursor: + assert cursor.fetchone() == ('delete',) + # An exclusive lock forces both constructors through their actual WAL retry branches. + blocker.execute('BEGIN EXCLUSIVE').close() + + for backend, process, retry_entered in zip( + backends, + processes, + retry_events, + strict=True, + ): + process.start() + started_processes.append(process) + # Observe this worker's retry before starting the next process so pair order remains meaningful. + assert retry_entered.wait(timeout=_PROCESS_TIMEOUT_SECONDS), ( + backend, + process.exitcode, + ) + + assert forced_processes == [] + assert surviving_processes == [] + assert exit_codes == [0, 0] + + # Check the workers' persisted mode before those reads perform WAL setup and could mask a worker failure. + with closing(sqlite3.connect(cache_path)) as connection: + with closing(connection.execute('PRAGMA journal_mode')) as cursor: + assert cursor.fetchone() == ('wal',) + + def assert_cache_hit(*, reader_backend: Backend) -> None: + source_calls = 0 + + @cachew(cache_path=cache_path, force_file=True, backend=reader_backend) + def read_cache(version: int) -> Iterator[int]: # noqa: ARG001 + nonlocal source_calls + source_calls += 1 + yield -1 + + # A public cache hit proves that one contending first call published a complete, reusable result. + assert list(read_cache(version=1)) == [10, 11] + assert source_calls == 0 + + for reader_backend in backends: + assert_cache_hit(reader_backend=reader_backend) + + @pytest.mark.parametrize(('writer_backend', 'reader_backend'), _SQLITE_BACKEND_PAIRS) def test_sqlite_backend_cache_compatibility( *, @@ -231,3 +429,167 @@ def acquire_write_then_rollback() -> None: assert source_calls == [1, 2, 2] assert list(items(version=2)) == new_items assert source_calls == [1, 2, 2] + + +@pytest.mark.parametrize('throw_on_error', [False, True], ids=['fallback', 'strict']) +def test_cachew_handles_non_lock_wal_failure( + *, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + throw_on_error: bool, +) -> None: + """ + A genuine SQLite setup failure must follow Cachew's configured error policy instead of being retried forever as transient contention. + Strict mode must raise the original error without running the source, while defensive mode must run the source uncached; either path must close the failed connection and allow later calls to populate and reuse the cache. + """ + + monkeypatch.setattr(settings, 'THROW_ON_ERROR', throw_on_error) + real_connect = sqlite3.connect + wal_error = sqlite3.OperationalError('forced WAL failure') + # The explicit SQLite code sends this error through the non-retry branch regardless of its message. + setattr(wal_error, 'sqlite_errorcode', sqlite3.SQLITE_IOERR) + + # A narrow fake records the constructor's behavior without replacing methods on the C connection type. + class FailingConnection: + def __init__(self) -> None: + self.execute_calls = 0 + self.close_calls = 0 + + def execute(self, statement: str) -> sqlite3.Cursor: + assert statement == 'PRAGMA journal_mode=WAL' + # A retry would mean the non-lock IO error was misclassified and would otherwise loop forever. + assert self.execute_calls == 0 + self.execute_calls += 1 + raise wal_error + + def close(self) -> None: + self.close_calls += 1 + + connection = FailingConnection() + + def connect(*_args: object, **_kwargs: object) -> sqlite3.Connection: + return cast(sqlite3.Connection, connection) + + monkeypatch.setattr(sqlite_raw.sqlite3, 'connect', connect) + cache_path = tmp_path / 'cache.sqlite' + source_calls = 0 + + @cachew(cache_path=cache_path, force_file=True, backend='sqlite_raw') + def items() -> Iterator[int]: + nonlocal source_calls + source_calls += 1 + yield 1 + + if throw_on_error: + with pytest.raises(sqlite3.OperationalError) as raised: + list(items()) + assert raised.value is wal_error + assert source_calls == 0 + else: + assert list(items()) == [1] + assert source_calls == 1 + + # Raw SQLite preserves the original error and uses a note only to attach the cache path. + assert getattr(wal_error, '__notes__') == [f'while setting WAL mode on cache {cache_path}'] + assert connection.execute_calls == 1 + assert connection.close_calls == 1 + + # Restore real SQLite, then prove that the next call can write the cache and the following call can read it without running the source. + monkeypatch.setattr(sqlite_raw.sqlite3, 'connect', real_connect) + expected_source_calls = source_calls + 1 + assert list(items()) == [1] + assert source_calls == expected_source_calls + assert list(items()) == [1] + assert source_calls == expected_source_calls + + +@pytest.mark.parametrize('throw_on_error', [False, True], ids=['fallback', 'strict']) +def test_cachew_recovers_after_raw_transaction_entry_failure( + *, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + throw_on_error: bool, +) -> None: + """ + A failed cache transaction must raise before running the source in strict mode or run it once through defensive fallback. + Closing the unusable connection must also let the next call populate the cache and the following call reuse it. + """ + + monkeypatch.setattr(settings, 'THROW_ON_ERROR', throw_on_error) + fail_next_entry = True + failed_backends: list[sqlite_raw.SqliteRawBackend] = [] + failed_connections: list[sqlite3.Connection] = [] + + class FailFirstTransactionEntry(sqlite_raw.SqliteRawBackend): + def __init__(self, cache_path: Path, *, logger: logging.Logger) -> None: + nonlocal fail_next_entry + super().__init__(cache_path, logger=logger) + if fail_next_entry: + fail_next_entry = False + connection = self.connection + assert connection is not None + failed_backends.append(self) + failed_connections.append(connection) + # This transaction makes the BEGIN in the inherited __enter__ fail naturally. + # The inherited BEGIN will fail, and any leaked connection would keep this write lock and break recovery below. + connection.execute('BEGIN IMMEDIATE').close() + + monkeypatch.setitem(BACKENDS, 'sqlite_raw', FailFirstTransactionEntry) + cache_path = tmp_path / 'cache.sqlite' + source_calls = 0 + + @cachew(cache_path=cache_path, force_file=True, backend='sqlite_raw') + def items() -> Iterator[int]: + nonlocal source_calls + source_calls += 1 + yield 1 + + if throw_on_error: + with pytest.raises(sqlite3.OperationalError, match='cannot start a transaction within a transaction'): + list(items()) + assert source_calls == 0 + else: + assert list(items()) == [1] + assert source_calls == 1 + + [failed_backend] = failed_backends + [failed_connection] = failed_connections + assert failed_backend.connection is None + with pytest.raises(sqlite3.ProgrammingError, match='closed database'): + failed_connection.execute('SELECT 1') + + expected_source_calls = source_calls + 1 + assert list(items()) == [1] + assert source_calls == expected_source_calls + assert list(items()) == [1] + assert source_calls == expected_source_calls + + +@pytest.mark.parametrize( + ('error_code', 'expected'), + [ + pytest.param(sqlite3.SQLITE_BUSY , True , id='busy'), + pytest.param(sqlite3.SQLITE_LOCKED , True , id='locked'), + # SQLite stores extended error details above the low-byte primary result code. + pytest.param(sqlite3.SQLITE_BUSY | (1 << 8), True , id='extended-busy'), + pytest.param(sqlite3.SQLITE_LOCKED | (1 << 8), True , id='extended-locked'), + pytest.param(sqlite3.SQLITE_ERROR , False, id='error'), + pytest.param(sqlite3.SQLITE_IOERR | (1 << 8), False, id='extended-io-error'), + ], +) # fmt: skip +def test_sqlite_raw_lock_error_classification( + *, + error_code: int, + expected: bool, +) -> None: + """ + This deliberately remains a unit test because manufacturing every synthetic extended result code through Cachew would obscure the classification rule. + Cachew treats only SQLite lock contention as recoverable: WAL setup retries it, and write contention runs the source without updating the cache. + Unrelated failures must propagate to Cachew's normal strict-mode or defensive-fallback handling instead of being mistaken for locks. + Extended SQLite result codes retain that classification through their low-byte primary code. + """ + + error = sqlite3.OperationalError('forced error') + setattr(error, 'sqlite_errorcode', error_code) + + assert sqlite_raw._is_lock_error(error) is expected