diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 010aabf..7cc0d17 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -121,7 +121,7 @@ reads of the cache are observable. `warn_unused` (`eagle/decorators.py`) does the same around a single function call or `with` block, for code that runs outside the request cycle. Both are thin wrappers over `begin_request()` / `end_request()`. Everything recorded in between lands in a -thread-local `Collector`. +context-local `Collector`. ![img_1.png](assets/architecture/lifecycle.png) @@ -172,17 +172,27 @@ queryset-level one. ## The collector — per-request state -All tracking state lives in a single thread-local `Collector` (`unused/state.py`): +All tracking state lives in a `Collector` backed by a `contextvars.ContextVar` +(`unused/state.py`): ```python -class Collector(threading.local): +# The ContextVar holds the active request's state; the Collector reads/writes through it. +_state: ContextVar[_CollectorState] + +@dataclass +class _CollectorState: active: bool # is a request being tracked? loaded: dict[(model_name, cache_name), LoadedRelation] # kind + call-site consumed: set[(model_name, cache_name)] # relations that were accessed ``` -- **Thread-local** because Django serves each sync request on its own thread; per-thread - state keeps concurrent requests from contaminating each other. +- **Context-local** (a `contextvars.ContextVar`) so concurrent + requests never contaminate each other. A `ContextVar` isolates per thread *and* per + asyncio task, covering both WSGI (each sync request on its own thread) and ASGI (many + requests multiplexed onto one event-loop thread, where a thread-local would let one + request's `begin_request()` wipe another's state mid-`await`). `begin_request()` / + `end_request()` bind a *fresh* state object rather than mutating in place, so a request + starting while another is suspended cannot clobber the suspended request's tracking. - **Keyed by `(model_name, cache_name)`** — model *name* (not class) and the ORM cache key for the relation. This is why `mark_considered` accepts a model name string. - **First write wins** for `loaded`, so the originally captured call-site survives repeated diff --git a/README.md b/README.md index 5179bf4..25ca2a8 100644 --- a/README.md +++ b/README.md @@ -148,7 +148,7 @@ with warn_unused(): process(eagle.height) # ...but location is never read. ``` -`warn_unused` begins tracking before the scoped code runs and ends it on exit — exactly as the middleware does for a request — so the wasted join above emits the same `UnusedRelatedAccess` warning. Tracking always ends, even if the scoped code raises, so a failure never leaks tracking state onto the thread. The decorator form works on sync and async callables and preserves wrapper metadata (`__name__`, `__doc__`), and either form is a transparent passthrough when `EAGLE_ENABLED` is falsy. +`warn_unused` begins tracking before the scoped code runs and ends it on exit — exactly as the middleware does for a request — so the wasted join above emits the same `UnusedRelatedAccess` warning. Tracking always ends, even if the scoped code raises, so a failure never leaks tracking state into later work in the same context. The decorator form works on sync and async callables and preserves wrapper metadata (`__name__`, `__doc__`), and either form is a transparent passthrough when `EAGLE_ENABLED` is falsy. ## Suppressing false positives diff --git a/eagle/decorators.py b/eagle/decorators.py index def380c..1eb5129 100644 --- a/eagle/decorators.py +++ b/eagle/decorators.py @@ -92,7 +92,7 @@ def __exit__( End tracking, emitting warnings for relations loaded but never accessed. Tracking ends even when the block raises, so a failing block never leaks - an active collector into later work on the same thread. + an active collector into later work in the same context. Args: exc_type: Exception type raised in the block, if any. @@ -152,7 +152,7 @@ def task(): ... ... Tracking always ends even if the scoped code raises, so a failure never - leaks an active collector into later work on the same thread. When + leaks an active collector into later work in the same context. When ``EAGLE_ENABLED`` is falsy the scope is a transparent passthrough. The decorator forms work for both synchronous and asynchronous callables and preserve wrapper metadata (``__name__``, ``__doc__``). diff --git a/eagle/unused/state.py b/eagle/unused/state.py index 95609bb..3a30563 100644 --- a/eagle/unused/state.py +++ b/eagle/unused/state.py @@ -1,5 +1,5 @@ -import threading -from dataclasses import dataclass +import contextvars +from dataclasses import dataclass, field @dataclass(frozen=True) @@ -13,30 +13,83 @@ class LoadedRelation: RelationKey = tuple[str, str] -class Collector(threading.local): - """Thread-local store for loaded and consumed relation keys during a single request.""" +@dataclass +class _CollectorState: + """Mutable tracking state for one request: whether tracking is active plus the loaded/consumed relations.""" - active: bool - loaded: dict[RelationKey, LoadedRelation] - consumed: set[RelationKey] + active: bool = False + loaded: dict[RelationKey, LoadedRelation] = field(default_factory=dict) + consumed: set[RelationKey] = field(default_factory=set) - def __init__(self) -> None: - self.reset() - def reset(self) -> None: - """Clear all tracking state and deactivate the collector.""" - self.active = False - self.loaded = {} - self.consumed = set() +# Holds the active request's state. +_state: contextvars.ContextVar[_CollectorState] = contextvars.ContextVar("eagle_collector_state") + + +class Collector: + """ + Context-local store for loaded and consumed relation keys during a single request. + + Reads and writes route through a :class:`contextvars.ContextVar`, so concurrent requests -- + whether on separate threads (WSGI) or interleaved coroutines on one event-loop thread (ASGI) + -- never contaminate each other's tracking. + """ + + def _current(self) -> _CollectorState: + """ + Return this context's state, lazily installing an empty inactive one on first access. + + Returns: + The state object bound to the current context. + """ + state = _state.get(None) + if state is None: + state = _CollectorState() + _state.set(state) + return state + + @property + def active(self) -> bool: + """ + Whether a request is currently being tracked in this context. + + Returns: + True when tracking is active, False otherwise. + """ + return self._current().active + + @property + def loaded(self) -> dict[RelationKey, LoadedRelation]: + """ + Relations eager-loaded during this request, keyed by ``(model_name, cache_name)``. + + Returns: + The mutable map of loaded relations for the current context. + """ + return self._current().loaded + + @property + def consumed(self) -> set[RelationKey]: + """ + Relation keys that were actually accessed during this request. + + Returns: + The mutable set of consumed relation keys for the current context. + """ + return self._current().consumed def start(self) -> None: - """Reset and activate the collector for a new request.""" - self.reset() - self.active = True + """ + Install a fresh, active state for a new request in the current context. + + A new state object is bound rather than mutated in place so that, under ASGI, a request + beginning while another is suspended cannot clobber the suspended request's tracking. + """ + _state.set(_CollectorState(active=True)) def stop(self) -> None: - """Deactivate and reset the collector at request end.""" - self.reset() + """Install a fresh, inactive state, discarding this context's tracking at request end.""" + _state.set(_CollectorState(active=False)) collector = Collector() diff --git a/tests/test_concurrency_isolation.py b/tests/test_concurrency_isolation.py new file mode 100644 index 0000000..4039101 --- /dev/null +++ b/tests/test_concurrency_isolation.py @@ -0,0 +1,181 @@ +import asyncio +import threading +import warnings +from collections.abc import Callable, Coroutine +from typing import Any + +import pytest +from django.db import connections + +from eagle import UnusedRelatedAccess, warn_unused +from test_project.models import Eagle +from tests.factories import EagleFactory + + +def _run_async_and_capture(make_coro: Callable[[], Coroutine[Any, Any, None]]) -> list[warnings.WarningMessage]: + """ + Run an async orchestrator and return any UnusedRelatedAccess warnings it emitted. + + Recording (rather than letting the configured ``error`` filter raise) lets a single + test observe how warnings land across two interleaved requests instead of failing at + the first one. + + Args: + make_coro: Zero-arg factory returning the coroutine to drive with ``asyncio.run``. + + Returns: + The captured ``UnusedRelatedAccess`` warning messages, in emission order. + """ + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + asyncio.run(make_coro()) + return [w for w in caught if issubclass(w.category, UnusedRelatedAccess)] + + +@pytest.mark.django_db(transaction=True) +class TestAsyncRequestIsolation: + """ + Concurrent ASGI-style requests sharing one event-loop thread must not contaminate each other. + + Each request is its own asyncio task, so each runs in a copied context; the ContextVar-backed + collector keeps their tracking separate even when one request begins while another is suspended + at an ``await``. A thread-local collector would share state across the tasks and fail these. + + ``transaction=True`` commits the fixture row so each task's own (context-local) database + connection can read it without contending on the test's transaction lock. + """ + + @pytest.fixture(autouse=True) + def _allow_sync_orm_in_async(self, monkeypatch: pytest.MonkeyPatch) -> None: + """ + Permit synchronous ORM calls inside the event loop for the duration of the test. + + The queries here run cooperatively on the loop thread (never truly in parallel), so + Django's async-safety guard would be a false alarm. + """ + monkeypatch.setenv("DJANGO_ALLOW_ASYNC_UNSAFE", "1") + + def test_suspended_request_keeps_its_own_consumption(self) -> None: + """A request that loaded and accessed its relation must not warn when another request interleaves.""" + pk = EagleFactory().pk + + async def orchestrate() -> None: + a_consumed = asyncio.Event() + b_loaded = asyncio.Event() + a_finished = asyncio.Event() + + async def request_a() -> None: + try: + with warn_unused(): + eagle = Eagle.objects.select_related("location").get(pk=pk) + _ = eagle.location # A reads its relation, so A alone must never warn. + a_consumed.set() + # Stay open (suspended) until B has loaded but not yet consumed. + await b_loaded.wait() + # Exiting here ends A's request while B's relation is still unconsumed. + a_finished.set() + finally: + connections.close_all() + + async def request_b() -> None: + try: + await a_consumed.wait() # Begin only once A is mid-flight. + with warn_unused(): + eagle = Eagle.objects.select_related("location").get(pk=pk) + b_loaded.set() # Loaded but deliberately not yet accessed. + await a_finished.wait() # Let A end first, with B's load outstanding. + _ = eagle.location # B reads its relation after A has ended. + finally: + connections.close_all() + + await asyncio.gather(request_a(), request_b()) + + # Both requests read their relation, so the correct outcome is zero warnings. A shared + # collector would let B's unconsumed load surface as a false positive against A. + assert _run_async_and_capture(orchestrate) == [] + + def test_suspended_unused_load_still_warns(self) -> None: + """A request's genuinely unused load must still warn even though another request runs and resets first.""" + pk = EagleFactory().pk + + async def orchestrate() -> None: + b_loaded = asyncio.Event() + a_finished = asyncio.Event() + + async def request_a() -> None: + try: + await b_loaded.wait() # Begin only after B has an outstanding unused load. + with warn_unused(): + eagle = Eagle.objects.select_related("location").get(pk=pk) + _ = eagle.location # A reads its relation, so A must not warn. + a_finished.set() + finally: + connections.close_all() + + async def request_b() -> None: + try: + with warn_unused(): + eagle = Eagle.objects.select_related("location").get(pk=pk) + _ = eagle # Loaded but the relation is never accessed: B must warn. + b_loaded.set() + await a_finished.wait() # A runs (and resets a shared collector) before B ends. + # Exiting here ends B's request; its unused load must still be reported. + finally: + connections.close_all() + + await asyncio.gather(request_a(), request_b()) + + captured = _run_async_and_capture(orchestrate) + assert len(captured) == 1 + assert 'select_related("location")' in str(captured[0].message) + + +@pytest.mark.django_db(transaction=True) +class TestThreadRequestIsolation: + """ + Per-thread isolation still holds under the ContextVar collector, so the sync request path is unaffected. + + Each thread runs with its own fresh context, so two overlapping tracking scopes on different + threads keep separate loaded/consumed sets -- exactly as the old thread-local collector did. + """ + + def test_overlapping_threads_track_independently(self) -> None: + """Two threads with overlapping scopes warn only for the thread whose load went unused.""" + pk = EagleFactory().pk + + b_loaded = threading.Event() + a_finished = threading.Event() + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + + def request_a() -> None: + try: + b_loaded.wait() # Start only once B has an outstanding unused load. + with warn_unused(): + loaded = Eagle.objects.select_related("location").get(pk=pk) + _ = loaded.location # Accessed, so thread A must not warn. + a_finished.set() + finally: + connections.close_all() + + def request_b() -> None: + try: + with warn_unused(): + loaded = Eagle.objects.select_related("location").get(pk=pk) + _ = loaded # Loaded but never accessed: thread B must warn. + b_loaded.set() + a_finished.wait() # A's scope overlaps B's before B ends. + finally: + connections.close_all() + + thread_a = threading.Thread(target=request_a) + thread_b = threading.Thread(target=request_b) + thread_a.start() + thread_b.start() + thread_a.join() + thread_b.join() + + unused_warnings = [w for w in caught if issubclass(w.category, UnusedRelatedAccess)] + assert len(unused_warnings) == 1 + assert 'select_related("location")' in str(unused_warnings[0].message)