Add full async support

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

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
 

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__``).

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()

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)