Coverage for src/local_deep_research/web_search_engines/engines/_searxng_rate_limiter.py: 100%

1"""Shared rate limiter for SearXNG engine instances. 

2 

3The research agent constructs a fresh `SearXNGSearchEngine` for every 

4tool call, so the previous per-instance `last_request_time` only 

5throttled requests on engines that were explicitly reused (for example 

6the shared engine used by `JournalReputationFilter` for Tier 4). This 

7module provides a process-wide tracker keyed by SearXNG ``instance_url`` 

8so the configured ``delay_between_requests`` actually applies across 

9per-call engine instances. 

10 

11The locking pattern mirrors other per-key locks in the codebase 

12(see ``database/backup/backup_service.py`` and 

13``database/encrypted_db.py``): a meta-lock guards lazy creation of a 

14per-URL lock, and that per-URL lock guards updates to the timestamp. 

15 

16Note on memory: 

17`_url_state` is bounded to a maximum capacity (`MAX_TRACKED_URLS`). When 

18capacity is reached, stale/least-recently-used entries are evicted under 

19`_meta_lock`. 

20""" 

21 

22import threading 

23import time 

24from dataclasses import dataclass, field 

25 

26from ...security import redact_url_for_log 

27from ...security.secure_logging import logger 

28 

29MAX_TRACKED_URLS = 1000 

30 

31 

32@dataclass 

33class _UrlState: 

34 """Lock for one tracked URL, and when it was last requested.""" 

35 

36 lock: threading.Lock = field(default_factory=threading.Lock) 

37 last_request: float = 0.0 # monotonic; 0.0 means "no request yet" 

38 

39 

40_meta_lock = threading.Lock() 

41_url_state: dict[str, _UrlState] = {} 

42 

43 

44def _normalize_url(url: str) -> str: 

45 """Normalize instance URL for keying rate limits.""" 

46 return url.rstrip("/") 

47 

48 

49def _evict_stale_locks_unlocked() -> None: 

50 """Evict oldest entries when tracked URLs reach capacity. 

51 

52 A lock another thread is currently holding is never a candidate: dropping 

53 it lets the next caller for that URL build a fresh lock and read no 

54 timestamp, so the holder's delay stops applying. When every tracked lock 

55 is held there is nothing to reclaim and the tracker grows past 

56 ``MAX_TRACKED_URLS`` until one is released. 

57 

58 ``locked()`` is not a complete answer, and this function does not make it 

59 one. ``respect_rate_limit`` fetches its state from ``_get_url_state`` under 

60 ``_meta_lock`` and acquires the lock afterwards, so between those two steps 

61 the object reports ``locked() == False`` and stays evictable. A URL evicted 

62 in that window loses its timestamp and skips one delay. The window is 

63 narrower than the one this check closes, which spans 

64 ``time.sleep(wait_time)``, and closing it as well would mean holding 

65 ``_meta_lock`` across the sleep. 

66 

67 Must be called while holding ``_meta_lock``. 

68 """ 

69 if len(_url_state) < MAX_TRACKED_URLS: 

70 return 

71 evictable = [ 

72 url for url, state in _url_state.items() if not state.lock.locked() 

73 ] 

74 if not evictable: 

75 logger.warning( 

76 f"SearXNG rate limiter: all {len(_url_state)} tracked URL locks " 

77 "are in use, so none can be evicted at capacity" 

78 ) 

79 return 

80 # Remove oldest half of entries based on their last_request timestamp 

81 sorted_urls = sorted(evictable, key=lambda u: _url_state[u].last_request) 

82 to_remove = sorted_urls[: max(1, len(sorted_urls) // 2)] 

83 for url in to_remove: 

84 _url_state.pop(url, None) 

85 

86 

87def _get_url_state(normalized_url: str) -> _UrlState: 

88 """Return the per-URL state, creating it lazily. 

89 

90 Expects an already normalized URL string. 

91 """ 

92 with _meta_lock: 

93 state = _url_state.get(normalized_url) 

94 if state is None: 

95 _evict_stale_locks_unlocked() 

96 state = _UrlState() 

97 _url_state[normalized_url] = state 

98 return state 

99 

100 

101def respect_rate_limit(instance_url: str, delay_seconds: float) -> None: 

102 """Ensure at least ``delay_seconds`` have passed since the previous call 

103 for this ``instance_url`` (does not wait on the first call for a URL). 

104 

105 A ``delay_seconds`` of ``0`` (or less) returns immediately without 

106 touching the tracker, preserving the prior "no throttling" behavior 

107 when the user has not configured any delay. 

108 """ 

109 if delay_seconds <= 0: 

110 return 

111 

112 normalized_url = _normalize_url(instance_url) 

113 state = _get_url_state(normalized_url) 

114 with state.lock: 

115 now = time.monotonic() 

116 last = state.last_request 

117 elapsed = now - last 

118 if last > 0 and elapsed < delay_seconds: 

119 wait_time = delay_seconds - elapsed 

120 logger.info( 

121 f"SearXNG rate limiting: waiting {wait_time:.2f}s for instance {redact_url_for_log(normalized_url)}" 

122 ) 

123 time.sleep(wait_time) 

124 now = time.monotonic() 

125 # Attribute write on a held reference: an evicted entry stays evicted. 

126 state.last_request = now 

127 

128 

129def reset_for_tests() -> None: 

130 """Clear all tracked state. Intended for unit tests only.""" 

131 with _meta_lock: 

132 _url_state.clear()