⏺ Phase 7: Async Primitive Cleanup — Complete

1 new file + additions to 1 existing file, 16 new tests, all passing, mypy clean, no regressions.

  synapse/util/async_helpers.py — New native async utility functions:

  ┌──────────────────────────┬─────────────────────────────┬────────────────────────────┐
  │     Twisted Original     │      Native Equivalent      │        Description         │
  ├──────────────────────────┼─────────────────────────────┼────────────────────────────┤
  │ yieldable_gather_results │ native_gather_results       │ asyncio.gather wrapper     │
  ├──────────────────────────┼─────────────────────────────┼────────────────────────────┤
  │ concurrently_execute     │ native_concurrently_execute │ asyncio.Semaphore + gather │
  ├──────────────────────────┼─────────────────────────────┼────────────────────────────┤
  │ timeout_deferred         │ native_timeout              │ asyncio.wait_for wrapper   │
  ├──────────────────────────┼─────────────────────────────┼────────────────────────────┤
  │ stop_cancellation        │ native_stop_cancellation    │ asyncio.shield wrapper     │
  ├──────────────────────────┼─────────────────────────────┼────────────────────────────┤
  │ AwakenableSleeper        │ NativeAwakenableSleeper     │ asyncio.Event + wait_for   │
  ├──────────────────────────┼─────────────────────────────┼────────────────────────────┤
  │ DeferredEvent            │ NativeEvent                 │ asyncio.Event wrapper      │
  └──────────────────────────┴─────────────────────────────┴────────────────────────────┘

  synapse/util/caches/future_cache.py — FutureCache:
  - asyncio-native equivalent of DeferredCache
  - Two-layer architecture: _pending (in-flight futures) + _completed (LRU dict)
  - Uses ObservableFuture for multiple observers of the same pending entry
  - Invalidation callbacks on both layers
  - Max entries with eviction of oldest completed entries
  - Failed futures are NOT cached (same behavior as DeferredCache)
  - No make_deferred_yieldable() needed — futures are directly awaitable

  ---
  Running totals across Phases 0-7:
  - 10 new files, ~3200 lines of asyncio-native implementation
  - 144 tests all passing
  - All mypy clean
  - Existing 4462-test suite unaffected

synapse/util/async_helpers.py

@@ -1270,6 +1270,125 @@ class NativeReadWriteLock:
             await event.wait()
 
 
+async def native_gather_results(
+    func: Callable[..., Awaitable[R]],
+    iter: Iterable[Any],
+    *args: Any,
+    **kwargs: Any,
+) -> list[R]:
+    """asyncio-native equivalent of yieldable_gather_results.
+
+    Executes func with each iterable item concurrently using asyncio.gather.
+    Returns list of results in order.
+    """
+
+    async def _run(item: Any) -> R:
+        return await func(item, *args, **kwargs)
+
+    return list(await asyncio.gather(*[_run(item) for item in iter]))
+
+
+async def native_concurrently_execute(
+    func: Callable[[Any], Awaitable[Any]],
+    args: Iterable[Any],
+    limit: int,
+) -> None:
+    """asyncio-native equivalent of concurrently_execute.
+
+    Executes func on each arg with at most `limit` concurrent executions.
+    """
+    sem = asyncio.Semaphore(limit)
+
+    async def _run(item: Any) -> None:
+        async with sem:
+            await func(item)
+
+    await asyncio.gather(*[_run(item) for item in args])
+
+
+async def native_timeout(
+    awaitable: Awaitable[R],
+    timeout: float,
+) -> R:
+    """asyncio-native equivalent of timeout_deferred.
+
+    Wraps an awaitable with a timeout. Raises asyncio.TimeoutError on timeout.
+    """
+    return await asyncio.wait_for(asyncio.ensure_future(awaitable), timeout)
+
+
+def native_stop_cancellation(future: "asyncio.Future[T]") -> "asyncio.Future[T]":
+    """asyncio-native equivalent of stop_cancellation.
+
+    Wraps a future so that cancelling the returned future does not
+    propagate cancellation to the original. The returned future will
+    fail with CancelledError, but the original keeps running.
+    """
+    return asyncio.shield(future)
+
+
+class NativeAwakenableSleeper:
+    """asyncio-native equivalent of AwakenableSleeper.
+
+    Allows explicit waking of sleeping coroutines by name.
+    """
+
+    def __init__(self) -> None:
+        self._streams: dict[str, set[asyncio.Event]] = {}
+
+    def wake(self, name: str) -> None:
+        """Wake everything sleeping under `name`."""
+        events = self._streams.pop(name, set())
+        for event in events:
+            event.set()
+
+    async def sleep(self, name: str, delay_ms: int) -> None:
+        """Sleep for delay_ms, or return early if wake(name) is called."""
+        event = asyncio.Event()
+        stream_set = self._streams.setdefault(name, set())
+        stream_set.add(event)
+
+        try:
+            await asyncio.wait_for(event.wait(), timeout=delay_ms / 1000)
+        except asyncio.TimeoutError:
+            pass
+        finally:
+            curr = self._streams.get(name)
+            if curr is not None:
+                curr.discard(event)
+                if not curr:
+                    self._streams.pop(name, None)
+
+
+class NativeEvent:
+    """asyncio-native equivalent of DeferredEvent.
+
+    Like threading.Event but for asyncio code.
+    """
+
+    def __init__(self) -> None:
+        self._event = asyncio.Event()
+
+    def set(self) -> None:
+        self._event.set()
+
+    def clear(self) -> None:
+        self._event.clear()
+
+    def is_set(self) -> bool:
+        return self._event.is_set()
+
+    async def wait(self, timeout_seconds: float) -> bool:
+        """Wait for the event, returning True if set, False on timeout."""
+        if self._event.is_set():
+            return True
+        try:
+            await asyncio.wait_for(self._event.wait(), timeout=timeout_seconds)
+        except asyncio.TimeoutError:
+            pass
+        return self._event.is_set()
+
+
 class AwakenableSleeper:
     """Allows explicitly waking up deferreds related to an entity that are
     currently sleeping.

synapse/util/caches/future_cache.py

@@ -0,0 +1,259 @@
+#
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2025 New Vector, Ltd
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# See the GNU Affero General Public License for more details:
+# <https://www.gnu.org/licenses/agpl-3.0.html>.
+#
+
+"""asyncio-native cache using asyncio.Future instead of Deferred.
+
+Phase 7 of the Twisted → asyncio migration. FutureCache is the asyncio-native
+equivalent of DeferredCache, using asyncio.Future and ObservableFuture
+instead of Deferred and ObservableDeferred.
+
+This module is unused until later phases switch the cache decorators to use it.
+"""
+
+import asyncio
+import logging
+from typing import Any, Callable, Generic, Hashable, TypeVar
+
+from synapse.util.async_helpers import ObservableFuture
+
+logger = logging.getLogger(__name__)
+
+VT = TypeVar("VT")
+
+
+class FutureCacheEntry(Generic[VT]):
+    """A single pending cache entry backed by an ObservableFuture."""
+
+    __slots__ = ["_observable", "_callbacks"]
+
+    def __init__(self, future: "asyncio.Future[VT]") -> None:
+        self._observable = ObservableFuture(future, consumeErrors=True)
+        self._callbacks: list[Callable[[], None]] = []
+
+    def observe(self) -> "asyncio.Future[VT]":
+        """Return a new Future that resolves with the same value."""
+        return self._observable.observe()
+
+    def has_completed(self) -> bool:
+        return self._observable.has_called()
+
+    def has_succeeded(self) -> bool:
+        return self._observable.has_succeeded()
+
+    def get_result(self) -> VT:
+        result = self._observable.get_result()
+        if isinstance(result, BaseException):
+            raise result
+        return result
+
+    def add_invalidation_callback(self, callback: Callable[[], None]) -> None:
+        self._callbacks.append(callback)
+
+    def run_invalidation_callbacks(self) -> None:
+        for cb in self._callbacks:
+            try:
+                cb()
+            except Exception:
+                logger.exception("Error running cache invalidation callback")
+        self._callbacks.clear()
+
+
+class FutureCache(Generic[VT]):
+    """asyncio-native cache using asyncio.Future.
+
+    This is the asyncio-native equivalent of DeferredCache. It uses a
+    two-layer architecture:
+
+    1. _pending: Stores FutureCacheEntry for in-flight operations
+       (futures that haven't completed yet). Multiple callers can observe
+       the same pending entry.
+
+    2. _completed: Stores completed values (plain Python objects, not futures).
+       This is the actual LRU cache.
+
+    The lifecycle is:
+    - set(key, future) → entry goes into _pending
+    - When future completes successfully → value moves to _completed, removed from _pending
+    - When future fails → entry removed from _pending (no caching of errors)
+    - get(key) → returns from _completed, or observer of _pending, or raises KeyError
+
+    Unlike DeferredCache, this does NOT need make_deferred_yieldable() wrapping
+    since asyncio Futures are directly awaitable.
+    """
+
+    def __init__(
+        self,
+        name: str,
+        max_entries: int = 1000,
+    ) -> None:
+        self.name = name
+        self._max_entries = max_entries
+
+        # Pending (in-flight) futures
+        self._pending: dict[Hashable, FutureCacheEntry[VT]] = {}
+
+        # Completed values (simple LRU approximation using ordered dict)
+        self._completed: dict[Hashable, VT] = {}
+
+        # Invalidation callbacks for completed entries
+        self._completed_callbacks: dict[Hashable, list[Callable[[], None]]] = {}
+
+    def get(
+        self,
+        key: Hashable,
+        callback: Callable[[], None] | None = None,
+    ) -> "asyncio.Future[VT]":
+        """Get a value from the cache.
+
+        Returns a Future that resolves with the cached value.
+        If the value is pending (in-flight), returns an observer of the
+        pending future.
+
+        Args:
+            key: Cache key.
+            callback: Optional invalidation callback, fired when this key is
+                invalidated.
+
+        Returns:
+            A Future resolving to the cached value.
+
+        Raises:
+            KeyError: If the key is not in the cache (neither pending nor completed).
+        """
+        # Check pending first
+        entry = self._pending.get(key)
+        if entry is not None:
+            if callback:
+                entry.add_invalidation_callback(callback)
+            return entry.observe()
+
+        # Check completed
+        if key in self._completed:
+            if callback:
+                self._completed_callbacks.setdefault(key, []).append(callback)
+            loop = asyncio.get_running_loop()
+            f: asyncio.Future[VT] = loop.create_future()
+            f.set_result(self._completed[key])
+            return f
+
+        raise KeyError(key)
+
+    def set(
+        self,
+        key: Hashable,
+        future: "asyncio.Future[VT]",
+        callback: Callable[[], None] | None = None,
+    ) -> "asyncio.Future[VT]":
+        """Store a future in the cache.
+
+        If the future completes successfully, the result is moved to the
+        completed cache. If it fails, the entry is removed.
+
+        Args:
+            key: Cache key.
+            future: The future producing the value.
+            callback: Optional invalidation callback.
+
+        Returns:
+            An observer future that resolves with the same value.
+        """
+        # Invalidate any existing entry
+        self.invalidate(key)
+
+        entry = FutureCacheEntry(future)
+        if callback:
+            entry.add_invalidation_callback(callback)
+        self._pending[key] = entry
+
+        def _on_complete(f: "asyncio.Future[VT]") -> None:
+            # Remove from pending
+            if self._pending.get(key) is entry:
+                del self._pending[key]
+
+            if f.cancelled():
+                return
+
+            exc = f.exception()
+            if exc is not None:
+                # Don't cache errors
+                return
+
+            # Move to completed cache
+            result = f.result()
+            self._completed[key] = result
+
+            # Move callbacks from entry to completed callbacks
+            if entry._callbacks:
+                self._completed_callbacks.setdefault(key, []).extend(
+                    entry._callbacks
+                )
+                entry._callbacks.clear()
+
+            # Evict old entries if over limit
+            self._maybe_evict()
+
+        future.add_done_callback(_on_complete)
+        return entry.observe()
+
+    def invalidate(self, key: Hashable) -> None:
+        """Remove an entry from the cache and fire invalidation callbacks."""
+        # Invalidate pending
+        entry = self._pending.pop(key, None)
+        if entry is not None:
+            entry.run_invalidation_callbacks()
+
+        # Invalidate completed
+        self._completed.pop(key, None)
+        callbacks = self._completed_callbacks.pop(key, [])
+        for cb in callbacks:
+            try:
+                cb()
+            except Exception:
+                logger.exception("Error running cache invalidation callback")
+
+    def invalidate_all(self) -> None:
+        """Remove all entries and fire all invalidation callbacks."""
+        # Pending
+        for entry in self._pending.values():
+            entry.run_invalidation_callbacks()
+        self._pending.clear()
+
+        # Completed
+        for callbacks in self._completed_callbacks.values():
+            for cb in callbacks:
+                try:
+                    cb()
+                except Exception:
+                    logger.exception("Error running cache invalidation callback")
+        self._completed.clear()
+        self._completed_callbacks.clear()
+
+    def __contains__(self, key: Hashable) -> bool:
+        return key in self._pending or key in self._completed
+
+    def __len__(self) -> int:
+        return len(self._pending) + len(self._completed)
+
+    def _maybe_evict(self) -> None:
+        """Evict oldest completed entries if over the max size."""
+        while len(self._completed) > self._max_entries:
+            # Pop the first (oldest) item
+            oldest_key = next(iter(self._completed))
+            self._completed.pop(oldest_key)
+            callbacks = self._completed_callbacks.pop(oldest_key, [])
+            for cb in callbacks:
+                try:
+                    cb()
+                except Exception:
+                    pass

tests/util/test_native_async.py

@@ -1437,5 +1437,261 @@ class NativeReplicationProtocolTest(unittest.IsolatedAsyncioTestCase):
         await server.wait_closed()
 
 
+class NativeAsyncUtilitiesTest(unittest.IsolatedAsyncioTestCase):
+    """Tests for Phase 7 native async utility functions."""
+
+    async def test_native_gather_results(self) -> None:
+        from synapse.util.async_helpers import native_gather_results
+
+        async def double(x: int) -> int:
+            return x * 2
+
+        results = await native_gather_results(double, [1, 2, 3])
+        self.assertEqual(results, [2, 4, 6])
+
+    async def test_native_concurrently_execute(self) -> None:
+        from synapse.util.async_helpers import native_concurrently_execute
+
+        results: list[int] = []
+        concurrent = 0
+        max_concurrent = 0
+
+        async def work(x: int) -> None:
+            nonlocal concurrent, max_concurrent
+            concurrent += 1
+            max_concurrent = max(max_concurrent, concurrent)
+            results.append(x)
+            await asyncio.sleep(0.01)
+            concurrent -= 1
+
+        await native_concurrently_execute(work, range(10), limit=3)
+
+        self.assertEqual(sorted(results), list(range(10)))
+        self.assertLessEqual(max_concurrent, 3)
+
+    async def test_native_stop_cancellation(self) -> None:
+        from synapse.util.async_helpers import native_stop_cancellation
+
+        loop = asyncio.get_running_loop()
+        inner: asyncio.Future[str] = loop.create_future()
+        shielded = native_stop_cancellation(inner)
+
+        # Cancel the shielded future
+        shielded.cancel()
+
+        # Inner should NOT be cancelled
+        self.assertFalse(inner.cancelled())
+
+        # Resolve inner
+        inner.set_result("ok")
+        self.assertEqual(inner.result(), "ok")
+
+    async def test_native_awakeable_sleeper(self) -> None:
+        from synapse.util.async_helpers import NativeAwakenableSleeper
+
+        sleeper = NativeAwakenableSleeper()
+
+        woke_early = False
+
+        async def sleeping() -> None:
+            nonlocal woke_early
+            await sleeper.sleep("test", delay_ms=5000)
+            woke_early = True
+
+        task = asyncio.create_task(sleeping())
+        await asyncio.sleep(0.01)
+
+        sleeper.wake("test")
+        await asyncio.sleep(0.01)
+
+        self.assertTrue(woke_early)
+        task.cancel()
+        try:
+            await task
+        except asyncio.CancelledError:
+            pass
+
+    async def test_native_awakeable_sleeper_timeout(self) -> None:
+        from synapse.util.async_helpers import NativeAwakenableSleeper
+
+        sleeper = NativeAwakenableSleeper()
+
+        # Should return after timeout without wake
+        await sleeper.sleep("test", delay_ms=20)
+
+    async def test_native_event(self) -> None:
+        from synapse.util.async_helpers import NativeEvent
+
+        event = NativeEvent()
+        self.assertFalse(event.is_set())
+
+        event.set()
+        self.assertTrue(event.is_set())
+
+        result = await event.wait(timeout_seconds=1.0)
+        self.assertTrue(result)
+
+        event.clear()
+        self.assertFalse(event.is_set())
+
+    async def test_native_event_timeout(self) -> None:
+        from synapse.util.async_helpers import NativeEvent
+
+        event = NativeEvent()
+        result = await event.wait(timeout_seconds=0.02)
+        self.assertFalse(result)
+
+
+class FutureCacheTest(unittest.IsolatedAsyncioTestCase):
+    """Tests for the asyncio-native FutureCache."""
+
+    async def test_set_and_get(self) -> None:
+        from synapse.util.caches.future_cache import FutureCache
+
+        cache: FutureCache[str] = FutureCache("test", max_entries=100)
+
+        loop = asyncio.get_running_loop()
+        f: asyncio.Future[str] = loop.create_future()
+        cache.set("key1", f)
+
+        # Resolve the future
+        f.set_result("value1")
+        await asyncio.sleep(0)  # Let callbacks run
+
+        # Now get should return the cached value
+        result = await cache.get("key1")
+        self.assertEqual(result, "value1")
+
+    async def test_get_pending(self) -> None:
+        from synapse.util.caches.future_cache import FutureCache
+
+        cache: FutureCache[str] = FutureCache("test")
+
+        loop = asyncio.get_running_loop()
+        f: asyncio.Future[str] = loop.create_future()
+        cache.set("key1", f)
+
+        # Get while still pending — returns observer
+        observer = cache.get("key1")
+        self.assertFalse(observer.done())
+
+        # Resolve
+        f.set_result("hello")
+        await asyncio.sleep(0)
+
+        result = await observer
+        self.assertEqual(result, "hello")
+
+    async def test_get_missing_raises_keyerror(self) -> None:
+        from synapse.util.caches.future_cache import FutureCache
+
+        cache: FutureCache[str] = FutureCache("test")
+
+        with self.assertRaises(KeyError):
+            cache.get("nonexistent")
+
+    async def test_failed_future_not_cached(self) -> None:
+        from synapse.util.caches.future_cache import FutureCache
+
+        cache: FutureCache[str] = FutureCache("test")
+
+        loop = asyncio.get_running_loop()
+        f: asyncio.Future[str] = loop.create_future()
+        cache.set("key1", f)
+
+        f.set_exception(ValueError("boom"))
+        await asyncio.sleep(0)
+
+        # Should not be cached
+        with self.assertRaises(KeyError):
+            cache.get("key1")
+
+    async def test_invalidate(self) -> None:
+        from synapse.util.caches.future_cache import FutureCache
+
+        cache: FutureCache[str] = FutureCache("test")
+
+        loop = asyncio.get_running_loop()
+        f: asyncio.Future[str] = loop.create_future()
+        cache.set("key1", f)
+        f.set_result("value1")
+        await asyncio.sleep(0)
+
+        self.assertIn("key1", cache)
+
+        cache.invalidate("key1")
+        self.assertNotIn("key1", cache)
+
+    async def test_invalidation_callback(self) -> None:
+        from synapse.util.caches.future_cache import FutureCache
+
+        cache: FutureCache[str] = FutureCache("test")
+        callback_called = False
+
+        def on_invalidate() -> None:
+            nonlocal callback_called
+            callback_called = True
+
+        loop = asyncio.get_running_loop()
+        f: asyncio.Future[str] = loop.create_future()
+        cache.set("key1", f, callback=on_invalidate)
+        f.set_result("value1")
+        await asyncio.sleep(0)
+
+        cache.invalidate("key1")
+        self.assertTrue(callback_called)
+
+    async def test_invalidate_all(self) -> None:
+        from synapse.util.caches.future_cache import FutureCache
+
+        cache: FutureCache[str] = FutureCache("test")
+
+        loop = asyncio.get_running_loop()
+        for i in range(5):
+            f: asyncio.Future[str] = loop.create_future()
+            cache.set(f"key{i}", f)
+            f.set_result(f"val{i}")
+        await asyncio.sleep(0)
+
+        self.assertEqual(len(cache), 5)
+        cache.invalidate_all()
+        self.assertEqual(len(cache), 0)
+
+    async def test_max_entries_eviction(self) -> None:
+        from synapse.util.caches.future_cache import FutureCache
+
+        cache: FutureCache[int] = FutureCache("test", max_entries=3)
+
+        loop = asyncio.get_running_loop()
+        for i in range(5):
+            f: asyncio.Future[int] = loop.create_future()
+            cache.set(f"key{i}", f)
+            f.set_result(i)
+            await asyncio.sleep(0)
+
+        # Should have evicted oldest entries
+        self.assertLessEqual(len(cache._completed), 3)
+
+    async def test_multiple_observers(self) -> None:
+        from synapse.util.caches.future_cache import FutureCache
+
+        cache: FutureCache[str] = FutureCache("test")
+
+        loop = asyncio.get_running_loop()
+        f: asyncio.Future[str] = loop.create_future()
+        cache.set("key1", f)
+
+        obs1 = cache.get("key1")
+        obs2 = cache.get("key1")
+
+        f.set_result("shared_value")
+        await asyncio.sleep(0)
+
+        r1 = await obs1
+        r2 = await obs2
+        self.assertEqual(r1, "shared_value")
+        self.assertEqual(r2, "shared_value")
+
+
 if __name__ == "__main__":
     unittest.main()