Coverage for src/local_deep_research/web_search_engines/engines/_searxng_rate_limiter.py: 99%
120 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""Shared rate limiter for SearXNG engine instances.
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.
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.
16Note on memory:
17`MAX_TRACKED_URLS` is an eviction target, not a hard bound: the tracker can
18grow while all existing locks are held. At capacity, evictable entries are
19removed under `_meta_lock`. Each URL has exactly one tracked `_UrlState` carrying both
20its lock and its timestamp (#5688), so a timestamp write is an attribute
21write on a reference the caller already holds -- it cannot recreate an
22entry eviction has removed, the way writing into a *second*, separately
23keyed map could.
24"""
26import threading
27import time
28from dataclasses import dataclass, field
30from ...security import redact_url_for_log
31from ...security.secure_logging import logger
33MAX_TRACKED_URLS = 1000
36@dataclass
37class _UrlState:
38 """Lock for one tracked URL, and when it was last requested."""
40 lock: threading.Lock = field(default_factory=threading.Lock)
41 last_request: float = 0.0 # monotonic; 0.0 means "no request yet"
44_meta_lock = threading.Lock()
45_url_state: dict[str, _UrlState] = {}
47# How long the "nothing to evict" warning stays quiet before repeating while
48# the condition persists.
49ALL_HELD_REWARN_SECONDS = 300.0
51# Hysteresis. Leaving the all-held state needs several free slots, not merely
52# one. A single evictable entry at capacity is usually just a lock sitting
53# between ``_get_url_state`` and its caller's ``with state.lock:`` -- treating
54# that as recovery would clear the throttle and let the warning fire again on
55# the very next URL, which is the behaviour this state exists to prevent.
56#
57# The divisor scales the recovery band with capacity. Below 20 it supplies
58# at most one slot; requiring two prevents a briefly free lock from clearing
59# the warning between lookup and acquisition. The threshold is capped at
60# capacity so recovery remains possible for a one-slot tracker.
61ALL_HELD_RECOVERY_HEADROOM_DIVISOR = 10
62ALL_HELD_RECOVERY_MIN_SLOTS = 2
64# When that warning was last emitted, or None when the condition is not
65# active. A timestamp rather than a flag so a condition that never clears
66# keeps reporting instead of going silent after one line. Guarded by
67# ``_meta_lock``.
68_all_held_warned_at: float | None = None
69# Onset of the current saturated episode. Distinct from
70# ``_all_held_warned_at``, which is rewritten on every re-warn so it can keep
71# governing ``ALL_HELD_REWARN_SECONDS``. Deriving the duration from that
72# would report the time since the last re-warn rather than the time at
73# capacity -- a twenty-minute episode with three re-warns would claim ~300s.
74_all_held_since: float | None = None
76# Orders the log lines the state transitions produce. The decision is made
77# under ``_meta_lock`` but emitted after releasing it, so two threads can
78# reach ``_emit_capacity_log`` in the opposite order to the decisions they
79# carry. ``_emit_lock`` serializes the emissions and ``_emitted_generation``
80# drops any record a newer transition has already overtaken. Never acquired
81# while holding ``_meta_lock``. ``reset_for_tests`` is the one path that takes
82# ``_meta_lock`` while holding it; the hierarchy stays acyclic because no
83# thread ever waits on ``_emit_lock`` while holding ``_meta_lock``.
84_emit_lock = threading.Lock()
85_state_generation = 0
86_emitted_generation = 0
89def _normalize_url(url: str) -> str:
90 """Normalize instance URL for keying rate limits."""
91 return url.rstrip("/")
94def _recovery_headroom_slots() -> int:
95 """Free slots required before the all-held state is considered over.
97 Read from ``MAX_TRACKED_URLS`` on each call so tests that shrink the
98 capacity get a proportional threshold instead of a stale constant.
100 Capped at the capacity itself: while anything is tracked, headroom can
101 never exceed ``MAX_TRACKED_URLS``, so a larger threshold would make
102 recovery unreachable and pin the state as warned for good.
103 """
104 return min(
105 max(MAX_TRACKED_URLS, 0),
106 max(
107 ALL_HELD_RECOVERY_MIN_SLOTS,
108 MAX_TRACKED_URLS // ALL_HELD_RECOVERY_HEADROOM_DIVISOR,
109 ),
110 )
113def _count_evictable_unlocked(limit: int) -> int:
114 """Count tracked locks nobody holds, counting no further than ``limit``.
116 The result is exact when it comes back below ``limit`` and a lower bound
117 when it reaches it. Callers pass the smallest limit that still separates
118 the branches they have to choose between, which caps the *free* entries
119 counted -- not the entries walked. A held lock is stepped over without
120 advancing the count, so the early exit only pays off while free entries
121 are plentiful; a tracker whose locks are mostly held is scanned in full.
122 That is the saturated state this reporting exists for, and while it lasts
123 the scan runs on every ``_get_url_state`` under ``_meta_lock``. It stays
124 affordable because the tracker is keyed by SearXNG instance URL, so it
125 holds a handful of entries in practice rather than ``MAX_TRACKED_URLS``
126 of them.
128 Must be called while holding ``_meta_lock``.
129 """
130 if limit <= 0: 130 ↛ 131line 130 didn't jump to line 131 because the condition on line 130 was never true
131 return 0
132 count = 0
133 for state in _url_state.values():
134 if not state.lock.locked():
135 count += 1
136 if count >= limit:
137 break
138 return count
141def _capacity_pressure_unlocked() -> (
142 tuple[bool, str, tuple[object, ...], int] | None
143):
144 """Update the all-held state and return the line it wants logged.
146 Returns ``None`` when nothing needs saying, otherwise
147 ``(is_warning, message, args, generation)`` for the caller to emit
148 *after* releasing ``_meta_lock``. The database sink queues the record
149 from the worker threads this code runs on rather than writing it inline
150 (``utilities/log_utils.py``), but the formatting, the queueing and every
151 other sink still run on the calling thread, and ``_meta_lock`` serializes
152 every lock lookup in the process. The emission is moved out to keep that
153 critical section short and free of I/O. ``generation`` increases with
154 every transition so a thread descheduled between deciding and emitting
155 cannot publish a line a newer transition has already superseded (see
156 ``_emit_capacity_log``).
158 The decision runs on every ``_get_url_state`` call rather than only when
159 an entry is created, so an exhaustion that clears while traffic continues
160 on already-tracked URLs is observed -- otherwise a later, genuinely
161 distinct episode would be silently swallowed by the re-warn interval.
163 Saturation is measured as *headroom*: the number of entries the tracker
164 can still admit without dropping a lock that is in use, i.e. free
165 capacity plus evictable entries. Warning at ``headroom <= 0`` and
166 recovering only at ``headroom >= _recovery_headroom_slots()`` gives the
167 state hysteresis, so one transient entry cannot flip it.
169 Must be called while holding ``_meta_lock``.
170 """
171 global _all_held_warned_at, _all_held_since, _state_generation
173 tracked = len(_url_state)
174 if _all_held_warned_at is None and tracked < MAX_TRACKED_URLS:
175 # Free capacity and no outstanding warning: nothing can change, and
176 # the common path avoids the scan below.
177 return None
179 slots = _recovery_headroom_slots()
180 # Once headroom is provably this large no further free lock can change
181 # which branch runs, so the count stops there. Unwarned, only the
182 # ``headroom <= 0`` boundary matters and one free lock settles it.
183 needed = slots if _all_held_warned_at is not None else 1
184 evictable = _count_evictable_unlocked(tracked - MAX_TRACKED_URLS + needed)
185 headroom = MAX_TRACKED_URLS - tracked + evictable
187 if evictable < tracked and headroom <= 0:
188 # ``evictable < tracked`` keeps the claim honest: at least one lock
189 # really is in use. Without it a capacity of zero would warn about a
190 # lone free, unheld lock on the very first lookup.
191 now = time.monotonic()
192 if (
193 _all_held_warned_at is None
194 or now - _all_held_warned_at >= ALL_HELD_REWARN_SECONDS
195 ):
196 if _all_held_warned_at is None:
197 # Entry to the episode. Re-warns rewrite the timer below but
198 # must not move the onset, or the recovery line would report
199 # the re-warn interval instead of the real duration.
200 _all_held_since = now
201 _all_held_warned_at = now
202 _state_generation += 1
203 # ``headroom <= 0`` is exactly ``tracked - evictable >=
204 # MAX_TRACKED_URLS``: evicting every free entry still would not
205 # get back under the limit. Some of them may well be evictable,
206 # so the line reports both counts instead of claiming none are.
207 return (
208 True,
209 "SearXNG rate limiter: only {} of {} tracked URL locks are "
210 "evictable, not enough to bring the tracker back under its "
211 "capacity of {}",
212 (evictable, tracked, MAX_TRACKED_URLS),
213 _state_generation,
214 )
215 return None
217 if _all_held_warned_at is None:
218 return None
219 if tracked and headroom < slots:
220 # Still saturated, just not to the last slot. Stay warned so the
221 # re-warn interval keeps governing.
222 return None
224 # ``_all_held_since`` is written alongside ``_all_held_warned_at`` on entry,
225 # so it is set here; the fallback only guards a torn state. Compare with
226 # ``is not None`` rather than truthiness -- a 0.0 onset is a real reading.
227 onset = (
228 _all_held_since if _all_held_since is not None else _all_held_warned_at
229 )
230 held_for = time.monotonic() - onset
231 _all_held_warned_at = None
232 _all_held_since = None
233 _state_generation += 1
234 # The bounded count above is a lower bound once it hits its limit, and
235 # this line quotes real numbers, so finish counting. Once per transition,
236 # not once per call.
237 evictable = _count_evictable_unlocked(tracked)
238 return (
239 False,
240 "SearXNG rate limiter: {} of {} tracked URL locks are evictable "
241 "again after {:.1f}s at capacity",
242 (evictable, tracked, held_for),
243 _state_generation,
244 )
247def _emit_capacity_log(
248 pending: tuple[bool, str, tuple[object, ...], int] | None,
249) -> None:
250 """Emit the record ``_capacity_pressure_unlocked`` produced, if any.
252 Call with ``_meta_lock`` released: the emission is out here to keep the
253 meta-lock's critical section short and free of I/O, not because the sink
254 blocks (see ``_capacity_pressure_unlocked``).
256 ``_emit_lock`` puts the lines back in the order their transitions were
257 decided, and a record a newer transition has already overtaken is dropped
258 rather than logged late. Emitting it late would leave an operator reading
259 "evictable again" as the last word on a tracker that is in fact saturated
260 and warned -- with the next warning then suppressed for
261 ``ALL_HELD_REWARN_SECONDS``, or indefinitely if headroom settles inside
262 the hysteresis band.
264 A recovery is held to the same standard in the other direction: it is
265 emitted only when the transition it closes has already been emitted. A
266 recovery that overtakes its own still-unemitted warning would otherwise
267 print "evictable again" with nothing preceding it to recover from, and
268 would stamp that warning away as stale on the way past. Dropping the
269 recovery instead leaves the warning to be emitted on its own, which reads
270 as an episode still open rather than one that never happened.
271 """
272 global _emitted_generation
274 if pending is None:
275 return
276 is_warning, message, args, generation = pending
277 with _emit_lock:
278 if generation <= _emitted_generation:
279 return
280 if not is_warning and generation > _emitted_generation + 1:
281 return
282 _emitted_generation = generation
283 if is_warning:
284 logger.warning(message, *args)
285 else:
286 logger.info(message, *args)
289def _eviction_sort_key(state: "_UrlState") -> float:
290 """Order entries for eviction: oldest genuine timestamp first.
292 ``last_request == 0.0`` marks a URL whose ``_UrlState`` has been created
293 but has not recorded a request yet. That is reachable and unlocked: an
294 entry a *different* thread just created shows up this way for the
295 window between ``_get_url_state`` returning it and that thread's
296 ``respect_rate_limit`` reaching ``with state.lock:`` -- the same window
297 ``_evict_stale_locks_unlocked`` documents as leaving the entry evictable.
298 (It is not reachable for a request whose delay is still being applied:
299 that path only sleeps when ``last_request > 0``, which never hits this
300 branch, and it holds ``state.lock`` for the whole sleep, which the
301 caller's lock-held check excludes regardless of this key.)
303 It sorts last, not first, so a just-created entry with no timestamp yet
304 is never the preferred eviction victim; defaulting it to the front would
305 make the newest entries the ones eviction reaches for first, ahead of
306 entries that actually went stale.
307 """
308 return state.last_request if state.last_request > 0.0 else float("inf")
311def _evict_stale_locks_unlocked() -> None:
312 """Evict oldest entries when tracked URLs reach capacity.
314 A lock another thread is currently holding is never a candidate: dropping
315 it lets the next caller for that URL build a fresh state and read no
316 timestamp, so the holder's delay stops applying. When every tracked lock
317 is held there is nothing to reclaim and the tracker grows past
318 ``MAX_TRACKED_URLS`` until one is released. Reporting that condition is
319 ``_capacity_pressure_unlocked``'s job, not this function's.
321 ``locked()`` is not a complete answer, and this function does not make it
322 one. ``respect_rate_limit`` fetches its state from ``_get_url_state``
323 under ``_meta_lock`` and acquires the lock afterwards, so between those
324 two steps the object reports ``locked() == False`` and stays evictable. A
325 URL evicted in that window loses its timestamp and skips one delay. The
326 window is narrower than the one this check closes, which spans
327 ``time.sleep(wait_time)``.
329 Closing it too is possible without serializing the sleeps -- pin the entry
330 with a refcount incremented under ``_meta_lock`` in ``_get_url_state`` and
331 decremented after the caller releases, and skip pinned entries here -- but
332 that is more machinery than the residual window costs: one skipped delay
333 for one URL, self-healing on that URL's next request.
335 Must be called while holding ``_meta_lock``.
336 """
337 if len(_url_state) < MAX_TRACKED_URLS:
338 return
339 evictable = [
340 url for url, state in _url_state.items() if not state.lock.locked()
341 ]
342 if not evictable:
343 return
344 # Remove oldest half of entries based on _eviction_sort_key.
345 sorted_urls = sorted(
346 evictable, key=lambda u: _eviction_sort_key(_url_state[u])
347 )
348 to_remove = sorted_urls[: max(1, len(sorted_urls) // 2)]
349 for url in to_remove:
350 _url_state.pop(url, None)
353def _get_url_state(normalized_url: str) -> _UrlState:
354 """Return the per-URL state, creating it lazily.
356 Expects an already normalized URL string.
357 """
358 with _meta_lock:
359 state = _url_state.get(normalized_url)
360 if state is None:
361 _evict_stale_locks_unlocked()
362 state = _UrlState()
363 _url_state[normalized_url] = state
364 pending = _capacity_pressure_unlocked()
365 _emit_capacity_log(pending)
366 return state
369def respect_rate_limit(instance_url: str, delay_seconds: float) -> None:
370 """Ensure at least ``delay_seconds`` have passed since the previous call
371 for this ``instance_url`` (does not wait on the first call for a URL).
373 A ``delay_seconds`` of ``0`` (or less) returns immediately without
374 touching the tracker, preserving the prior "no throttling" behavior
375 when the user has not configured any delay.
376 """
377 if delay_seconds <= 0:
378 return
380 normalized_url = _normalize_url(instance_url)
381 state = _get_url_state(normalized_url)
382 with state.lock:
383 now = time.monotonic()
384 last = state.last_request
385 elapsed = now - last
386 if last > 0 and elapsed < delay_seconds:
387 wait_time = delay_seconds - elapsed
388 logger.info(
389 f"SearXNG rate limiting: waiting {wait_time:.2f}s for instance {redact_url_for_log(normalized_url)}"
390 )
391 time.sleep(wait_time)
392 now = time.monotonic()
393 # Other callers may still hold this state after eviction. Keep their
394 # timing current under the shared lock; an attribute write cannot
395 # recreate the removed dictionary entry.
396 state.last_request = now
399def reset_for_tests() -> None:
400 """Clear all tracked state. Intended for unit tests only.
402 Both generation counters are rewound inside one ``_emit_lock``
403 section, so a transition decided *after* the reset cannot land between
404 two separate rewinds and be stamped away. It does not order a record
405 decided *before* the reset: such a record still carries a generation
406 above the rewound counter and will stamp ``_emitted_generation`` ahead
407 of ``_state_generation``, dropping transitions until it catches up.
408 Closing that would mean holding ``_emit_lock`` across the decision,
409 which is the serialisation moving emission out of ``_meta_lock``
410 exists to avoid. Callers must not reset while traffic is in flight.
411 """
412 global _all_held_warned_at, _all_held_since
413 global _state_generation, _emitted_generation
414 with _emit_lock:
415 with _meta_lock:
416 _url_state.clear()
417 _all_held_warned_at = None
418 _all_held_since = None
419 _state_generation = 0
420 _emitted_generation = 0