How to Implement Probabilistic Early Expiration in Python

A Python mock of probabilistic early expiration for caches, using a heap-based expiry queue and random eviction to approximate cache stampede protection.

Medium Python 3.9+ Aug 9, 2026 Caching & Redis 14 views 0 copies

Python code

54 lines
Python 3.9+
import heapq
import random
import time


class ProbabilisticEarlyExpirationMock:
    def __init__(self, capacity=1024, expiration_probability=0.1):
        self.capacity = capacity
        self.expiration_probability = expiration_probability
        self._items = {}
        self._expiry_heap = []
        self._next_id = 0

    def put(self, value):
        if len(self._items) >= self.capacity:
            self._evict_expired()
        if len(self._items) >= self.capacity:
            self._evict_random()
        item_id = self._next_id
        self._next_id += 1
        expires_at = time.monotonic() + random.uniform(0.1, 1.0)
        self._items[item_id] = (value, expires_at)
        heapq.heappush(self._expiry_heap, (expires_at, item_id))
        return item_id

    def get(self, item_id):
        if item_id not in self._items:
            return None
        value, expires_at = self._items[item_id]
        if time.monotonic() >= expires_at:
            del self._items[item_id]
            return None
        if random.random() < self.expiration_probability:
            del self._items[item_id]
            return None
        return value

    def _evict_expired(self):
        now = time.monotonic()
        while self._expiry_heap and self._expiry_heap[0][0] <= now:
            _, item_id = heapq.heappop(self._expiry_heap)
            self._items.pop(item_id, None)

    def _evict_random(self):
        if not self._items:
            return
        item_id = random.choice(list(self._items.keys()))
        del self._items[item_id]


if __name__ == "__main__":
    cache = ProbabilisticEarlyExpirationMock(capacity=5, expiration_probability=0.5)
    ids = [cache.put(i) for i in range(5)]
    print([cache.get(i) for i in ids])

Output

stdout
[0, 1, 2, 3, 4] or [0, 1, None, 3, 4] (probabilistic removal can drop an item before its nominal expiry)

How it works

This mock simulates the 'probabilistic early expiration' strategy Redis uses to avoid cache stampedes. Items are given a random expiry between 0.1 and 1.0 seconds, and get has an expiration_probability chance of treating an item as stale even before its real expiry. put maintains a heap of expirations for O(log n) eviction of truly expired items, and falls back to random eviction only when the capacity is full. The time.monotonic() calls are immune to system clock changes, so the mock behaves predictably in tests.

Common mistakes

  • Using `random.choice(list(self._items.keys()))` on large caches creates a full copy of keys, hurting performance.
  • Forgetting to pop expired entries from the heap in `get`, leaving stale heap entries that slow future evictions.
  • Using `time.time()` instead of `time.monotonic()` — wall-clock changes (NTP, DST) break expiry logic.

Variations

  1. Replace random expiration with a fixed probability constant to make behavior deterministic in unit tests.
  2. Use a `sortedcontainers.SortedList` instead of a heap for expiry management if you need to delete arbitrary entries.

Real-world use cases

  • Protecting a database-backed cache from stampede when 1000s of requests miss the same hot key at once.
  • Simulating Redis' `PROBABILISTIC_EARLY_EXPIRATION` behavior in an offline load-test harness.
  • Teaching cache invalidation strategies in infrastructure workshops with a lightweight, dependency-free model.

Sponsored

Run this sample

Open the browser IDE to tweak the example and see results without installing anything.

Open editor

More from Caching & Redis

Related tutorials and quizzes for this topic.