Cache Invalidation and Consistency

class CacheAsideWithInvalidation:
    def __init__(self, cache, database, invalidation_service):
        self.cache = cache
        self.database = database
        self.invalidation_service = invalidation_service
 
    def get(self, key: str):
        """Get data with cache-aside pattern"""
        # Try cache first
        cached_value = self.cache.get(key)
        if cached_value is not None:
            return cached_value
 
        # Load from database
        value = self.database.get(key)
        if value is not None:
            # Cache the value with dependency tracking
            self.cache.set(key, value)
            self.invalidation_service.track_dependency(key, value.dependencies)
 
        return value
 
    def update(self, key: str, new_value):
        """Update with intelligent invalidation"""
        # Update database first
        old_value = self.database.get(key)
        self.database.set(key, new_value)
 
        # Calculate invalidation scope
        invalidation_keys = self.calculate_invalidation_scope(
            key, old_value, new_value
        )
 
        # Perform cascading invalidation
        self.invalidation_service.invalidate_cascade(invalidation_keys)
 
    def calculate_invalidation_scope(self, key, old_value, new_value):
        """Determine what needs to be invalidated"""
        invalidation_keys = [key]  # Always invalidate the primary key
 
        # Check for dependent data changes
        if old_value and hasattr(old_value, 'dependencies'):
            for dep_key in old_value.dependencies:
                if self.dependency_affected(dep_key, old_value, new_value):
                    invalidation_keys.append(dep_key)
 
        # Check for derived data invalidation
        derived_keys = self.find_derived_keys(key)
        invalidation_keys.extend(derived_keys)
 
        return invalidation_keys

interface TTLStrategy {
  static_ttl: {
    description: 'Fixed expiration time for all cache entries';
    use_cases: ['reference_data', 'configuration'];
    implementation: {
      ttl_seconds: number;
      refresh_strategy: 'lazy' | 'proactive';
    };
  };
 
  dynamic_ttl: {
    description: 'Variable expiration based on data characteristics';
    use_cases: ['user_preferences', 'session_data'];
    implementation: {
      base_ttl: number;
      ttl_calculator: (data: any) => number;
      factors: ['access_frequency', 'data_volatility', 'business_criticality'];
    };
  };
 
  adaptive_ttl: {
    description: 'ML-driven TTL optimization based on access patterns';
    use_cases: ['personalized_content', 'recommendations'];
    implementation: {
      learning_algorithm: 'reinforcement_learning' | 'time_series_analysis';
      feedback_metrics: ['hit_ratio', 'staleness_tolerance', 'update_frequency'];
    };
  };
}

class EventDrivenInvalidation:
    def __init__(self, event_bus, cache_manager):
        self.event_bus = event_bus
        self.cache_manager = cache_manager
        self.dependency_graph = DependencyGraph()
        self.setup_event_handlers()
 
    def setup_event_handlers(self):
        """Set up event handlers for different data change events"""
        self.event_bus.subscribe('user.profile.updated', self.handle_user_profile_update)
        self.event_bus.subscribe('memory.conversation.added', self.handle_conversation_update)
        self.event_bus.subscribe('system.configuration.changed', self.handle_config_change)
 
    def handle_user_profile_update(self, event: UserProfileUpdateEvent):
        """Handle user profile updates with cascading invalidation"""
        user_id = event.user_id
        changed_fields = event.changed_fields
 
        # Direct invalidation
        direct_keys = [
            f"user:profile:{user_id}",
            f"user:preferences:{user_id}"
        ]
 
        # Conditional invalidation based on changed fields
        if 'language' in changed_fields:
            direct_keys.extend([
                f"user:recommendations:{user_id}",
                f"user:content_locale:{user_id}"
            ])
 
        if 'subscription_tier' in changed_fields:
            direct_keys.extend([
                f"user:permissions:{user_id}",
                f"user:features:{user_id}"
            ])
 
        # Find dependent keys using dependency graph
        dependent_keys = self.dependency_graph.find_dependents(direct_keys)
 
        # Perform invalidation with ordering
        self.cache_manager.invalidate_ordered(direct_keys + dependent_keys)
 
    def handle_conversation_update(self, event: ConversationUpdateEvent):
        """Handle memory conversation updates"""
        conversation_id = event.conversation_id
        user_id = event.user_id
        update_type = event.update_type
 
        invalidation_keys = []
 
        if update_type == 'new_message':
            invalidation_keys = [
                f"conversation:{conversation_id}",
                f"user:recent_conversations:{user_id}",
                f"user:context_summary:{user_id}"
            ]
        elif update_type == 'memory_extract':
            invalidation_keys = [
                f"user:memory_index:{user_id}",
                f"user:knowledge_graph:{user_id}",
                f"conversation:insights:{conversation_id}"
            ]
 
        self.cache_manager.invalidate_async(invalidation_keys)

class DependencyGraph:
    def __init__(self):
        self.dependencies = defaultdict(set)
        self.reverse_dependencies = defaultdict(set)
 
    def add_dependency(self, dependent_key: str, dependency_key: str):
        """Add a dependency relationship"""
        self.dependencies[dependent_key].add(dependency_key)
        self.reverse_dependencies[dependency_key].add(dependent_key)
 
    def remove_dependency(self, dependent_key: str, dependency_key: str):
        """Remove a dependency relationship"""
        self.dependencies[dependent_key].discard(dependency_key)
        self.reverse_dependencies[dependency_key].discard(dependent_key)
 
    def find_dependents(self, keys: List[str]) -> List[str]:
        """Find all keys that depend on the given keys"""
        visited = set()
        dependents = []
 
        def dfs(key):
            if key in visited:
                return
            visited.add(key)
 
            for dependent in self.reverse_dependencies[key]:
                dependents.append(dependent)
                dfs(dependent)
 
        for key in keys:
            dfs(key)
 
        return dependents
 
    def detect_cycles(self) -> List[List[str]]:
        """Detect dependency cycles that could cause invalidation loops"""
        visited = set()
        rec_stack = set()
        cycles = []
 
        def dfs_cycle_detection(node, path):
            if node in rec_stack:
                # Found a cycle
                cycle_start = path.index(node)
                cycles.append(path[cycle_start:] + [node])
                return
 
            if node in visited:
                return
 
            visited.add(node)
            rec_stack.add(node)
            path.append(node)
 
            for dependency in self.dependencies[node]:
                dfs_cycle_detection(dependency, path)
 
            rec_stack.remove(node)
            path.pop()
 
        for node in self.dependencies:
            if node not in visited:
                dfs_cycle_detection(node, [])
 
        return cycles

class VersionedCache:
    def __init__(self):
        self.cache = {}
        self.versions = {}
        self.global_version = 0
 
    def get(self, key: str, version: Optional[int] = None):
        """Get cached value with version checking"""
        if key not in self.cache:
            return None
 
        cached_item = self.cache[key]
 
        if version is not None and cached_item.version != version:
            # Requested specific version that doesn't match
            return None
 
        if cached_item.version < self.get_minimum_valid_version(key):
            # Cache entry is too old
            self.invalidate(key)
            return None
 
        return cached_item.value
 
    def set(self, key: str, value, dependencies: List[str] = None):
        """Set cached value with dependency tracking"""
        self.global_version += 1
 
        cached_item = CachedItem(
            value=value,
            version=self.global_version,
            dependencies=dependencies or [],
            timestamp=time.time()
        )
 
        self.cache[key] = cached_item
        self.versions[key] = self.global_version
 
    def invalidate_with_version_bump(self, keys: List[str]):
        """Invalidate keys and bump their minimum valid version"""
        self.global_version += 1
 
        for key in keys:
            if key in self.cache:
                del self.cache[key]
            self.versions[key] = self.global_version
 
    def get_minimum_valid_version(self, key: str) -> int:
        """Get minimum valid version for a cache key"""
        return self.versions.get(key, 0)

class ProbabilisticInvalidation:
    def __init__(self, base_invalidation_rate=0.1):
        self.base_rate = base_invalidation_rate
        self.access_patterns = {}
 
    def should_invalidate(self, key: str, data_age: timedelta) -> bool:
        """Determine if cache entry should be probabilistically invalidated"""
 
        # Base probability increases with age
        age_factor = min(data_age.total_seconds() / 3600, 10)  # Max 10x multiplier
 
        # Access pattern factor
        access_pattern = self.access_patterns.get(key, {})
        access_frequency = access_pattern.get('frequency', 1)
        last_access = access_pattern.get('last_access', time.time())
 
        # Recent high-frequency access decreases invalidation probability
        frequency_factor = 1 / (1 + math.log(access_frequency))
 
        # Time since last access increases invalidation probability
        staleness_factor = (time.time() - last_access) / 3600
 
        invalidation_probability = (
            self.base_rate *
            age_factor *
            frequency_factor *
            (1 + staleness_factor)
        )
 
        return random.random() < min(invalidation_probability, 0.9)
 
    def update_access_pattern(self, key: str):
        """Update access patterns for probabilistic calculations"""
        current_time = time.time()
 
        if key not in self.access_patterns:
            self.access_patterns[key] = {
                'frequency': 1,
                'last_access': current_time,
                'creation_time': current_time
            }
        else:
            pattern = self.access_patterns[key]
            pattern['frequency'] += 1
            pattern['last_access'] = current_time

class EventuallyConsistentCache:
    def __init__(self):
        self.local_cache = {}
        self.vector_clocks = {}
        self.conflict_resolver = ConflictResolver()
 
    def get_with_read_repair(self, key: str):
        """Get value with read repair for consistency"""
        local_value = self.local_cache.get(key)
 
        # Perform read repair by checking other nodes
        remote_values = self.fetch_from_other_nodes(key)
 
        if not remote_values:
            return local_value
 
        # Find the most recent value using vector clocks
        all_values = [(local_value, self.vector_clocks.get(key))] + remote_values
        most_recent = self.find_most_recent(all_values)
 
        # Detect and resolve conflicts
        conflicts = self.detect_conflicts(all_values)
        if conflicts:
            resolved_value = self.conflict_resolver.resolve(conflicts)
            self.propagate_resolution(key, resolved_value)
            return resolved_value
 
        # Update local cache if we found a newer value
        if most_recent != local_value:
            self.local_cache[key] = most_recent
            self.update_vector_clock(key, most_recent)
 
        return most_recent
 
    def detect_conflicts(self, values_with_clocks):
        """Detect conflicting values that can't be ordered by vector clocks"""
        conflicts = []
 
        for i, (value1, clock1) in enumerate(values_with_clocks):
            for j, (value2, clock2) in enumerate(values_with_clocks[i+1:], i+1):
                if not self.can_order_clocks(clock1, clock2):
                    conflicts.append((value1, clock1, value2, clock2))
 
        return conflicts
 
    def can_order_clocks(self, clock1: VectorClock, clock2: VectorClock) -> bool:
        """Check if two vector clocks can be totally ordered"""
        if not clock1 or not clock2:
            return True
 
        clock1_before = all(
            clock1.get(node, 0) <= clock2.get(node, 0)
            for node in set(clock1.keys()) | set(clock2.keys())
        )
 
        clock2_before = all(
            clock2.get(node, 0) <= clock1.get(node, 0)
            for node in set(clock1.keys()) | set(clock2.keys())
        )
 
        return clock1_before or clock2_before

class BatchInvalidationManager:
    def __init__(self, batch_size=100, flush_interval=5):
        self.batch_size = batch_size
        self.flush_interval = flush_interval
        self.pending_invalidations = set()
        self.last_flush = time.time()
        self.flush_timer = None
 
    def queue_invalidation(self, key: str):
        """Queue a cache key for batch invalidation"""
        self.pending_invalidations.add(key)
 
        # Flush if batch is full
        if len(self.pending_invalidations) >= self.batch_size:
            self.flush_invalidations()
 
        # Schedule flush if not already scheduled
        elif not self.flush_timer:
            self.schedule_flush()
 
    def flush_invalidations(self):
        """Flush all pending invalidations"""
        if not self.pending_invalidations:
            return
 
        keys_to_invalidate = list(self.pending_invalidations)
        self.pending_invalidations.clear()
 
        # Cancel scheduled flush
        if self.flush_timer:
            self.flush_timer.cancel()
            self.flush_timer = None
 
        # Group keys by cache layer for efficient invalidation
        grouped_keys = self.group_keys_by_layer(keys_to_invalidate)
 
        # Perform invalidation in parallel across layers
        with ThreadPoolExecutor(max_workers=len(grouped_keys)) as executor:
            futures = []
            for layer, keys in grouped_keys.items():
                future = executor.submit(self.invalidate_layer, layer, keys)
                futures.append(future)
 
            # Wait for all invalidations to complete
            for future in futures:
                try:
                    future.result(timeout=10)
                except Exception as e:
                    logger.error(f"Batch invalidation failed: {e}")
 
    def group_keys_by_layer(self, keys: List[str]) -> Dict[str, List[str]]:
        """Group cache keys by their target cache layer"""
        grouped = defaultdict(list)
 
        for key in keys:
            layers = self.determine_cache_layers(key)
            for layer in layers:
                grouped[layer].append(key)
 
        return grouped
 
    def invalidate_layer(self, layer: str, keys: List[str]):
        """Invalidate keys in a specific cache layer"""
        cache_client = self.get_cache_client(layer)
 
        if hasattr(cache_client, 'delete_multi'):
            # Use bulk delete if available
            cache_client.delete_multi(keys)
        else:
            # Fall back to individual deletes
            for key in keys:
                cache_client.delete(key)

class IntelligentPrefetcher:
    def __init__(self, cache, predictor_model):
        self.cache = cache
        self.predictor_model = predictor_model
        self.access_graph = AccessGraph()
 
    def prefetch_with_invalidation_awareness(self, current_key: str):
        """Prefetch likely-needed keys while respecting invalidation patterns"""
 
        # Predict likely next accesses
        predicted_keys = self.predictor_model.predict_next_accesses(current_key)
 
        # Filter out keys likely to be invalidated soon
        stable_keys = self.filter_stable_keys(predicted_keys)
 
        # Prefetch in priority order
        for key in stable_keys:
            if not self.cache.exists(key):
                self.prefetch_key(key)
 
    def filter_stable_keys(self, keys: List[str]) -> List[str]:
        """Filter out keys that are likely to be invalidated soon"""
        stable_keys = []
 
        for key in keys:
            invalidation_probability = self.calculate_invalidation_probability(key)
 
            if invalidation_probability < 0.3:  # Threshold for stability
                stable_keys.append(key)
 
        return stable_keys
 
    def calculate_invalidation_probability(self, key: str) -> float:
        """Calculate probability that a key will be invalidated soon"""
 
        # Factors affecting invalidation probability
        factors = {
            'data_volatility': self.get_data_volatility(key),
            'dependency_instability': self.get_dependency_instability(key),
            'update_frequency': self.get_update_frequency(key),
            'time_since_last_update': self.get_time_since_update(key)
        }
 
        # Weighted combination of factors
        probability = (
            factors['data_volatility'] * 0.3 +
            factors['dependency_instability'] * 0.25 +
            factors['update_frequency'] * 0.25 +
            factors['time_since_last_update'] * 0.2
        )
 
        return min(probability, 1.0)

class InvalidationMetrics:
    def __init__(self):
        self.metrics = {
            'invalidations_total': Counter('cache_invalidations_total'),
            'invalidation_latency': Histogram('cache_invalidation_latency_seconds'),
            'invalidation_errors': Counter('cache_invalidation_errors_total'),
            'dependency_chain_length': Histogram('invalidation_chain_length'),
            'batch_size': Histogram('invalidation_batch_size'),
            'stale_reads': Counter('cache_stale_reads_total')
        }
 
    def record_invalidation(self, keys: List[str], latency: float, success: bool):
        """Record invalidation metrics"""
        self.metrics['invalidations_total'].inc(len(keys))
        self.metrics['invalidation_latency'].observe(latency)
 
        if not success:
            self.metrics['invalidation_errors'].inc()
 
    def record_dependency_chain(self, chain_length: int):
        """Record dependency chain length for analysis"""
        self.metrics['dependency_chain_length'].observe(chain_length)
 
    def record_batch_invalidation(self, batch_size: int):
        """Record batch invalidation size"""
        self.metrics['batch_size'].observe(batch_size)
 
    def record_stale_read(self, key: str, staleness: timedelta):
        """Record detection of stale data read"""
        self.metrics['stale_reads'].inc()
 
        # Additional logging for investigation
        logger.warning(f"Stale read detected for key {key}, staleness: {staleness}")

class InvalidationAuditLog:
    def __init__(self, storage_backend):
        self.storage = storage_backend
 
    def log_invalidation(self, invalidation_event: InvalidationEvent):
        """Log detailed invalidation information for audit purposes"""
 
        audit_entry = {
            'timestamp': invalidation_event.timestamp.isoformat(),
            'invalidation_id': invalidation_event.id,
            'trigger_type': invalidation_event.trigger_type,
            'keys_invalidated': invalidation_event.keys,
            'invalidation_scope': invalidation_event.scope,
            'dependency_chain': invalidation_event.dependency_chain,
            'latency_ms': invalidation_event.latency_ms,
            'success': invalidation_event.success,
            'error_details': invalidation_event.error_details,
            'context': {
                'user_id': invalidation_event.context.get('user_id'),
                'request_id': invalidation_event.context.get('request_id'),
                'service': invalidation_event.context.get('service')
            }
        }
 
        self.storage.append(audit_entry)
 
    def analyze_invalidation_patterns(self, time_range: TimeRange) -> AnalysisReport:
        """Analyze invalidation patterns for optimization opportunities"""
 
        entries = self.storage.query(time_range)
 
        analysis = {
            'total_invalidations': len(entries),
            'average_chain_length': self.calculate_average_chain_length(entries),
            'most_frequently_invalidated_keys': self.find_frequent_keys(entries),
            'invalidation_hotspots': self.identify_hotspots(entries),
            'efficiency_metrics': self.calculate_efficiency_metrics(entries)
        }
 
        return AnalysisReport(analysis)

class InvalidationTestFramework:
    def __init__(self):
        self.test_cache = TestCache()
        self.dependency_tracker = DependencyTracker()
 
    def test_cascading_invalidation(self):
        """Test that dependent caches are properly invalidated"""
 
        # Set up dependency chain
        self.test_cache.set('user:1:profile', {'name': 'John'})
        self.test_cache.set('user:1:recommendations', ['item1', 'item2'])
        self.dependency_tracker.add_dependency(
            'user:1:recommendations', 'user:1:profile'
        )
 
        # Update base data
        self.test_cache.set('user:1:profile', {'name': 'Jane'})
 
        # Trigger invalidation
        invalidation_service = InvalidationService(
            self.test_cache, self.dependency_tracker
        )
        invalidation_service.invalidate('user:1:profile')
 
        # Verify dependent cache was invalidated
        assert self.test_cache.get('user:1:recommendations') is None
 
    def test_invalidation_loop_prevention(self):
        """Test that circular dependencies don't cause invalidation loops"""
 
        # Create circular dependency
        self.dependency_tracker.add_dependency('key_a', 'key_b')
        self.dependency_tracker.add_dependency('key_b', 'key_a')
 
        # Attempt invalidation
        invalidation_service = InvalidationService(
            self.test_cache, self.dependency_tracker
        )
 
        with pytest.raises(CircularDependencyError):
            invalidation_service.invalidate('key_a')
 
    def test_partial_invalidation_failure_recovery(self):
        """Test graceful handling of partial invalidation failures"""
 
        # Set up multi-layer cache
        cache_layers = [
            MockCacheLayer('L1', failure_rate=0.0),
            MockCacheLayer('L2', failure_rate=0.5),  # 50% failure rate
            MockCacheLayer('L3', failure_rate=0.0)
        ]
 
        invalidation_service = InvalidationService(cache_layers)
 
        # Perform invalidation
        result = invalidation_service.invalidate(['key1', 'key2', 'key3'])
 
        # Verify partial success handling
        assert result.partial_success is True
        assert len(result.failed_layers) == 1
        assert result.failed_layers[0].name == 'L2'
 
    def test_performance_under_load(self):
        """Test invalidation performance under high load"""
 
        # Generate large dataset
        keys = [f'key_{i}' for i in range(10000)]
        for key in keys:
            self.test_cache.set(key, f'value_{key}')
 
        # Measure batch invalidation performance
        start_time = time.time()
 
        invalidation_service = InvalidationService(self.test_cache)
        invalidation_service.invalidate_batch(keys)
 
        end_time = time.time()
        invalidation_time = end_time - start_time
 
        # Assert performance requirements
        assert invalidation_time < 5.0  # Should complete within 5 seconds
        assert len(keys) / invalidation_time > 1000  # At least 1000 invalidations/second