diff --git a/CHANGELOG.md b/CHANGELOG.md index 95002c9..f667a32 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/BACKENDS.md b/docs/BACKENDS.md index 887b7de..d2120ea 100644 --- a/docs/BACKENDS.md +++ b/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 diff --git a/fastapi_cachex/backends/memory.py b/fastapi_cachex/backends/memory.py index 82d93ca..fad2b13 100644 --- a/fastapi_cachex/backends/memory.py +++ b/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. diff --git a/i18n/zh-TW/docs/BACKENDS.md b/i18n/zh-TW/docs/BACKENDS.md index 443d6f1..cf086ac 100644 --- a/i18n/zh-TW/docs/BACKENDS.md +++ b/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} diff --git a/tests/backends/test_memory.py b/tests/backends/test_memory.py index 95e8820..b4fc3d4 100644 --- a/tests/backends/test_memory.py +++ b/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.