Async Cache Invalidation: What to Do When a Key Is Invalidated Mid-Fetch

CPython-only mid-fetch invalidation demo

Run this example in native CPython, not this site's browser compiler. The browser editor's synchronous execution path cannot await this asyncio.run(main()) program; no browser Run button is attached to it.

Expected CPython output: “fetching” prints twice, and both callers get value-for-user:7-fetch-2. The first fetch's result was discarded; the second fetch's result is the one served. The corrected example also shields shared work from a cancelled waiter and removes an in-flight entry only when that entry still belongs to the completing fetch.

import asyncio

class SingleFlightCache:
    """Share each generation's fetch; invalidation always wins."""

    def __init__(self):
        self._data = {}
        self._inflight = {}
        self._version = {}

    def invalidate(self, key):
        self._version[key] = self._version.get(key, 0) + 1
        self._data.pop(key, None)
        # Let the old fetch finish, but allow a new generation to start.
        self._inflight.pop(key, None)

    async def _fetch(self, key, loader, version):
        task = asyncio.current_task()
        try:
            value = await loader(key)
            if self._version.get(key, 0) == version:
                self._data[key] = value
        finally:
            # An old fetch must not remove a newer generation's task.
            if self._inflight.get(key) is task:
                self._inflight.pop(key, None)

    async def get(self, key, loader):
        while True:
            if key in self._data:
                return self._data[key]
            task = self._inflight.get(key)
            if task is None:
                version = self._version.get(key, 0)
                task = asyncio.create_task(self._fetch(key, loader, version))
                self._inflight[key] = task
                # Retrieve failures even if every waiter is cancelled.
                task.add_done_callback(
                    lambda done: done.exception() if not done.cancelled() else None
                )
            await asyncio.shield(task)
            # Never serve the awaited value: re-check the current cache.

counter = {"n": 0}

async def slow_loader(key):
    counter["n"] += 1
    fetch_number = counter["n"]
    print(f"  fetching {key} (fetch #{fetch_number}) ...")
    await asyncio.sleep(0.5)
    return f"value-for-{key}-fetch-{fetch_number}"

async def main():
    cache = SingleFlightCache()
    t1 = asyncio.create_task(cache.get("user:7", slow_loader))
    t2 = asyncio.create_task(cache.get("user:7", slow_loader))
    await asyncio.sleep(0.1)
    print("invalidating user:7 mid-flight")
    cache.invalidate("user:7")
    print("caller 1 got:", await t1)
    print("caller 2 got:", await t2)

asyncio.run(main())

  fetching user:7 (fetch #1) ...
invalidating user:7 mid-flight
  fetching user:7 (fetch #2) ...
caller 1 got: value-for-user:7-fetch-2
caller 2 got: value-for-user:7-fetch-2

Why single-flight needs a generation check

The pattern is called single-flight or request coalescing: concurrent callers for the same key share one in-flight fetch, represented by one asyncio.Task, instead of fetching separately. Sharing work does not mean its result is still valid when it arrives.

Three rules keep this example consistent: never store after the key's version changes; make every waiter re-check the cache after the shared task finishes; and do not cancel shared work just to invalidate its value. Cancelling a shared task raises into its waiters and can interrupt work with side effects. Discarding stale results avoids that propagation.

Await through asyncio.shield so cancellation of one waiting caller does not cancel the fetch used by other callers. Shielding does not make the waiting caller immune to cancellation: that caller still receives CancelledError.

Let the fetch clean up its own entry

A failed fetch is not cached, so the next get retries. A try/finally inside the fetch removes its entry even when the loader raises. Leaving a failed task in the map would make future callers re-raise the same old error.

Cleanup must compare task identity before removing the entry. After invalidation, the old fetch may still be running when a fresh fetch starts. An unconditional pop from the old task could delete that fresh entry and allow duplicate work. The version check prevents stale writes; the identity check protects fresh in-flight work.

The loader captures its fetch number before sleeping so overlapping generations cannot accidentally label an old result with a newer counter. This demonstration has no TTL, eviction policy, or distributed synchronization; each event-loop operation between await points is single-threaded.