mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-03 02:22:24 +00:00
fix(tickerr): fix semaphore leak, remove 500 from error map, fix semaphore test
- Semaphore leak: wrap t.start() in try/except and release _inflight on RuntimeError so the slot is not permanently lost if the OS thread limit is hit - Remove 500 from _ERROR_TYPE_MAP: 500 is a generic internal server error (crash/bug/misconfiguration), not a capacity/overload condition — sending 'overloaded' for 500s would corrupt crowd-sourced signal - Rewrite semaphore cap test to actually invoke real _fire_and_forget with exhausted semaphore and assert threading.Thread is never called - Add test for semaphore release on thread start failure - Update error-type test to explicitly assert 500 is excluded
This commit is contained in:
parent
72d1a93ea1
commit
94a1638630
2 changed files with 32 additions and 20 deletions
|
|
@ -57,12 +57,13 @@ _PROVIDER_MAP: Dict[str, str] = {
|
|||
"nlp_cloud": "nlp_cloud",
|
||||
}
|
||||
|
||||
# Only map codes that unambiguously indicate the specific error type
|
||||
# Only map codes that unambiguously indicate the specific error type.
|
||||
# 500 is intentionally excluded: it is a generic "Internal Server Error" that
|
||||
# indicates a crash or bug, not a capacity/overload condition.
|
||||
_ERROR_TYPE_MAP: Dict[int, str] = {
|
||||
429: "rate_limit",
|
||||
529: "overloaded",
|
||||
503: "overloaded",
|
||||
500: "overloaded",
|
||||
408: "timeout",
|
||||
524: "timeout",
|
||||
401: "auth",
|
||||
|
|
@ -149,7 +150,12 @@ def _fire_and_forget(payload: Dict[str, Any]) -> None:
|
|||
_inflight.release()
|
||||
|
||||
t = threading.Thread(target=_send, daemon=True)
|
||||
t.start()
|
||||
try:
|
||||
t.start()
|
||||
except Exception:
|
||||
# Thread could not be started (e.g. OS thread limit).
|
||||
# Release the slot so future reports are not permanently blocked.
|
||||
_inflight.release()
|
||||
|
||||
|
||||
class TickerrLogger(CustomLogger):
|
||||
|
|
|
|||
|
|
@ -119,8 +119,9 @@ def test_error_type_known_codes():
|
|||
|
||||
|
||||
def test_error_type_no_default_for_unknown_codes():
|
||||
# Unknown codes (400, 404, 502) must NOT map to "overloaded" or any value
|
||||
for code in (400, 404, 502, 422, 301):
|
||||
# 500 is a generic server error (crash/bug), not definitively "overloaded".
|
||||
# Unknown codes must NOT map to any value.
|
||||
for code in (400, 404, 500, 502, 422, 301):
|
||||
assert code not in _ERROR_TYPE_MAP, f"code {code} should not be in _ERROR_TYPE_MAP"
|
||||
|
||||
|
||||
|
|
@ -215,30 +216,35 @@ def test_report_omits_error_type_for_unknown_code():
|
|||
|
||||
|
||||
def test_fire_and_forget_respects_semaphore_cap():
|
||||
"""Reports beyond _MAX_INFLIGHT are dropped silently."""
|
||||
sent = []
|
||||
|
||||
def slow_send(payload):
|
||||
sent.append(payload)
|
||||
# simulate slow network
|
||||
import time
|
||||
time.sleep(0.1)
|
||||
|
||||
# Exhaust the semaphore
|
||||
"""Reports beyond _MAX_INFLIGHT are dropped silently without blocking."""
|
||||
# Exhaust the semaphore by acquiring all slots directly
|
||||
for _ in range(_MAX_INFLIGHT):
|
||||
_inflight.acquire()
|
||||
acquired = _inflight.acquire(blocking=False)
|
||||
assert acquired, "semaphore should have slots available at test start"
|
||||
|
||||
try:
|
||||
# This call should be dropped (semaphore exhausted)
|
||||
with patch("litellm.integrations.tickerr._fire_and_forget"):
|
||||
# With semaphore exhausted, _fire_and_forget must return immediately
|
||||
# without starting a thread (non-blocking acquire fails → early return)
|
||||
with patch("threading.Thread") as mock_thread:
|
||||
_fire_and_forget({"provider": "openai"})
|
||||
# Since semaphore is exhausted, the thread should not be started
|
||||
mock_thread.assert_not_called()
|
||||
finally:
|
||||
# Restore semaphore
|
||||
for _ in range(_MAX_INFLIGHT):
|
||||
_inflight.release()
|
||||
|
||||
|
||||
def test_semaphore_released_on_thread_start_failure():
|
||||
"""If t.start() raises, the semaphore slot must be released so future reports work."""
|
||||
before = _inflight._value
|
||||
|
||||
with patch("threading.Thread") as mock_thread:
|
||||
mock_thread.return_value.start.side_effect = RuntimeError("OS thread limit")
|
||||
_fire_and_forget({"provider": "openai"})
|
||||
|
||||
# Slot must be back to its original value after the exception
|
||||
assert _inflight._value == before
|
||||
|
||||
|
||||
# ── Network failure is silent ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue