Coverage for src/local_deep_research/web_search_engines/rate_limiting/tracker.py: 97%
329 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1"""
2Adaptive rate limit tracker that learns optimal retry wait times for each search engine.
3"""
5import random
6import threading
7import time
8from collections import deque
9from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple
11from cachetools import LRUCache
13if TYPE_CHECKING:
14 from sqlalchemy.orm import Session
15from ...security.secure_logging import logger
17from ...config.thread_settings import (
18 NoSettingsContextError,
19 get_setting_from_snapshot,
20 get_settings_context,
21)
22from ...settings.env_registry import is_ci_environment
23from ...utilities.thread_context import get_search_context
25# Lazy imports to avoid database initialization in programmatic mode
26_db_imports = None
29def _get_db_imports():
30 """Lazy load database imports only when needed."""
31 global _db_imports
32 if _db_imports is None:
33 try:
34 from ...database.models import RateLimitAttempt, RateLimitEstimate
35 from ...database.session_context import get_user_db_session
37 _db_imports = {
38 "RateLimitAttempt": RateLimitAttempt,
39 "RateLimitEstimate": RateLimitEstimate,
40 "get_user_db_session": get_user_db_session,
41 }
42 except (ImportError, RuntimeError):
43 # Database not available - programmatic mode
44 _db_imports = {}
45 return _db_imports
48class AdaptiveRateLimitTracker:
49 """
50 Tracks and learns optimal retry wait times for each search engine.
51 Persists learned patterns to the main application database using SQLAlchemy.
52 """
54 def __init__(self, settings_snapshot=None, programmatic_mode=False):
55 self.settings_snapshot = settings_snapshot or {}
56 self.programmatic_mode = programmatic_mode
58 # Helper function to get settings with defaults
59 def get_setting_or_default(key, default, type_fn=None):
60 try:
61 value = get_setting_from_snapshot(
62 key,
63 settings_snapshot=self.settings_snapshot,
64 )
65 # Handle None values - return default instead of calling type_fn(None)
66 if value is None:
67 return default
68 return type_fn(value) if type_fn else value
69 except NoSettingsContextError:
70 return default
72 # Get settings with explicit defaults
73 self.memory_window = get_setting_or_default(
74 "rate_limiting.memory_window", 100, int
75 )
76 self.exploration_rate = get_setting_or_default(
77 "rate_limiting.exploration_rate", 0.1, float
78 )
79 self.learning_rate = get_setting_or_default(
80 "rate_limiting.learning_rate", 0.3, float
81 )
82 self.decay_per_day = get_setting_or_default(
83 "rate_limiting.decay_per_day", 0.95, float
84 )
86 # In programmatic mode, default to disabled
87 self.enabled = get_setting_or_default(
88 "rate_limiting.enabled",
89 not self.programmatic_mode, # Default based on mode
90 bool,
91 )
93 profile = get_setting_or_default("rate_limiting.profile", "balanced")
95 if self.programmatic_mode and self.enabled:
96 logger.info(
97 "Rate limiting enabled in programmatic mode - using memory-only tracking without persistence"
98 )
100 # Apply rate limiting profile
101 self._apply_profile(profile)
103 # Bounded in-memory caches for fast access (prevent memory leaks)
104 # Lock required: LRUCache is not thread-safe (mutates on read via LRU reordering)
105 self._cache_lock = threading.Lock()
106 self.recent_attempts: LRUCache = LRUCache(maxsize=100)
107 self.current_estimates: LRUCache = LRUCache(maxsize=100)
109 # Initialize the _estimates_loaded flag
110 self._estimates_loaded = False
112 # Load estimates from database
113 self._load_estimates()
115 logger.info(
116 f"AdaptiveRateLimitTracker initialized: enabled={self.enabled}, profile={profile}"
117 )
119 def _apply_profile(self, profile: str) -> None:
120 """Apply rate limiting profile settings."""
121 if profile == "conservative":
122 # More conservative: lower exploration, slower learning
123 self.exploration_rate = min(
124 self.exploration_rate * 0.5, 0.05
125 ) # 5% max exploration
126 self.learning_rate = min(
127 self.learning_rate * 0.7, 0.2
128 ) # Slower learning
129 logger.info("Applied conservative rate limiting profile")
130 elif profile == "aggressive":
131 # More aggressive: higher exploration, faster learning
132 self.exploration_rate = min(
133 self.exploration_rate * 1.5, 0.2
134 ) # Up to 20% exploration
135 self.learning_rate = min(
136 self.learning_rate * 1.3, 0.5
137 ) # Faster learning
138 logger.info("Applied aggressive rate limiting profile")
139 else: # balanced
140 # Use settings as-is
141 logger.info("Applied balanced rate limiting profile")
143 def _load_estimates(self) -> None:
144 """Load estimates from database into memory."""
145 # Skip database operations in programmatic mode
146 if self.programmatic_mode:
147 logger.debug(
148 "Skipping rate limit estimate loading in programmatic mode"
149 )
150 self._estimates_loaded = (
151 True # Mark as loaded to skip future attempts
152 )
153 return
155 # During initialization, we don't have user context yet
156 # Mark that we need to load estimates when user context becomes available
157 self._estimates_loaded = False
158 logger.debug(
159 "Rate limit estimates will be loaded on-demand when user context is available"
160 )
162 def _ensure_estimates_loaded(self) -> None:
163 """Load estimates from user's encrypted database if not already loaded."""
164 # Early return if already loaded or should skip
165 if self._estimates_loaded or self.programmatic_mode:
166 if not self._estimates_loaded:
167 self._estimates_loaded = True
168 return
170 # Get database imports
171 db_imports = _get_db_imports()
172 RateLimitEstimate = (
173 db_imports.get("RateLimitEstimate") if db_imports else None
174 )
176 if not db_imports or not RateLimitEstimate:
177 # Database not available
178 self._estimates_loaded = True
179 return
181 # Try to get research context from search tracker
183 context = get_search_context()
184 if not context:
185 # No context available (e.g., in tests or programmatic access)
186 self._estimates_loaded = True
187 return
189 username = context.get("username")
190 password = context.get("user_password")
192 if username and password:
193 try:
194 # Use thread-safe metrics writer to read from user's encrypted database
195 from ...database.thread_metrics import metrics_writer
197 # Set password for this thread
198 metrics_writer.set_user_password(username, password)
200 session: "Session"
201 with metrics_writer.get_session(username) as session:
202 estimates: list[Any] = session.query(
203 RateLimitEstimate
204 ).all()
206 for estimate in estimates:
207 # Apply decay for old estimates
208 age_hours = (time.time() - estimate.last_updated) / 3600
209 decay = self.decay_per_day ** (age_hours / 24)
211 with self._cache_lock:
212 self.current_estimates[estimate.engine_type] = {
213 "base": estimate.base_wait_seconds,
214 "min": estimate.min_wait_seconds,
215 "max": estimate.max_wait_seconds,
216 "confidence": decay,
217 }
219 logger.debug(
220 f"Loaded estimate for {estimate.engine_type}: base={estimate.base_wait_seconds:.2f}s, confidence={decay:.2f}"
221 )
223 self._estimates_loaded = True
224 logger.info(
225 f"Loaded {len(estimates)} rate limit estimates from encrypted database"
226 )
228 except Exception:
229 logger.warning("Could not load rate limit estimates")
230 # Mark as loaded anyway to avoid repeated attempts
231 self._estimates_loaded = True
233 def get_wait_time(self, engine_type: str) -> float:
234 """
235 Get adaptive wait time for a search engine.
236 Includes exploration to discover better rates.
238 Args:
239 engine_type: Name of the search engine
241 Returns:
242 Wait time in seconds
243 """
244 # If rate limiting is disabled, return minimal wait time
245 if not self.enabled:
246 return 0.1
248 # Check if we have a user context - if not, handle appropriately
249 context = get_search_context()
250 if not context and not self.programmatic_mode:
251 # No context and not in programmatic mode - this is unexpected
252 logger.warning(
253 f"No user context available for rate limiting on {engine_type} "
254 "but programmatic_mode=False. Disabling rate limiting. "
255 "This may indicate a configuration issue."
256 )
257 return 0.0
259 # In programmatic mode, we continue with memory-only rate limiting even without context
261 # Ensure estimates are loaded from database
262 self._ensure_estimates_loaded()
264 with self._cache_lock:
265 if engine_type not in self.current_estimates:
266 # First time seeing this engine - start optimistic and learn from real responses
267 # Use engine-specific optimistic defaults only for what we know for sure
268 optimistic_defaults = {
269 "SearXNGSearchEngine": 0.1, # Self-hosted default engine
270 }
272 wait_time = optimistic_defaults.get(
273 engine_type, 0.1
274 ) # Default optimistic for others
275 logger.info(
276 f"No rate limit data for {engine_type}, starting optimistic with {wait_time}s"
277 )
278 return wait_time
280 estimate = self.current_estimates[engine_type]
281 base_wait = estimate["base"]
283 # Security: random used for non-security rate-limit jitter, not tokens or secrets
284 # Exploration vs exploitation
285 if random.random() < self.exploration_rate:
286 # Explore: try a faster rate to see if API limits have relaxed
287 wait_time = base_wait * random.uniform(0.5, 0.9)
288 logger.debug(
289 f"Exploring faster rate for {engine_type}: {wait_time:.2f}s"
290 )
291 else:
292 # Exploit: use learned estimate with jitter
293 wait_time = base_wait * random.uniform(0.9, 1.1)
295 # Enforce bounds
296 wait_time = max(
297 float(estimate["min"]), min(wait_time, float(estimate["max"]))
298 )
299 return float(wait_time)
301 def apply_rate_limit(self, engine_type: str) -> float:
302 """
303 Apply rate limiting for the given engine type.
304 This is a convenience method that combines checking if rate limiting
305 is enabled, getting the wait time, and sleeping if necessary.
307 Args:
308 engine_type: The type of search engine
310 Returns:
311 The wait time that was applied (0 if rate limiting is disabled)
312 """
313 if not self.enabled:
314 return 0.0
316 wait_time = self.get_wait_time(engine_type)
317 if wait_time > 0:
318 logger.debug(
319 f"{engine_type} waiting {wait_time:.2f}s before request"
320 )
321 time.sleep(wait_time)
322 return wait_time
324 def record_outcome(
325 self,
326 engine_type: str,
327 wait_time: float,
328 success: bool,
329 retry_count: int,
330 error_type: Optional[str] = None,
331 search_result_count: Optional[int] = None,
332 ) -> None:
333 """
334 Record the outcome of a retry attempt.
336 Args:
337 engine_type: Name of the search engine
338 wait_time: How long we waited before this attempt
339 success: Whether the attempt succeeded
340 retry_count: Which retry attempt this was (1, 2, 3, etc.)
341 error_type: Type of error if failed
342 search_result_count: Number of search results returned (for quality monitoring)
343 """
344 # If rate limiting is disabled, don't record outcomes
345 if not self.enabled:
346 logger.info(
347 f"Rate limiting disabled - not recording outcome for {engine_type}"
348 )
349 return
351 logger.debug(
352 f"Recording rate limit outcome for {engine_type}: success={success}, wait_time={wait_time}s"
353 )
354 timestamp = time.time()
356 # NOTE: Database writes for rate limiting are disabled to prevent
357 # database locking issues under heavy parallel search load.
358 # Rate limiting still works via in-memory tracking below.
359 # Historical rate limit data is not persisted to DB.
361 # Update in-memory tracking
362 with self._cache_lock:
363 if engine_type not in self.recent_attempts:
364 # Get current memory window setting from thread context
365 settings_context = get_settings_context()
366 if settings_context:
367 current_memory_window = int(
368 settings_context.get_setting(
369 "rate_limiting.memory_window", self.memory_window
370 )
371 )
372 else:
373 current_memory_window = self.memory_window
375 self.recent_attempts[engine_type] = deque(
376 maxlen=current_memory_window
377 )
379 self.recent_attempts[engine_type].append(
380 {
381 "wait_time": wait_time,
382 "success": success,
383 "timestamp": timestamp,
384 "retry_count": retry_count,
385 "search_result_count": search_result_count,
386 }
387 )
389 # Update estimates
390 self._update_estimate(engine_type)
392 def _update_estimate(self, engine_type: str) -> None:
393 """Update wait time estimate based on recent attempts."""
394 with self._cache_lock:
395 if (
396 engine_type not in self.recent_attempts
397 or len(self.recent_attempts[engine_type]) < 3
398 ):
399 attempt_count = len(self.recent_attempts.get(engine_type) or [])
400 logger.info(
401 f"Not updating estimate for {engine_type} - only {attempt_count} attempts (need 3)"
402 )
403 return
405 attempts = list(self.recent_attempts[engine_type])
406 old_estimate = self.current_estimates.get(engine_type)
408 # Calculate success rate and optimal wait time
409 successful_waits = [a["wait_time"] for a in attempts if a["success"]]
410 failed_waits = [a["wait_time"] for a in attempts if not a["success"]]
412 if not successful_waits:
413 # All attempts failed - increase wait time with a cap
414 new_base = max(failed_waits) * 1.5 if failed_waits else 10.0
415 # Cap the base wait time to prevent runaway growth
416 new_base = min(new_base, 10.0) # Max 10 seconds base when all fail
417 else:
418 # Use 50th percentile (median) of successful waits for more stability
419 # This provides a balanced approach between speed and reliability
420 successful_waits.sort()
421 # Lower-median index: `(n - 1) // 2` is the middle element for odd n
422 # and the lower of the two middles for even n. The prior
423 # `int(n * 0.50) - 1` (left from a 25th-percentile version) sat below
424 # the median for odd n: at n == 3, the minimum qualifying sample, it
425 # picked index 0, the fastest wait, not the median.
426 median_index = (len(successful_waits) - 1) // 2
427 new_base = successful_waits[median_index]
429 # Update estimate with learning rate (exponential moving average)
430 if old_estimate is not None:
431 old_base = old_estimate["base"]
432 # Get current learning rate from settings context
433 settings_context = get_settings_context()
434 if settings_context:
435 current_learning_rate = float(
436 settings_context.get_setting(
437 "rate_limiting.learning_rate", self.learning_rate
438 )
439 )
440 else:
441 current_learning_rate = self.learning_rate
443 new_base = (
444 1 - current_learning_rate
445 ) * old_base + current_learning_rate * new_base
447 # Apply absolute cap to prevent extreme wait times
448 new_base = min(new_base, 10.0) # Cap base at 10 seconds
450 # Calculate bounds with more reasonable limits
451 min_wait = max(0.01, new_base * 0.5)
452 max_wait = min(10.0, new_base * 3.0) # Max 10 seconds absolute cap
454 # Update in memory
455 with self._cache_lock:
456 self.current_estimates[engine_type] = {
457 "base": new_base,
458 "min": min_wait,
459 "max": max_wait,
460 "confidence": min(len(attempts) / 20.0, 1.0),
461 }
463 # Persist to database
464 success_rate = len(successful_waits) / len(attempts) if attempts else 0
466 # Skip database operations in programmatic mode
467 if self.programmatic_mode:
468 logger.debug(
469 f"Skipping estimate persistence in programmatic mode for {engine_type}"
470 )
471 else:
472 # Try to get research context from search tracker
474 context = get_search_context()
475 username = None
476 password = None
477 if context is not None:
478 username = context.get("username")
479 password = context.get("user_password")
481 if username and password:
482 try:
483 # Use thread-safe metrics writer to save to user's encrypted database
484 from ...database.thread_metrics import metrics_writer
486 # Set password for this thread if not already set
487 metrics_writer.set_user_password(username, password)
489 db_imports = _get_db_imports()
490 RateLimitEstimate = db_imports.get("RateLimitEstimate")
492 session_inner: "Session"
493 with metrics_writer.get_session(username) as session_inner:
494 # Check if estimate exists
495 estimate: Any = (
496 session_inner.query(RateLimitEstimate)
497 .filter_by(engine_type=engine_type)
498 .first()
499 )
501 if estimate:
502 # Update existing estimate
503 estimate.base_wait_seconds = new_base
504 estimate.min_wait_seconds = min_wait
505 estimate.max_wait_seconds = max_wait
506 estimate.last_updated = time.time()
507 estimate.total_attempts = len(attempts)
508 estimate.success_rate = success_rate
509 else:
510 # Create new estimate
511 estimate = RateLimitEstimate(
512 engine_type=engine_type,
513 base_wait_seconds=new_base,
514 min_wait_seconds=min_wait,
515 max_wait_seconds=max_wait,
516 last_updated=time.time(),
517 total_attempts=len(attempts),
518 success_rate=success_rate,
519 )
520 session_inner.add(estimate)
522 except Exception:
523 logger.warning("Failed to persist rate limit estimate")
524 else:
525 logger.debug(
526 "Skipping rate limit estimate save - no user context"
527 )
529 logger.info(
530 f"Updated rate limit for {engine_type}: {new_base:.2f}s "
531 f"(success rate: {success_rate:.1%})"
532 )
534 def _get_in_memory_stats(
535 self, engine_type: Optional[str] = None
536 ) -> List[Tuple[str, float, float, float, float, int, float]]:
537 """Return stats from in-memory caches (no DB access)."""
538 with self._cache_lock:
539 stats = []
540 engines_to_check = (
541 [engine_type]
542 if engine_type
543 else list(self.current_estimates.keys())
544 )
545 for engine in engines_to_check:
546 if engine in self.current_estimates:
547 est = self.current_estimates[engine]
548 attempts = self.recent_attempts.get(engine)
549 attempt_count = len(attempts) if attempts else 0
550 stats.append(
551 (
552 engine,
553 est["base"],
554 est["min"],
555 est["max"],
556 time.time(),
557 attempt_count,
558 est.get("confidence", 0.0),
559 )
560 )
561 return stats
563 def get_stats(
564 self, engine_type: Optional[str] = None
565 ) -> List[Tuple[str, float, float, float, float, int, float]]:
566 """
567 Get statistics for monitoring.
569 Args:
570 engine_type: Specific engine to get stats for, or None for all
572 Returns:
573 List of tuples with engine statistics
574 """
575 # Skip database operations in CI mode
576 if is_ci_environment():
577 logger.debug("Skipping database stats in CI mode")
578 return self._get_in_memory_stats(engine_type)
580 # Skip database operations in programmatic mode
581 if self.programmatic_mode:
582 return self._get_in_memory_stats(engine_type)
584 try:
585 context = get_search_context()
586 if not context:
587 return self._get_in_memory_stats(engine_type)
589 username = context.get("username")
590 password = context.get("user_password")
591 if not username or not password: 591 ↛ 592line 591 didn't jump to line 592 because the condition on line 591 was never true
592 return self._get_in_memory_stats(engine_type)
594 db_imports = _get_db_imports()
595 RateLimitEstimate = db_imports.get("RateLimitEstimate")
597 from ...database.thread_metrics import metrics_writer
599 metrics_writer.set_user_password(username, password)
601 session_est: "Session"
602 with metrics_writer.get_session(username) as session_est:
603 if engine_type:
604 estimates: list[Any] = (
605 session_est.query(RateLimitEstimate)
606 .filter_by(engine_type=engine_type)
607 .all()
608 )
609 else:
610 estimates = (
611 session_est.query(RateLimitEstimate)
612 .order_by(RateLimitEstimate.engine_type)
613 .all()
614 )
616 return [
617 (
618 est.engine_type,
619 est.base_wait_seconds,
620 est.min_wait_seconds,
621 est.max_wait_seconds,
622 est.last_updated,
623 est.total_attempts,
624 est.success_rate,
625 )
626 for est in estimates
627 ]
628 except Exception:
629 logger.warning("Failed to get rate limit stats from DB")
630 # Return in-memory stats as fallback
631 return self._get_in_memory_stats(engine_type)
633 def reset_engine(self, engine_type: str) -> None:
634 """
635 Reset learned values for a specific engine.
637 Args:
638 engine_type: Engine to reset
639 """
640 # Always clear from memory first
641 with self._cache_lock:
642 if engine_type in self.recent_attempts:
643 del self.recent_attempts[engine_type]
644 if engine_type in self.current_estimates:
645 del self.current_estimates[engine_type]
647 # Skip database operations in programmatic mode
648 if self.programmatic_mode:
649 logger.debug(
650 f"Reset rate limit data for {engine_type} (memory only in programmatic mode)"
651 )
652 return
654 # Skip database operations in CI mode
655 if is_ci_environment():
656 logger.debug(
657 f"Reset rate limit data for {engine_type} (memory only in CI mode)"
658 )
659 return
661 try:
662 context = get_search_context()
663 if not context: 663 ↛ 664line 663 didn't jump to line 664 because the condition on line 663 was never true
664 logger.debug(
665 f"Reset rate limit data for {engine_type} (memory only - no user context)"
666 )
667 return
669 username = context.get("username")
670 password = context.get("user_password")
671 if not username or not password: 671 ↛ 672line 671 didn't jump to line 672 because the condition on line 671 was never true
672 logger.debug(
673 f"Reset rate limit data for {engine_type} (memory only - no credentials)"
674 )
675 return
677 db_imports = _get_db_imports()
678 RateLimitAttempt = db_imports.get("RateLimitAttempt")
679 RateLimitEstimate = db_imports.get("RateLimitEstimate")
681 from ...database.thread_metrics import metrics_writer
683 metrics_writer.set_user_password(username, password)
685 session_reset: "Session"
686 with metrics_writer.get_session(username) as session_reset:
687 # Delete historical attempts
688 session_reset.query(RateLimitAttempt).filter_by(
689 engine_type=engine_type
690 ).delete()
692 # Delete estimates
693 session_reset.query(RateLimitEstimate).filter_by(
694 engine_type=engine_type
695 ).delete()
697 session_reset.commit()
699 logger.info(f"Reset rate limit data for {engine_type}")
701 except Exception:
702 logger.warning(
703 f"Failed to reset rate limit data in database for {engine_type}"
704 "In-memory data was cleared successfully."
705 )
706 # Don't re-raise in test contexts - the memory cleanup is sufficient
708 def get_search_quality_stats(
709 self, engine_type: Optional[str] = None
710 ) -> List[Dict]:
711 """
712 Get basic search quality statistics for monitoring.
714 Note: no HTTP route consumes this method any more —
715 ``/benchmark/api/search-quality`` now reports success_rate from the
716 persisted ``RateLimitEstimate`` table instead of this in-memory,
717 per-process view. It is retained as a tracker diagnostic capability
718 (with its own test coverage) for CLI / programmatic use; the
719 ``recent_avg_results`` / ``sample_size`` shape it returns is no longer
720 exposed over the API.
722 Args:
723 engine_type: Specific engine to get stats for, or None for all
725 Returns:
726 List of dictionaries with search quality metrics
727 """
728 stats = []
730 with self._cache_lock:
731 engines_to_check = (
732 [engine_type]
733 if engine_type
734 else list(self.recent_attempts.keys())
735 )
736 # Snapshot the data under lock, then process outside
737 engine_attempts = {}
738 for engine in engines_to_check:
739 if engine in self.recent_attempts:
740 engine_attempts[engine] = list(self.recent_attempts[engine])
742 for engine, recent in engine_attempts.items():
743 search_counts = [
744 attempt.get("search_result_count", 0)
745 for attempt in recent
746 if attempt.get("search_result_count") is not None
747 ]
749 if not search_counts:
750 continue
752 recent_avg = sum(search_counts) / len(search_counts)
754 stats.append(
755 {
756 "engine_type": engine,
757 "recent_avg_results": recent_avg,
758 "min_recent_results": min(search_counts),
759 "max_recent_results": max(search_counts),
760 "sample_size": len(search_counts),
761 "total_attempts": len(recent),
762 "status": self._get_quality_status(recent_avg),
763 }
764 )
766 return stats
768 def _get_quality_status(self, recent_avg: float) -> str:
769 """Get quality status string based on average results."""
770 if recent_avg < 1:
771 return "CRITICAL"
772 if recent_avg < 3:
773 return "WARNING"
774 if recent_avg < 5:
775 return "CAUTION"
776 if recent_avg >= 10:
777 return "EXCELLENT"
778 return "GOOD"
780 def cleanup_old_data(self, days: int = 30) -> None:
781 """
782 Remove old retry attempt data to prevent database bloat.
784 Args:
785 days: Remove data older than this many days
786 """
787 cutoff_time = time.time() - (days * 24 * 3600)
789 # Skip database operations in programmatic mode
790 if self.programmatic_mode:
791 logger.debug("Skipping database cleanup in programmatic mode")
792 return
794 # Skip database operations in CI mode
795 if is_ci_environment():
796 logger.debug("Skipping database cleanup in CI mode")
797 return
799 try:
800 context = get_search_context()
801 if not context: 801 ↛ 802line 801 didn't jump to line 802 because the condition on line 801 was never true
802 logger.debug("Skipping database cleanup - no user context")
803 return
805 username = context.get("username")
806 password = context.get("user_password")
807 if not username or not password: 807 ↛ 808line 807 didn't jump to line 808 because the condition on line 807 was never true
808 logger.debug("Skipping database cleanup - no credentials")
809 return
811 db_imports = _get_db_imports()
812 RateLimitAttempt = db_imports.get("RateLimitAttempt")
814 from ...database.thread_metrics import metrics_writer
816 metrics_writer.set_user_password(username, password)
818 session_clean: "Session"
819 with metrics_writer.get_session(username) as session_clean:
820 # Count and delete old attempts
821 old_attempts: Any = session_clean.query(
822 RateLimitAttempt
823 ).filter(RateLimitAttempt.timestamp < cutoff_time)
824 deleted_count = old_attempts.count()
825 old_attempts.delete()
827 session_clean.commit()
829 if deleted_count > 0:
830 logger.info(f"Cleaned up {deleted_count} old retry attempts")
832 except Exception:
833 logger.warning("Failed to cleanup old rate limit data")
836def get_tracker() -> AdaptiveRateLimitTracker:
837 """Create a fresh rate limit tracker instance.
839 Returns a new instance each call so that per-user DB credentials
840 are never cached across requests in a multi-user Flask app.
841 """
842 return AdaptiveRateLimitTracker()