#!/usr/bin/env python3 """ Redis cache-aside demo server. Run this file and visit http://localhost:8080 to read records through a Redis cache that sits in front of a deliberately slow primary store. The page shows hit/miss counters, measured Redis latency, and a button that fires concurrent reads at a freshly-invalidated key to demonstrate the single-flight stampede protection in ``cache.py``. """ from __future__ import annotations import argparse from concurrent.futures import ThreadPoolExecutor from html import escape from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import json from pathlib import Path import sys import time from urllib.parse import parse_qs, urlparse sys.path.insert(0, str(Path(__file__).resolve().parent)) try: import redis from cache import RedisCache from primary import MockPrimaryStore except ImportError as exc: print(f"Error: {exc}") print("Make sure the 'redis' package is installed: pip install redis") sys.exit(1) class CacheAsideDemoHandler(BaseHTTPRequestHandler): """Serve a small interactive page that uses Redis as a cache-aside layer.""" cache: RedisCache | None = None primary: MockPrimaryStore | None = None def do_GET(self) -> None: parsed = urlparse(self.path) if parsed.path in {"/", "/index.html"}: self._render_index() return if parsed.path == "/products": self._send_json({"products": self.primary.list_ids()}, 200) return if parsed.path == "/read": self._handle_read(parse_qs(parsed.query)) return if parsed.path == "/stats": self._send_json(self._build_stats(), 200) return self.send_error(404) def do_POST(self) -> None: parsed = urlparse(self.path) if parsed.path == "/invalidate": self._handle_invalidate() return if parsed.path == "/update": self._handle_update() return if parsed.path == "/stampede": self._handle_stampede() return if parsed.path == "/reset": self.cache.reset_stats() self.primary.reset_reads() self._send_json(self._build_stats(), 200) return self.send_error(404) def _render_index(self) -> None: self._send_html(self._html_page()) def _handle_read(self, query: dict[str, list[str]]) -> None: product_id = query.get("id", [""])[0] if not product_id: self._send_json({"error": "Missing 'id' query parameter."}, 400) return started = time.perf_counter() record, hit, redis_ms = self.cache.get(product_id, self.primary.read) total_ms = (time.perf_counter() - started) * 1000.0 if record is None: self._send_json({"error": f"No record for '{product_id}'."}, 404) return self._send_json( { "id": product_id, "record": record, "hit": hit, "redis_latency_ms": round(redis_ms, 2), "total_latency_ms": round(total_ms, 2), "ttl_remaining": self.cache.ttl_remaining(product_id), "stats": self._build_stats(), }, 200, ) def _handle_invalidate(self) -> None: params = self._read_form_data() product_id = params.get("id", [""])[0] if not product_id: self._send_json({"error": "Missing 'id'."}, 400) return deleted = self.cache.invalidate(product_id) self._send_json( {"id": product_id, "deleted": deleted, "stats": self._build_stats()}, 200, ) def _handle_update(self) -> None: params = self._read_form_data() product_id = params.get("id", [""])[0] field = params.get("field", [""])[0] value = params.get("value", [""])[0] if not product_id or not field: self._send_json({"error": "Missing 'id' or 'field'."}, 400) return # Cache-aside on writes: update the primary first, then invalidate # the cache so the next read repopulates from the new source value. if not self.primary.update_field(product_id, field, value): self._send_json({"error": "Unknown product."}, 404) return self.cache.invalidate(product_id) self._send_json( {"id": product_id, "field": field, "value": value, "stats": self._build_stats()}, 200, ) def _handle_stampede(self) -> None: params = self._read_form_data() product_id = params.get("id", [""])[0] try: concurrency = max(2, min(50, int(params.get("concurrency", ["20"])[0]))) except ValueError: concurrency = 20 if not product_id: self._send_json({"error": "Missing 'id'."}, 400) return # Start clean so the test demonstrates a true cold-cache stampede. self.cache.invalidate(product_id) primary_reads_before = self.primary.reads() results: list[dict] = [] started = time.perf_counter() with ThreadPoolExecutor(max_workers=concurrency) as pool: futures = [ pool.submit(self.cache.get, product_id, self.primary.read) for _ in range(concurrency) ] for future in futures: record, hit, redis_ms = future.result() results.append( { "hit": hit, "redis_latency_ms": round(redis_ms, 2), "found": record is not None, } ) elapsed_ms = (time.perf_counter() - started) * 1000.0 primary_reads_during = self.primary.reads() - primary_reads_before self._send_json( { "id": product_id, "concurrency": concurrency, "primary_reads": primary_reads_during, "elapsed_ms": round(elapsed_ms, 2), "results": results, "stats": self._build_stats(), }, 200, ) def _build_stats(self) -> dict: stats = self.cache.stats() stats["primary_reads_total"] = self.primary.reads() stats["primary_read_latency_ms"] = self.primary.read_latency_ms return stats def _read_form_data(self) -> dict[str, list[str]]: content_length = int(self.headers.get("Content-Length", "0")) raw_body = self.rfile.read(content_length).decode("utf-8") return parse_qs(raw_body) def _send_html(self, html: str, status: int = 200) -> None: self.send_response(status) self.send_header("Content-Type", "text/html; charset=utf-8") self.end_headers() self.wfile.write(html.encode("utf-8")) def _send_json(self, payload: dict, status: int) -> None: self.send_response(status) self.send_header("Content-Type", "application/json") self.end_headers() self.wfile.write(json.dumps(payload).encode("utf-8")) def log_message(self, format: str, *args) -> None: # noqa: A002 sys.stderr.write(f"[demo] {format % args}\n") def _html_page(self) -> str: product_options = "".join( f'' for pid in self.primary.list_ids() ) primary_latency = self.primary.read_latency_ms cache_ttl = self.cache.ttl return f""" Redis Cache-Aside Demo
redis-py + Python standard library HTTP server

Redis Cache-Aside Demo

Read product records through Redis. The first read of any key falls through to a deliberately slow primary store ({primary_latency} ms per read); subsequent reads come from Redis until the {cache_ttl}-second TTL expires or the entry is invalidated. The stampede test fires concurrent reads at a cold key to show a single-flight Lua lock funnelling them down to one primary read.

Read a product

Read once to populate the cache, then again to see the hit. Wait for the TTL to pass or click Invalidate to force a miss.

Update a field

Updating writes to the primary and deletes the cache entry, so the next read sees the new value.

Stampede test

Invalidates the selected key, then fires N concurrent reads. With single-flight enabled, only one of those reads should hit the primary.

Cache stats

Loading...

Last result

Read a product to see the cached record and timing.

""" def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Run the Redis cache-aside demo server.") parser.add_argument("--host", default="127.0.0.1", help="HTTP bind host") parser.add_argument("--port", type=int, default=8080, help="HTTP bind port") parser.add_argument("--redis-host", default="localhost", help="Redis host") parser.add_argument("--redis-port", type=int, default=6379, help="Redis port") parser.add_argument("--ttl", type=int, default=30, help="Cache TTL in seconds") parser.add_argument( "--primary-latency-ms", type=int, default=150, help="Simulated primary store read latency in milliseconds", ) return parser.parse_args() def main() -> None: args = parse_args() redis_client = redis.Redis( host=args.redis_host, port=args.redis_port, decode_responses=True, ) CacheAsideDemoHandler.cache = RedisCache(redis_client=redis_client, ttl=args.ttl) CacheAsideDemoHandler.primary = MockPrimaryStore(read_latency_ms=args.primary_latency_ms) server = ThreadingHTTPServer((args.host, args.port), CacheAsideDemoHandler) print(f"Redis cache-aside demo server listening on http://{args.host}:{args.port}") print(f"Using Redis at {args.redis_host}:{args.redis_port} with cache TTL {args.ttl}s") print(f"Mock primary read latency: {args.primary_latency_ms} ms") server.serve_forever() if __name__ == "__main__": main()