6p2ny redis (#4224)

* Refactor cache to use lazy init and fix module hierarchy

- Move cache_manager.py and redis_pubsub.py from utils/ to database/
  to follow module hierarchy (database → utils → routers → main)
- Replace startup/shutdown event handlers with lazy initialization
  and atexit cleanup for simpler usage (just import and use)
- Remove init_cache/shutdown_cache from main.py and pusher/main.py

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* Add singleflight pattern to prevent thundering herd

- Add get_or_fetch() method with lock-per-key singleflight pattern
- Only ONE concurrent request fetches from Redis, others wait
- Update get_popular_apps, get_available_apps, get_approved_available_apps
- Add tests for concurrent access and cache hit scenarios

With 1000 concurrent requests on cold cache: only 1 hits Redis (was 1000)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>

Author: beastoin

backend/database/cache.py

@@ -1,79 +1,65 @@
 """
 Global cache instances for in-memory caching.
 
-This module provides singleton access to cache managers.
-Instances are initialized during application startup in main.py.
+Uses lazy initialization - caches are created on first access.
+Cleanup is handled automatically via atexit.
 """
 
+import atexit
 from typing import Optional
-from utils.cache_manager import InMemoryCacheManager
-from utils.redis_pubsub import RedisPubSubManager
+from database.cache_manager import InMemoryCacheManager
+from database.redis_pubsub import RedisPubSubManager
 from database.redis_db import r
 
-# Global cache instances (initialized in main.py via init_cache())
-memory_cache: Optional[InMemoryCacheManager] = None
-pubsub_manager: Optional[RedisPubSubManager] = None
+# Global cache instances (lazily initialized)
+_memory_cache: Optional[InMemoryCacheManager] = None
+_pubsub_manager: Optional[RedisPubSubManager] = None
+_initialized: bool = False
 
 
-def get_memory_cache() -> InMemoryCacheManager:
-    """
-    Get the global memory cache instance.
-
-    Returns:
-        The initialized memory cache instance
-
-    Raises:
-        RuntimeError: If cache not initialized
-    """
-    if memory_cache is None:
-        raise RuntimeError("Memory cache not initialized. Call init_cache() first.")
-    return memory_cache
-
-
-def get_pubsub_manager() -> RedisPubSubManager:
-    """
-    Get the global pub/sub manager instance.
-
-    Returns:
-        The initialized pub/sub manager instance
-
-    Raises:
-        RuntimeError: If pub/sub manager not initialized
-    """
-    if pubsub_manager is None:
-        raise RuntimeError("Pub/sub manager not initialized. Call init_cache() first.")
-    return pubsub_manager
-
-
-def init_cache(max_memory_mb: int = 100):
-    """
-    Initialize global cache instances.
+def _ensure_initialized():
+    """Initialize caches on first access."""
+    global _memory_cache, _pubsub_manager, _initialized
 
-    Should be called once during application startup in main.py.
+    if _initialized:
+        return
 
-    Args:
-        max_memory_mb: Maximum memory in MB for in-memory cache (default: 100MB)
-    """
-    global memory_cache, pubsub_manager
-
-    memory_cache = InMemoryCacheManager(max_memory_mb=max_memory_mb)
-    pubsub_manager = RedisPubSubManager(r)
+    _memory_cache = InMemoryCacheManager(max_memory_mb=100)
+    _pubsub_manager = RedisPubSubManager(r)
 
     # Register callbacks: when invalidation message received, clear memory cache
-    pubsub_manager.register_callback(
+    _pubsub_manager.register_callback(
         'get_public_approved_apps_data*',
-        lambda keys: [memory_cache.delete(k) for k in keys]
+        lambda keys: [_memory_cache.delete(k) for k in keys]
     )
-    pubsub_manager.register_callback(
+    _pubsub_manager.register_callback(
         'get_popular_apps_data',
-        lambda keys: [memory_cache.delete(k) for k in keys]
+        lambda keys: [_memory_cache.delete(k) for k in keys]
     )
 
     # Start pub/sub subscription
-    pubsub_manager.start()
+    _pubsub_manager.start()
+    _initialized = True
+
+
+def get_memory_cache() -> InMemoryCacheManager:
+    """Get the global memory cache instance (lazy init)."""
+    _ensure_initialized()
+    return _memory_cache
+
+
+def get_pubsub_manager() -> RedisPubSubManager:
+    """Get the global pub/sub manager instance (lazy init)."""
+    _ensure_initialized()
+    return _pubsub_manager
+
+
+def _shutdown():
+    """Cleanup on process exit."""
+    global _pubsub_manager
+    if _pubsub_manager:
+        _pubsub_manager.stop()
 
 
-def shutdown_cache():
-    """Shutdown cache managers gracefully."""
-    if pubsub_manager:
-        pubsub_manager.stop()
+# Register cleanup handler
+atexit.register(_shutdown)

backend/utils/cache_manager.py backend/database/cache_manager.pybackend/utils/cache_manager.py renamed to backend/database/cache_manager.py

@@ -14,7 +14,7 @@
 import time
 from collections import OrderedDict
 from dataclasses import dataclass
-from typing import Any, Optional
+from typing import Any, Callable, Dict, Optional
 
 
 @dataclass
@@ -59,6 +59,10 @@ def __init__(self, max_memory_mb: int = 100):
         self.misses = 0
         self.evictions = 0
 
+        # Singleflight: per-key locks to prevent thundering herd
+        self._fetch_locks: Dict[str, threading.Lock] = {}
+        self._fetch_lock_manager = threading.Lock()
+
     def get(self, key: str) -> Optional[Any]:
         """
         Get cache entry if exists and not expired.
@@ -87,6 +91,43 @@ def get(self, key: str) -> Optional[Any]:
             self.hits += 1
             return entry.data
 
+    def get_or_fetch(self, key: str, fetch_fn: Callable[[], Any], ttl: int = 30) -> Any:
+        """
+        Get from cache or fetch with singleflight pattern.
+
+        Only ONE concurrent request will call fetch_fn, others wait.
+        This prevents the thundering herd problem.
+
+        Args:
+            key: Cache key
+            fetch_fn: Function to call if cache miss (should return data)
+            ttl: Time to live in seconds (default: 30)
+
+        Returns:
+            Cached or fetched data
+        """
+        # Fast path: cache hit
+        if (value := self.get(key)) is not None:
+            return value
+
+        # Get or create lock for this key
+        with self._fetch_lock_manager:
+            if key not in self._fetch_locks:
+                self._fetch_locks[key] = threading.Lock()
+            fetch_lock = self._fetch_locks[key]
+
+        # Only one request fetches, others wait
+        with fetch_lock:
+            # Double-check after acquiring lock (another thread may have fetched)
+            if (value := self.get(key)) is not None:
+                return value
+
+            # Fetch and cache
+            value = fetch_fn()
+            if value is not None:
+                self.set(key, value, ttl=ttl)
+            return value
+
     def set(self, key: str, data: Any, ttl: int = 30):
         """
         Set cache entry with automatic eviction if needed.

backend/utils/redis_pubsub.py backend/database/redis_pubsub.pybackend/utils/redis_pubsub.py renamed to backend/database/redis_pubsub.py

@@ -43,7 +43,6 @@
 
 from utils.other.timeout import TimeoutMiddleware
 from utils.observability import log_langsmith_status
-from database.cache import init_cache, shutdown_cache
 
 # Log LangSmith tracing status at startup
 log_langsmith_status()
@@ -108,24 +107,6 @@
 app.add_middleware(TimeoutMiddleware, methods_timeout=methods_timeout)
 
 
-# Startup and shutdown event handlers for cache management
-@app.on_event("startup")
-async def startup_event():
-    """Initialize cache managers on startup."""
-    try:
-        init_cache(max_memory_mb=100)
-        print("Cache managers initialized successfully")
-    except Exception as e:
-        print(f"Failed to initialize cache managers: {e}")
-        # Continue startup even if cache managers fail
-
-
-@app.on_event("shutdown")
-async def shutdown_event():
-    """Stop cache managers on shutdown."""
-    shutdown_cache()
-    print("Cache managers stopped")
-
 modal_app = App(
     name='backend',
     secrets=[Secret.from_name("gcp-credentials"), Secret.from_name('envs')],

backend/main.py

@@ -5,7 +5,6 @@
 from fastapi import FastAPI
 
 from routers import pusher
-from database.cache import init_cache, shutdown_cache
 
 if os.environ.get('SERVICE_ACCOUNT_JSON'):
     service_account_info = json.loads(os.environ["SERVICE_ACCOUNT_JSON"])
@@ -26,21 +25,3 @@
 @app.get('/health')
 def health_check():
     return {"status": "healthy"}
-
-
-@app.on_event("startup")
-async def startup_event():
-    """Initialize cache managers on startup."""
-    try:
-        init_cache(max_memory_mb=100)
-        print("Cache managers initialized successfully")
-    except Exception as e:
-        print(f"Failed to initialize cache managers: {e}")
-        # Continue startup even if cache managers fail
-
-
-@app.on_event("shutdown")
-async def shutdown_event():
-    """Stop cache managers on shutdown."""
-    shutdown_cache()
-    print("Cache managers stopped")

backend/pusher/main.py

@@ -10,7 +10,7 @@
 # Add parent directory to path
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
-from utils.cache_manager import InMemoryCacheManager
+from database.cache_manager import InMemoryCacheManager
 
 
 class TestInMemoryCacheManager(unittest.TestCase):
@@ -148,6 +148,66 @@ def test_lru_ordering(self):
         stats = self.cache.get_stats()
         self.assertGreater(stats['entries'], 0)
 
+    def test_singleflight_prevents_thundering_herd(self):
+        """Test that get_or_fetch prevents multiple concurrent fetches."""
+        import threading
+        import time
+
+        fetch_count = 0
+        fetch_lock = threading.Lock()
+
+        def slow_fetch():
+            """Simulate a slow fetch that takes 100ms."""
+            nonlocal fetch_count
+            with fetch_lock:
+                fetch_count += 1
+            time.sleep(0.1)  # Simulate slow operation
+            return {'data': 'fetched'}
+
+        results = []
+        errors = []
+
+        def worker():
+            try:
+                result = self.cache.get_or_fetch('test_key', slow_fetch, ttl=30)
+                results.append(result)
+            except Exception as e:
+                errors.append(e)
+
+        # Launch 10 concurrent threads
+        threads = [threading.Thread(target=worker) for _ in range(10)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join()
+
+        # All threads should get the same result
+        self.assertEqual(len(results), 10)
+        self.assertEqual(len(errors), 0)
+        for result in results:
+            self.assertEqual(result, {'data': 'fetched'})
+
+        # Only ONE fetch should have been called (singleflight)
+        self.assertEqual(fetch_count, 1)
+
+    def test_get_or_fetch_cache_hit(self):
+        """Test that get_or_fetch returns cached data without calling fetch_fn."""
+        fetch_called = False
+
+        def fetch_fn():
+            nonlocal fetch_called
+            fetch_called = True
+            return {'data': 'new'}
+
+        # Pre-populate cache
+        self.cache.set('existing_key', {'data': 'cached'}, ttl=30)
+
+        # get_or_fetch should return cached data
+        result = self.cache.get_or_fetch('existing_key', fetch_fn, ttl=30)
+
+        self.assertEqual(result, {'data': 'cached'})
+        self.assertFalse(fetch_called)
+
 
 if __name__ == '__main__':
     unittest.main()