#!/usr/bin/env python3
"""Performance benchmarks for Query Cache (Phase 1).
Tests Phase 1 success criteria:
- L1 cache hit rate >30% for typical workflows
- <1ms latency for L1 cache hits
- <10ms latency for L2 cache hits
- Zero memory leaks (no growth over time)
"""
from __future__ import annotations
import time
from unittest.mock import MagicMock
import pytest
from session_buddy.cache.query_cache import QueryCacheManager, QueryCacheEntry
from tests.helpers import PerformanceHelper
class TestQueryCachePerformance:
"""Performance benchmarks for query cache system."""
@pytest.mark.performance
def test_l1_cache_latency(self, perf_helper: PerformanceHelper):
"""Verify L1 cache hits are sub-millisecond.
Success criterion: <1ms latency for L1 cache hits
"""
manager = QueryCacheManager(l1_max_size=1000, l2_ttl_days=7)
manager._initialized = True
# Populate cache with test data
cache_key = "test_key_123"
manager.put(
cache_key=cache_key,
result_ids=["id1", "id2", "id3"],
normalized_query="test query",
project="test_project",
)
# Measure L1 cache hit latency (1000 samples)
latencies = []
for _ in range(1000):
start = time.perf_counter()
result = manager.get(cache_key, check_l2=False)
end = time.perf_counter()
assert result is not None
latencies.append((end - start) * 1000) # Convert to ms
# Calculate statistics
avg_latency = sum(latencies) / len(latencies)
max_latency = max(latencies)
p95_latency = sorted(latencies)[949] # 95th percentile
p99_latency = sorted(latencies)[989] # 99th percentile
# Verify success criteria
assert avg_latency < 1.0, (
f"❌ L1 cache average latency {avg_latency:.3f}ms exceeds 1ms target"
)
assert p95_latency < 1.0, (
f"❌ L1 cache p95 latency {p95_latency:.3f}ms exceeds 1ms target"
)
assert p99_latency < 2.0, (
f"❌ L1 cache p99 latency {p99_latency:.3f}ms exceeds 2ms"
)
print(f"\n✅ L1 Cache Latency Performance:")
print(f" Average: {avg_latency:.3f}ms")
print(f" Max: {max_latency:.3f}ms")
print(f" P95: {p95_latency:.3f}ms")
print(f" P99: {p99_latency:.3f}ms")
@pytest.mark.skip(reason="L2 operations require integration testing with real DuckDB database")
@pytest.mark.performance
def test_l2_cache_latency(self, perf_helper: PerformanceHelper):
"""Verify L2 cache hits are under 10ms.
Success criterion: <10ms latency for L2 cache hits
Note: This uses mocked L2 operations to isolate cache logic from DB performance.
"""
manager = QueryCacheManager(l1_max_size=1000, l2_ttl_days=7)
manager._conn = MagicMock() # Mock DuckDB connection
manager._initialized = True
# Mock L2 get operation to simulate database latency
original_get_from_l2 = manager._get_from_l2
def mock_get_from_l2(cache_key: str):
# Simulate 5ms database latency
time.sleep(0.005)
return QueryCacheEntry(
cache_key=cache_key,
normalized_query="test query",
result_ids=["id1", "id2"],
project="test_project",
)
manager._get_from_l2 = mock_get_from_l2
# Ensure cache miss on L1 to trigger L2 lookup
cache_key = "l2_test_key"
manager.put(
cache_key=cache_key,
result_ids=["id1", "id2"],
normalized_query="test query",
project="test_project",
)
# Clear L1 to force L2 lookup
manager._l1_cache.clear()
# Measure L2 cache hit latency (100 samples)
latencies = []
for _ in range(100):
start = time.perf_counter()
result = manager.get(cache_key, check_l2=True)
end = time.perf_counter()
assert result is not None
latencies.append((end - start) * 1000) # Convert to ms
# Calculate statistics
avg_latency = sum(latencies) / len(latencies)
max_latency = max(latencies)
p95_latency = sorted(latencies)[94] # 95th percentile
# Verify success criteria
assert avg_latency < 10.0, (
f"❌ L2 cache average latency {avg_latency:.3f}ms exceeds 10ms target"
)
assert p95_latency < 15.0, (
f"❌ L2 cache p95 latency {p95_latency:.3f}ms exceeds 15ms"
)
print(f"\n✅ L2 Cache Latency Performance:")
print(f" Average: {avg_latency:.3f}ms")
print(f" Max: {max_latency:.3f}ms")
print(f" P95: {p95_latency:.3f}ms")
@pytest.mark.performance
def test_cache_hit_rate_typical_workflow(self, perf_helper: PerformanceHelper):
"""Verify cache hit rate >30% for typical search workflow.
Simulates realistic usage patterns:
- 60% repeated queries (should hit cache)
- 40% new queries (should miss cache)
Success criterion: >30% overall hit rate
"""
manager = QueryCacheManager(l1_max_size=100, l2_ttl_days=7)
manager._initialized = True
# Define common queries (60% of searches)
common_queries = [
"async patterns",
"error handling",
"authentication",
"database optimization",
"testing strategies",
"API design",
]
# Define rare queries (40% of searches)
def rare_query_generator():
for i in range(1000):
yield f"unique search query {i}"
# Simulate search workflow
total_searches = 1000
cache_hits = 0
for i in range(total_searches):
# 60% common queries, 40% rare queries
if i < 600:
query = common_queries[i % len(common_queries)]
else:
query = next(rare_query_generator())
# Check cache first
cache_key = manager.compute_cache_key(query, project="test_project", limit=10)
cached_result = manager.get(cache_key, check_l2=False)
if cached_result is not None:
cache_hits += 1
else:
# Simulate storing search result
manager.put(
cache_key=cache_key,
result_ids=["id1", "id2"],
normalized_query=manager.normalize_query(query),
project="test_project",
)
# Calculate hit rate
hit_rate = cache_hits / total_searches
# Verify success criteria
assert hit_rate > 0.30, (
f"❌ Cache hit rate {hit_rate:.1%} below 30% target"
)
# Get cache statistics
stats = manager.get_stats()
print(f"\n✅ Cache Hit Rate Performance:")
print(f" Overall hit rate: {hit_rate:.1%}")
print(f" L1 hits: {stats['l1_hits']}")
print(f" L1 misses: {stats['l1_misses']}")
print(f" L1 hit rate: {stats['l1_hit_rate']:.1%}")
print(f" L1 size: {stats['l1_size']} entries")
@pytest.mark.performance
def test_memory_stability_no_leaks(self, perf_helper: PerformanceHelper):
"""Verify no memory leaks during cache operations.
Success criterion: No unbounded memory growth over time
"""
manager = QueryCacheManager(l1_max_size=1000, l2_ttl_days=7)
manager._initialized = True
# Record initial memory state
initial_l1_size = len(manager._l1_cache)
# Perform many cache operations
for i in range(10000):
cache_key = f"memory_test_key_{i % 500}" # Only 500 unique keys
manager.put(
cache_key=cache_key,
result_ids=["id1", "id2"],
normalized_query=f"query {i % 500}",
project="test_project",
)
# Randomly access entries
if i % 10 == 0:
manager.get(f"memory_test_key_{i % 100}")
# Verify L1 cache size is bounded
final_l1_size = len(manager._l1_cache)
# L1 should not exceed max_size (LRU eviction should work)
assert final_l1_size <= manager.l1_max_size, (
f"❌ L1 cache grew to {final_l1_size} entries, exceeding max of {manager.l1_max_size}"
)
# Get statistics
stats = manager.get_stats()
print(f"\n✅ Memory Stability Verification:")
print(f" Initial L1 size: {initial_l1_size}")
print(f" Final L1 size: {final_l1_size}")
print(f" L1 max size: {manager.l1_max_size}")
print(f" L1 evictions: {stats['l1_evictions']}")
print(f" ✅ No memory leaks detected - LRU eviction working correctly")
@pytest.mark.performance
def test_cache_key_computation_performance(self, perf_helper: PerformanceHelper):
"""Verify cache key computation is fast and consistent.
Cache key computation should be <100µs for efficient caching.
"""
manager = QueryCacheManager()
# Measure cache key computation performance (10000 samples)
latencies = []
for i in range(10000):
start = time.perf_counter()
key = manager.compute_cache_key(
query=f"search query {i % 100}",
project="test_project",
limit=10,
)
end = time.perf_counter()
latencies.append((end - start) * 1_000_000) # Convert to microseconds
# Calculate statistics
avg_latency_us = sum(latencies) / len(latencies)
max_latency_us = max(latencies)
p99_latency_us = sorted(latencies)[9899] # 99th percentile
# Verify performance requirements
assert avg_latency_us < 100, (
f"❌ Cache key computation average {avg_latency_us:.1f}µs exceeds 100µs target"
)
assert p99_latency_us < 500, (
f"❌ Cache key computation p99 {p99_latency_us:.1f}µs exceeds 500µs"
)
print(f"\n✅ Cache Key Computation Performance:")
print(f" Average: {avg_latency_us:.1f}µs")
print(f" Max: {max_latency_us:.1f}µs")
print(f" P99: {p99_latency_us:.1f}µs")
@pytest.mark.performance
def test_lru_eviction_performance(self, perf_helper: PerformanceHelper):
"""Verify LRU eviction maintains cache performance under pressure.
When cache is full, eviction should be fast and maintain hit rate.
"""
manager = QueryCacheManager(l1_max_size=100, l2_ttl_days=7)
manager._initialized = True
# Fill cache to max capacity
for i in range(100):
cache_key = f"eviction_test_key_{i}"
manager.put(
cache_key=cache_key,
result_ids=[f"id{i}"],
normalized_query=f"query {i}",
project="test_project",
)
# Access first 50 entries to make them recently used
for i in range(50):
manager.get(f"eviction_test_key_{i}", check_l2=False)
# Add 50 more entries - should evict least recently used
for i in range(100, 150):
cache_key = f"eviction_test_key_{i}"
manager.put(
cache_key=cache_key,
result_ids=[f"id{i}"],
normalized_query=f"query {i}",
project="test_project",
)
# Verify first 50 entries still cached (recently accessed)
for i in range(50):
result = manager.get(f"eviction_test_key_{i}", check_l2=False)
assert result is not None, f"Entry {i} should still be in cache"
# Verify entries 50-99 were evicted (least recently used)
evicted_count = 0
for i in range(50, 100):
result = manager.get(f"eviction_test_key_{i}", check_l2=False)
if result is None:
evicted_count += 1
assert evicted_count == 50, f"Expected 50 evictions, got {evicted_count}"
# Verify new entries 100-149 are cached
for i in range(100, 150):
result = manager.get(f"eviction_test_key_{i}", check_l2=False)
assert result is not None, f"Entry {i} should be in cache"
# Check statistics
stats = manager.get_stats()
print(f"\n✅ LRU Eviction Performance:")
print(f" L1 evictions: {stats['l1_evictions']}")
print(f" L1 size: {stats['l1_size']}")
print(f" ✅ LRU eviction working correctly")
