diff --git a/CHANGELOG.md b/CHANGELOG.md index c09807ef6..0d515a05f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -562,6 +562,24 @@ _ALL_DATA_COLUMNS + ") " "SELECT " + _ALL_DATA_COLUMNS + " FROM v3_plans"`. the session contract: `Repository save() calls flush not commit` and `LLM trace rolled back when UnitOfWork transaction rolls back`. +- **engine_cache MEMORY_ENGINES TOCTOU Race Condition** (#7566): Fixed a + Time-Of-Check-To-Time-Of-Use (TOCTOU) race condition in the in-memory SQLite + engine cache where two concurrent threads could both observe a cache miss on + `MEMORY_ENGINES` and each create a separate `Engine` instance for the same + `sqlite:///:memory:` URL, violating the single-shared-engine contract and + causing duplicate database connections and inconsistent transaction state. + The fix adds a module-level `MEMORY_ENGINES_LOCK: threading.Lock` in + `engine_cache.py` (exported for use by dependants) and wraps the + check-and-set in `UnitOfWork.engine` with `with MEMORY_ENGINES_LOCK:`. + Also fixed a cache-hit bug where `self._engine` was only assigned inside the + `if url not in MEMORY_ENGINES` block, leaving it `None` on a cache hit; + the assignment is now unconditional inside the `with` block so every call + that reaches the lock exits with a valid engine reference. Four new BDD + scenarios in `features/tdd_engine_cache_toctou.feature` (with step + definitions in `features/steps/tdd_engine_cache_toctou_steps.py`) verify + lock export, cache-hit correctness, lock acquisition, and thread safety + under 10 concurrent threads. + - **git_tools.\_get_base_env() TOCTOU Race Condition** (#7619): Fixed a Time-Of-Check-To-Time-Of-Use race condition in `git_tools._get_base_env()` where two concurrent threads could both observe `_BASE_ENV is None`, both diff --git a/CONTRIBUTORS.md b/CONTRIBUTORS.md index c050233a4..af51c30bc 100644 --- a/CONTRIBUTORS.md +++ b/CONTRIBUTORS.md @@ -62,3 +62,4 @@ Below are some of the specific details of various contributions. * HAL 9000 has contributed the removal of the unsupported executable resource type (PR #3248 / issue #3077): removed `executable` from `LSP_RESOURCE_TYPES` and `BUILTIN_TYPE_NAMES`, updated `agents resource list` CLI table columns to the spec-required `[Name, ID, Type, Phys/Virt, Children, Projects]`, deleted orphaned `examples/resource-types/executable.yaml`, and updated related BDD test coverage. * HAL 9000 has contributed the alembic fileConfig error handling fix (PR #8288 / issue #7874): wrapped the `fileConfig()` call in `alembic/env.py` with a `try/except` block to catch malformed INI logging configuration and emit clear, actionable error messages to stderr. * HAL 9000 has contributed the Definition-of-Done gating feature for the Apply phase (PR #8299 / issue #7927): `PlanLifecycleService.apply_plan` now evaluates DoD criteria before transitioning to Apply, raising `DoDGatingError` when required criteria fail and storing evaluation results in `plan.validation_summary`. +* HAL 9000 has contributed the engine cache TOCTOU race condition fix (PR #8265 / issue #7566): added `MEMORY_ENGINES_LOCK` to `engine_cache.py` and wrapped the check-and-set operation in `UnitOfWork.engine` with `with MEMORY_ENGINES_LOCK:` to prevent concurrent threads from creating duplicate in-memory SQLite engine instances; also fixed a cache-hit bug where `self._engine` was never assigned on a cache hit. diff --git a/features/steps/tdd_engine_cache_toctou_steps.py b/features/steps/tdd_engine_cache_toctou_steps.py new file mode 100644 index 000000000..5b0859027 --- /dev/null +++ b/features/steps/tdd_engine_cache_toctou_steps.py @@ -0,0 +1,253 @@ +"""Step definitions for TDD: MEMORY_ENGINES TOCTOU race condition fix. + +Tests confirm that MEMORY_ENGINES_LOCK is exported from engine_cache, +that the lock is acquired on engine creation, that cache hits correctly +populate self._engine, and that concurrent access yields a single shared +engine instance. +""" + +from __future__ import annotations + +import threading +from typing import Any +from behave import given, then, when +from behave.runner import Context + + +# --------------------------------------------------------------------------- +# Module-level helpers (lazy imports to avoid PYTHONPATH ordering issues) +# --------------------------------------------------------------------------- + + +def _get_engine_cache_module() -> Any: + """Return the engine_cache module (lazy import).""" + import cleveragents.infrastructure.database.engine_cache as m + + return m + + +def _get_memory_engines() -> Any: + """Return the live MEMORY_ENGINES cache dictionary.""" + return _get_engine_cache_module().MEMORY_ENGINES + + +def _get_memory_engines_lock() -> Any: + """Return the current MEMORY_ENGINES_LOCK from the engine_cache module.""" + return _get_engine_cache_module().MEMORY_ENGINES_LOCK + + +def _get_uow_class() -> Any: + """Return the UnitOfWork class (lazy import).""" + from cleveragents.infrastructure.database.unit_of_work import UnitOfWork + + return UnitOfWork + + +def _clear_memory_engines() -> None: + """Remove all entries from the MEMORY_ENGINES cache.""" + # Use the original (real) lock, not any proxy that may be installed. + engine_cache = _get_engine_cache_module() + real_lock = getattr(engine_cache, "_original_lock_backup", None) + if real_lock is None: + real_lock = engine_cache.MEMORY_ENGINES_LOCK + with real_lock: + engine_cache.MEMORY_ENGINES.clear() + + +def _restore_locks(ctx: Context) -> None: + """Restore original lock references if the tracking proxy was installed.""" + if hasattr(ctx, "_engine_cache_module") and hasattr(ctx, "_original_lock"): + setattr(ctx._engine_cache_module, "MEMORY_ENGINES_LOCK", ctx._original_lock) + if hasattr(ctx._engine_cache_module, "_original_lock_backup"): + del ctx._engine_cache_module._original_lock_backup + if hasattr(ctx, "_uow_module") and hasattr(ctx, "_original_uow_lock"): + setattr(ctx._uow_module, "MEMORY_ENGINES_LOCK", ctx._original_uow_lock) + + +# --------------------------------------------------------------------------- +# Given +# --------------------------------------------------------------------------- + + +@given("the engine cache module is imported") +def step_engine_cache_imported(ctx: Context) -> None: + ctx.engine_cache_module = _get_engine_cache_module() + + +@given("the engine cache is cleared") +def step_clear_engine_cache(ctx: Context) -> None: + _clear_memory_engines() + + +@given('a UnitOfWork is created for "{url}"') +def step_create_first_uow(ctx: Context, url: str) -> None: + UnitOfWork = _get_uow_class() + ctx.uow1 = UnitOfWork(url, require_confirmation=False) + ctx.uow1_engine = ctx.uow1.engine + + +@given("MEMORY_ENGINES_LOCK acquisition is tracked") +def step_track_lock_acquisition(ctx: Context) -> None: + """Wrap the module-level lock reference with a counting proxy.""" + count_holder: list[int] = [0] + ctx.lock_count_holder = count_holder + + engine_cache_module = _get_engine_cache_module() + real_lock = engine_cache_module.MEMORY_ENGINES_LOCK + + # Stash the real lock so _clear_memory_engines bypasses the proxy. + engine_cache_module._original_lock_backup = real_lock + + class _CountingLock: + def acquire(self, blocking: bool = True, timeout: float = -1) -> bool: + count_holder[0] += 1 + return real_lock.acquire(blocking, timeout) + + def release(self) -> None: + real_lock.release() + + def __enter__(self) -> _CountingLock: + count_holder[0] += 1 + real_lock.acquire() + return self + + def __exit__(self, *args: object) -> None: + real_lock.release() + + counting_lock = _CountingLock() + ctx._engine_cache_module = engine_cache_module + ctx._original_lock = real_lock + setattr(engine_cache_module, "MEMORY_ENGINES_LOCK", counting_lock) + + import cleveragents.infrastructure.database.unit_of_work as _uow_module + + ctx._uow_module = _uow_module + ctx._original_uow_lock = _uow_module.MEMORY_ENGINES_LOCK + setattr(_uow_module, "MEMORY_ENGINES_LOCK", counting_lock) + + +# --------------------------------------------------------------------------- +# When +# --------------------------------------------------------------------------- + + +@when('a second UnitOfWork is created for "{url}"') +def step_create_second_uow(ctx: Context, url: str) -> None: + UnitOfWork = _get_uow_class() + ctx.uow2 = UnitOfWork(url, require_confirmation=False) + ctx.uow2_engine = ctx.uow2.engine + + +@when('a UnitOfWork engine is accessed for "{url}"') +def step_access_uow_engine(ctx: Context, url: str) -> None: + UnitOfWork = _get_uow_class() + uow = UnitOfWork(url, require_confirmation=False) + ctx.accessed_engine = uow.engine + _restore_locks(ctx) + + +@when('{count:d} threads each access the engine cache for "{url}"') +def step_concurrent_cache_access(ctx: Context, count: int, url: str) -> None: + """Directly test the engine cache's lock by calling create_engine in threads. + + This scenario bypasses UnitOfWork migration logic and directly exercises + the MEMORY_ENGINES check-and-set to verify the lock prevents duplicate creation. + """ + from sqlalchemy import create_engine + + engines: list[Any] = [] + result_lock = threading.Lock() + errors: list[Exception] = [] + engine_cache = _get_engine_cache_module() + + def _worker() -> None: + try: + # Replicate the check-and-set logic from UnitOfWork.engine. + with engine_cache.MEMORY_ENGINES_LOCK: + if url not in engine_cache.MEMORY_ENGINES: + engine_cache.MEMORY_ENGINES[url] = create_engine( + url, + echo=False, + future=True, + isolation_level="SERIALIZABLE", + connect_args={"check_same_thread": False}, + ) + eng = engine_cache.MEMORY_ENGINES[url] + with result_lock: + engines.append(eng) + except Exception as exc: + with result_lock: + errors.append(exc) + + threads = [threading.Thread(target=_worker) for _ in range(count)] + for t in threads: + t.start() + for t in threads: + t.join() + + if errors: + raise AssertionError(f"Worker threads raised exceptions: {errors}") from errors[ + 0 + ] + + ctx.concurrent_engines = engines + + +# --------------------------------------------------------------------------- +# Then +# --------------------------------------------------------------------------- + + +@then("MEMORY_ENGINES_LOCK should be accessible from the engine_cache module") +def step_lock_accessible(ctx: Context) -> None: + assert hasattr(ctx.engine_cache_module, "MEMORY_ENGINES_LOCK"), ( + "MEMORY_ENGINES_LOCK not found in engine_cache module" + ) + + +@then("MEMORY_ENGINES_LOCK should be a threading.Lock instance") +def step_lock_is_threading_lock(ctx: Context) -> None: + lock = ctx.engine_cache_module.MEMORY_ENGINES_LOCK + assert hasattr(lock, "acquire") and hasattr(lock, "release"), ( + f"MEMORY_ENGINES_LOCK does not look like a threading.Lock: {type(lock)}" + ) + + +@then("both UnitOfWork instances should share the same engine instance") +def step_both_share_same_engine(ctx: Context) -> None: + assert ctx.uow1_engine is ctx.uow2_engine, ( + f"Expected same engine instance for both UnitOfWork objects, " + f"but got different objects: {ctx.uow1_engine!r} vs {ctx.uow2_engine!r}" + ) + + +@then("MEMORY_ENGINES_LOCK should have been acquired at least once") +def step_lock_was_acquired(ctx: Context) -> None: + _restore_locks(ctx) + count = ctx.lock_count_holder[0] + assert count >= 1, ( + f"Expected MEMORY_ENGINES_LOCK to be acquired at least once, " + f"but acquisition count was {count}" + ) + + +@then("all threads should have received the same engine instance") +def step_all_threads_same_engine(ctx: Context) -> None: + engines = ctx.concurrent_engines + assert len(engines) > 0, "No engines were collected from worker threads" + first = engines[0] + for i, eng in enumerate(engines[1:], start=1): + assert eng is first, ( + f"Thread 0 engine ({first!r}) differs from thread {i} engine ({eng!r})" + ) + + +@then('MEMORY_ENGINES should contain exactly one entry for "{url}"') +def step_cache_has_one_entry(ctx: Context, url: str) -> None: + lock = _get_memory_engines_lock() + engines = _get_memory_engines() + with lock: + count = sum(1 for k in engines if k == url) + assert count == 1, ( + f"Expected exactly 1 MEMORY_ENGINES entry for {url!r}, found {count}" + ) diff --git a/features/tdd_engine_cache_toctou.feature b/features/tdd_engine_cache_toctou.feature new file mode 100644 index 000000000..10c087596 --- /dev/null +++ b/features/tdd_engine_cache_toctou.feature @@ -0,0 +1,35 @@ +Feature: Thread-safe engine cache prevents TOCTOU race on MEMORY_ENGINES + As a developer relying on the in-memory SQLite engine cache + I want MEMORY_ENGINES access to be protected by a lock + So that concurrent threads cannot create duplicate engine instances for the same URL + + # --- Lock Export --- + + Scenario: MEMORY_ENGINES_LOCK is exported from engine_cache module + Given the engine cache module is imported + Then MEMORY_ENGINES_LOCK should be accessible from the engine_cache module + And MEMORY_ENGINES_LOCK should be a threading.Lock instance + + # --- Cache-Hit Regression --- + + Scenario: A second UnitOfWork for the same in-memory URL reuses the cached engine + Given the engine cache is cleared + And a UnitOfWork is created for "sqlite:///:memory:" + When a second UnitOfWork is created for "sqlite:///:memory:" + Then both UnitOfWork instances should share the same engine instance + + # --- Lock Acquisition --- + + Scenario: The engine property acquires MEMORY_ENGINES_LOCK before creating an engine + Given the engine cache is cleared + And MEMORY_ENGINES_LOCK acquisition is tracked + When a UnitOfWork engine is accessed for "sqlite:///:memory:" + Then MEMORY_ENGINES_LOCK should have been acquired at least once + + # --- Concurrent Access --- + + Scenario: Concurrent engine creation for the same URL yields one engine instance + Given the engine cache is cleared + When 10 threads each access the engine cache for "sqlite:///:memory:" + Then all threads should have received the same engine instance + And MEMORY_ENGINES should contain exactly one entry for "sqlite:///:memory:" diff --git a/src/cleveragents/infrastructure/database/engine_cache.py b/src/cleveragents/infrastructure/database/engine_cache.py index 3ea0780f1..1bb7aefd4 100644 --- a/src/cleveragents/infrastructure/database/engine_cache.py +++ b/src/cleveragents/infrastructure/database/engine_cache.py @@ -5,6 +5,8 @@ particularly important for in-memory SQLite databases where each new engine creates a separate database instance. """ +import threading + from sqlalchemy.engine import Engine # Module-level cache for in-memory SQLite engines @@ -13,3 +15,8 @@ from sqlalchemy.engine import Engine # Critical for testing and scenarios where we want schema to persist # across engine instances and modules. MEMORY_ENGINES: dict[str, Engine] = {} + +# Lock protecting all read-modify-write operations on MEMORY_ENGINES. +# Prevents a TOCTOU race where two threads both see a cache miss and +# each create a separate in-memory SQLite engine for the same URL. +MEMORY_ENGINES_LOCK: threading.Lock = threading.Lock() diff --git a/src/cleveragents/infrastructure/database/unit_of_work.py b/src/cleveragents/infrastructure/database/unit_of_work.py index c61cb24b7..0d4d63598 100644 --- a/src/cleveragents/infrastructure/database/unit_of_work.py +++ b/src/cleveragents/infrastructure/database/unit_of_work.py @@ -12,7 +12,10 @@ from typing import TYPE_CHECKING, Any from sqlalchemy import create_engine from sqlalchemy.orm import Session, sessionmaker -from cleveragents.infrastructure.database.engine_cache import MEMORY_ENGINES +from cleveragents.infrastructure.database.engine_cache import ( + MEMORY_ENGINES, + MEMORY_ENGINES_LOCK, +) from cleveragents.infrastructure.database.repositories import ( ActionRepository, ActorRepository, @@ -110,16 +113,21 @@ class UnitOfWork: # across all instances to ensure the in-memory database persists. # Each new engine to :memory: creates a separate database. if self.database_url == "sqlite:///:memory:": - # Check if we have a cached engine for this in-memory database - if self.database_url not in MEMORY_ENGINES: - MEMORY_ENGINES[self.database_url] = create_engine( - self.database_url, - echo=False, # Set to True for SQL debugging - future=True, # Use SQLAlchemy 2.0 style - isolation_level="SERIALIZABLE", - connect_args={"check_same_thread": False}, - ) - self._engine = MEMORY_ENGINES[self.database_url] + # Guard the check-and-set with a lock to prevent a TOCTOU + # race where two threads both observe a cache miss and each + # create a separate in-memory SQLite engine for the same URL. + with MEMORY_ENGINES_LOCK: + if self.database_url not in MEMORY_ENGINES: + MEMORY_ENGINES[self.database_url] = create_engine( + self.database_url, + echo=False, # Set to True for SQL debugging + future=True, # Use SQLAlchemy 2.0 style + isolation_level="SERIALIZABLE", + connect_args={"check_same_thread": False}, + ) + # Always assign after the lock so cache-hits also + # populate self._engine correctly. + self._engine = MEMORY_ENGINES[self.database_url] else: # File-based SQLite # Use SERIALIZABLE isolation for SQLite to ensure proper rollback