fix(memory): restart the cleanup task on a new event loop and add aclose()

CHANGELOG.md

@@ -27,6 +27,10 @@ Note that 0.3.3 was never released; 0.3.4 follows 0.3.2.
   same `pymemcache` dependency as the old `memcache` extra.
   ([#201](https://github.com/allen0099/FastAPI-CacheX/issues/201))
 
+- **`MemoryBackend.aclose()` stops the cleanup task and waits for it.**
+  `stop_cleanup()` only requests cancellation and stays as it is.
+  ([#181](https://github.com/allen0099/FastAPI-CacheX/issues/181))
+
 ### Changed
 
 - **GitHub release notes list one line per change.** Each changelog entry now
@@ -137,6 +141,13 @@ Note that 0.3.3 was never released; 0.3.4 follows 0.3.2.
   expiry, instead of `SessionExpiredError`.
   ([#164](https://github.com/allen0099/FastAPI-CacheX/issues/164))
 
+- **`MemoryBackend` restarts its cleanup task on a new event loop.** The task
+  stayed tied to the loop of the first cache call. If that loop was closed
+  without cancelling it, a backend reused on another loop never cleaned up
+  again. The task is now started again on the current loop, and a task left
+  on a loop that is still open is cancelled there.
+  ([#181](https://github.com/allen0099/FastAPI-CacheX/issues/181))
+
 ## [0.3.7] - 2026-09-25
 
 ### Added

docs/BACKENDS.md

@@ -34,6 +34,26 @@ BackendProxy.set(backend)
 > The in-memory cache is not suitable for production with multiple processes.
 > Each process maintains its own separate cache.
 
+The cleanup task starts on the event loop of the first cache call. If a later call
+runs on a different loop, for example after the first loop was closed, the task is
+started again there. To stop it on shutdown, `await backend.aclose()` cancels the
+task and waits until it has finished. `stop_cleanup()` only requests cancellation.
+
+```python
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    yield
+    await backend.aclose()
+
+
+app = FastAPI(lifespan=lifespan)
+```
+
 `clear_pattern()` matches whole keys case-sensitively on every platform, like Redis.
 The glob syntax is Python's `fnmatch`, which differs from Redis in two places: negate
 a character class with `[!...]` (Redis uses `[^...]`), and escape a special character

fastapi_cachex/backends/memory.py

@@ -42,6 +42,22 @@ def _is_live(item: CacheItem, now: float) -> bool:
     return item.expiry is None or item.expiry > now
 
 
+def _cancel(task: "asyncio.Task[None]") -> None:
+    """Cancel ``task`` from any thread; a task on a closed loop is left alone."""
+    loop = task.get_loop()
+    if loop.is_closed():
+        # Cancelling would schedule a callback on the closed loop and raise.
+        return
+    try:
+        running = asyncio.get_running_loop()
+    except RuntimeError:
+        running = None
+    if running is loop:
+        task.cancel()
+    else:
+        loop.call_soon_threadsafe(task.cancel)
+
+
 class MemoryBackend(BaseCacheBackend):
     """In-memory cache backend implementation.
 
@@ -71,18 +87,26 @@ def __init__(self, cleanup_interval: int = 60) -> None:
         self._cleanup_task: asyncio.Task[None] | None = None
 
     def _ensure_cleanup_started(self) -> None:
-        """Ensure cleanup task is started in proper async context."""
-        if self._cleanup_task is None or self._cleanup_task.done():
-            try:
-                loop = asyncio.get_running_loop()
-            except RuntimeError:
-                # No running event loop yet; defer until first real async call.
+        """Ensure a cleanup task runs on the current event loop.
+
+        A task left on another loop, for example one that has since been
+        closed, never runs again, so it is replaced rather than reused.
+        """
+        try:
+            loop = asyncio.get_running_loop()
+        except RuntimeError:
+            # No running event loop yet; defer until first real async call.
+            return
+        task = self._cleanup_task
+        if task is not None and not task.done():
+            if task.get_loop() is loop:
                 return
-            self._cleanup_task = loop.create_task(self._cleanup_task_impl())
-            logger.debug(
-                "Started memory backend cleanup task (interval=%s)",
-                self.cleanup_interval,
-            )
+            _cancel(task)
+        self._cleanup_task = loop.create_task(self._cleanup_task_impl())
+        logger.debug(
+            "Started memory backend cleanup task (interval=%s)",
+            self.cleanup_interval,
+        )
 
     def start_cleanup(self) -> None:
         """Start the cleanup task if it's not already running.
@@ -92,12 +116,30 @@ def start_cleanup(self) -> None:
         self._ensure_cleanup_started()
 
     def stop_cleanup(self) -> None:
-        """Stop the cleanup task if it's running."""
+        """Stop the cleanup task if it's running.
+
+        This only requests cancellation. Use ``aclose()`` to also wait until
+        the task has finished.
+        """
         if self._cleanup_task is not None:
-            self._cleanup_task.cancel()
+            _cancel(self._cleanup_task)
             self._cleanup_task = None
             logger.debug("Stopped memory backend cleanup task")
 
+    async def aclose(self) -> None:
+        """Stop the cleanup task and wait until it has finished.
+
+        Safe to call more than once. A task that belongs to another event loop
+        cannot be awaited here; it is only asked to cancel, as ``stop_cleanup()``
+        does.
+        """
+        task = self._cleanup_task
+        self.stop_cleanup()
+        if task is not None and task.get_loop() is asyncio.get_running_loop():
+            # wait() does not raise the task's CancelledError, so a
+            # cancellation of aclose() itself still propagates.
+            await asyncio.wait([task])
+
     async def get(self, key: str) -> CacheEntry | None:
         """Retrieve a cached response.
 

i18n/zh-TW/docs/BACKENDS.md

@@ -28,6 +28,23 @@ BackendProxy.set(backend)
 > [!NOTE]
 > 記憶體快取不適合用於多行程的正式環境。每個行程都各自維護獨立的快取。
 
+清理 task 會在第一次快取呼叫所在的事件迴圈（event loop）上啟動。若之後的呼叫在另一個迴圈上執行（例如第一個迴圈已關閉），task 會在新的迴圈上重新啟動。關閉應用程式時，`await backend.aclose()` 會取消 task 並等待它結束；`stop_cleanup()` 只會要求取消。
+
+```python
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    yield
+    await backend.aclose()
+
+
+app = FastAPI(lifespan=lifespan)
+```
+
 `clear_pattern()` 在所有平台上都以區分大小寫的方式比對完整的鍵，與 Redis 相同。萬用字元語法採用 Python 的 `fnmatch`，與 Redis 有兩處不同：否定字元類別要寫 `[!...]`（Redis 為 `[^...]`）；跳脫特殊字元要放進中括號，例如 `[*]`（Redis 另外也接受 `\*`）。`*`、`?` 與 `[abc]` 在兩者上的行為相同。
 
 ## Redis {#redis}

tests/backends/test_memory.py

@@ -174,6 +174,93 @@ async def test_memory_backend_stop_cleanup_when_not_running(
     assert memory_backend._cleanup_task is None
 
 
+def _run(loop: asyncio.AbstractEventLoop, backend: MemoryBackend) -> None:
+    """Start the backend's cleanup task on ``loop`` and return."""
+
+    async def start() -> None:
+        backend.start_cleanup()
+
+    loop.run_until_complete(start())
+
+
+def _close(loop: asyncio.AbstractEventLoop) -> None:
+    """Close ``loop`` after letting its cancelled tasks finish."""
+    loop.run_until_complete(asyncio.sleep(0))
+    loop.close()
+
+
+def test_cleanup_restarts_on_a_new_loop_after_the_old_one_closed():
+    """A task left on a closed loop never runs; it must be replaced (#181)."""
+    backend = MemoryBackend()
+    first = asyncio.new_event_loop()
+    _run(first, backend)
+    stale = backend._cleanup_task
+    first.close()  # without cancelling the task, as some runners do
+
+    second = asyncio.new_event_loop()
+    try:
+        _run(second, backend)
+        task = backend._cleanup_task
+        assert task is not None
+        assert task is not stale
+        assert task.get_loop() is second
+        backend.stop_cleanup()  # the stale task's loop is closed: must not raise
+    finally:
+        _close(second)
+
+
+def test_cleanup_moving_loops_cancels_the_task_on_a_loop_still_open():
+    backend = MemoryBackend()
+    first = asyncio.new_event_loop()
+    second = asyncio.new_event_loop()
+    try:
+        _run(first, backend)
+        stale = backend._cleanup_task
+        assert stale is not None
+
+        _run(second, backend)
+        first.run_until_complete(asyncio.sleep(0))
+
+        assert stale.cancelled() or stale.done()
+        assert backend._cleanup_task is not stale
+        backend.stop_cleanup()
+    finally:
+        _close(first)
+        _close(second)
+
+
+@pytest.mark.asyncio
+async def test_aclose_waits_for_the_cleanup_task(memory_backend: MemoryBackend):
+    memory_backend.start_cleanup()
+    task = memory_backend._cleanup_task
+    assert task is not None
+
+    await memory_backend.aclose()
+
+    assert task.done()
+    assert memory_backend._cleanup_task is None
+    await memory_backend.aclose()  # a second call is a no-op
+
+
+def test_aclose_only_cancels_a_task_on_another_loop():
+    backend = MemoryBackend()
+    other = asyncio.new_event_loop()
+    current = asyncio.new_event_loop()
+    try:
+        _run(other, backend)
+        task = backend._cleanup_task
+        assert task is not None
+
+        current.run_until_complete(backend.aclose())  # cannot await it there
+
+        assert backend._cleanup_task is None
+        other.run_until_complete(asyncio.sleep(0))
+        assert task.done()
+    finally:
+        _close(other)
+        _close(current)
+
+
 @pytest.mark.asyncio
 async def test_memory_backend_cleanup_task_impl():
     """The sweeper itself has to drop expired entries.