From 1e9f49860ef119bc78be91c767b58a8f6e8c1563 Mon Sep 17 00:00:00 2001 From: sun04222 <3242054475@qq.com> Date: Sat, 24 Oct 2026 09:32:40 +0800 Subject: [PATCH 1/7] fix(generic_api): re-queue events whose batch send failed transiently `_send_batch` deleted the batch from `log_queue` in its `finally` block even when the POST failed, so any event queued for an endpoint that was briefly down (the default is max_retries=0) was silently lost with only a log line. This matters for anyone using generic_api as a billing or audit sink. - Transient failures (timeouts, transport errors, 5xx) now put the undelivered events back at the head of the queue, in order, so the next flush retries them. Permanent failures (4xx, payload errors) still drop the events: a re-queue there would grow the queue without bound. - In 'single' format, only the entries that actually failed are re-queued. - `_post_with_retries` now raises on 4xx/5xx responses: previously a real 500 response was treated as a successful send (httpx does not raise for status on its own), and the configured retries never ran for status failures. Fixes #48006 (part 2) AI disclosure: this change was developed with an AI coding assistant and reviewed by a human. --- .../generic_api/generic_api_callback.py | 29 +++- .../generic_api/test_generic_api_callback.py | 126 ++++++++++++++++++ 2 files changed, 154 insertions(+), 1 deletion(-) diff --git a/litellm/integrations/generic_api/generic_api_callback.py b/litellm/integrations/generic_api/generic_api_callback.py index 5f9851f595f..429762c00df 100644 --- a/litellm/integrations/generic_api/generic_api_callback.py +++ b/litellm/integrations/generic_api/generic_api_callback.py @@ -259,12 +259,17 @@ class GenericAPILogger(CustomBatchLogger): total_attempts: Final = self.max_retries + 1 for attempt in range(total_attempts): try: - return await self.async_httpx_client.post( + response = await self.async_httpx_client.post( url=self.endpoint, headers=self.headers, data=data, timeout=self.timeout, ) + # A 4xx/5xx response is a failed send: raise so `_should_retry_exception` can + # retry transient statuses instead of treating the batch as delivered. + if response.status_code >= 400: + response.raise_for_status() + return response except Exception as e: is_last_attempt = attempt == self.max_retries should_retry = self._should_retry_exception(e) @@ -370,6 +375,7 @@ class GenericAPILogger(CustomBatchLogger): batch: Final = tuple(self.log_queue) if not batch: return + failed_events: list[Mapping[str, object] | StandardLoggingPayload] = [] try: verbose_logger.debug( "Generic API Logger - about to flush %s events in '%s' format", len(batch), self.log_format @@ -389,6 +395,9 @@ class GenericAPILogger(CustomBatchLogger): for idx, result in enumerate(responses): if isinstance(result, Exception): verbose_logger.exception("Generic API Logger - Error sending log %s: %s", idx, result) + # Transient failures stay queued for the next flush; permanent ones are dropped. + if self._should_retry_exception(result): + failed_events.append(batch[idx]) else: # result is a Response object verbose_logger.debug( @@ -417,8 +426,26 @@ class GenericAPILogger(CustomBatchLogger): except Exception as e: verbose_logger.exception("Generic API Logger Error sending batch - %s\n%s", e, traceback.format_exc()) + # Transient failures (timeouts, transport errors, 5xx) are re-queued so the next + # flush retries them; permanent failures (4xx, payload errors) are dropped. + if self._should_retry_exception(e): + failed_events = list(batch) finally: del self.log_queue[: len(batch)] + if failed_events: + # Put undelivered events back at the head of the queue, in order, so the + # next flush retries them. Bound the queue like `CustomBatchLogger.flush_queue` + # does for loggers that surface failures: drop the oldest events beyond + # `max_queue_size` so a persistently unreachable endpoint can't grow it unbounded. + self.log_queue[:0] = failed_events + overflow: Final = len(self.log_queue) - self.max_queue_size + if overflow > 0: + del self.log_queue[:overflow] + verbose_logger.warning( + "Generic API Logger: log queue exceeded max_queue_size=%s; dropped %s oldest events.", + self.max_queue_size, + overflow, + ) self.last_flush_time = time.time() def _get_v1_logging_payload(self, kwargs, response_obj, start_time, end_time) -> dict: diff --git a/tests/unit/integrations/generic_api/test_generic_api_callback.py b/tests/unit/integrations/generic_api/test_generic_api_callback.py index afaae02321d..17874f64dda 100644 --- a/tests/unit/integrations/generic_api/test_generic_api_callback.py +++ b/tests/unit/integrations/generic_api/test_generic_api_callback.py @@ -587,6 +587,132 @@ async def test_generic_api_callback_does_not_retry_4xx(): mock_post.assert_called_once() +@pytest.mark.asyncio +async def test_generic_api_batch_failure_requeues_events(): + """ + Test that a transiently failed batch send (transport error, max_retries=0) puts the + events back into the queue instead of dropping them. + """ + test_endpoint = "https://example.com/api/logs" + generic_logger = GenericAPILogger(endpoint=test_endpoint, max_retries=0, retry_delay=0) + + mock_post = AsyncMock() + mock_post.side_effect = httpx.ConnectError("connection refused") + generic_logger.async_httpx_client.post = mock_post + generic_logger.log_queue = [{"event": "lost-on-failure"}] + + await generic_logger.async_send_batch() + + mock_post.assert_called_once() + assert generic_logger.log_queue == [{"event": "lost-on-failure"}] + + +@pytest.mark.asyncio +async def test_generic_api_5xx_response_is_retried_then_delivered(): + """ + Test that a real 5xx response (not just a raised exception) triggers the configured + retries and is delivered on the successful attempt. + """ + test_endpoint = "https://example.com/api/logs" + generic_logger = GenericAPILogger(endpoint=test_endpoint, max_retries=1, retry_delay=0) + + request = httpx.Request("POST", test_endpoint) + mock_post = AsyncMock() + mock_post.side_effect = [httpx.Response(503, request=request), httpx.Response(200, request=request)] + generic_logger.async_httpx_client.post = mock_post + generic_logger.log_queue = [{"event": "5xx-response"}] + + await generic_logger.async_send_batch() + + assert mock_post.call_count == 2 + assert generic_logger.log_queue == [] + + +@pytest.mark.asyncio +async def test_generic_api_5xx_response_after_retries_requeues_events(): + """ + Test that events whose batch keeps failing with 5xx responses stay queued for the next flush. + """ + test_endpoint = "https://example.com/api/logs" + generic_logger = GenericAPILogger(endpoint=test_endpoint, max_retries=1, retry_delay=0) + + request = httpx.Request("POST", test_endpoint) + mock_post = AsyncMock() + mock_post.side_effect = [httpx.Response(503, request=request), httpx.Response(503, request=request)] + generic_logger.async_httpx_client.post = mock_post + generic_logger.log_queue = [{"event": "x"}, {"event": "y"}] + + await generic_logger.async_send_batch() + + assert mock_post.call_count == 2 + assert generic_logger.log_queue == [{"event": "x"}, {"event": "y"}] + + +@pytest.mark.asyncio +async def test_generic_api_4xx_response_drops_events(): + """ + Test that a permanent 4xx response still drops the events (re-queueing them would + grow the queue without bound on an auth/payload error). + """ + test_endpoint = "https://example.com/api/logs" + generic_logger = GenericAPILogger(endpoint=test_endpoint, max_retries=0, retry_delay=0) + + request = httpx.Request("POST", test_endpoint) + mock_post = AsyncMock() + mock_post.side_effect = [httpx.Response(401, request=request)] + generic_logger.async_httpx_client.post = mock_post + generic_logger.log_queue = [{"event": "bad-auth"}] + + await generic_logger.async_send_batch() + + mock_post.assert_called_once() + assert generic_logger.log_queue == [] + + +@pytest.mark.asyncio +async def test_generic_api_single_format_requeues_only_failed_events(): + """ + Test that in 'single' format only the entries whose send failed transiently are + re-queued, in order, while delivered entries are removed. + """ + test_endpoint = "https://example.com/api/logs" + generic_logger = GenericAPILogger(endpoint=test_endpoint, max_retries=0, retry_delay=0, log_format="single") + + request = httpx.Request("POST", test_endpoint) + mock_post = AsyncMock() + mock_post.side_effect = [ + httpx.Response(200, request=request), + httpx.ConnectError("connection refused"), + httpx.Response(200, request=request), + ] + generic_logger.async_httpx_client.post = mock_post + generic_logger.log_queue = [{"event": "a"}, {"event": "b"}, {"event": "c"}] + + await generic_logger.async_send_batch() + + assert mock_post.call_count == 3 + assert generic_logger.log_queue == [{"event": "b"}] + + +@pytest.mark.asyncio +async def test_generic_api_requeue_respects_max_queue_size(): + """ + Test that re-queued events are bounded by max_queue_size: the oldest events + are dropped, mirroring `CustomBatchLogger.flush_queue`'s overflow guard. + """ + test_endpoint = "https://example.com/api/logs" + generic_logger = GenericAPILogger(endpoint=test_endpoint, max_retries=0, retry_delay=0, max_queue_size=3) + + mock_post = AsyncMock() + mock_post.side_effect = [httpx.ConnectError("connection refused")] + generic_logger.async_httpx_client.post = mock_post + generic_logger.log_queue = [{"event": "a"}, {"event": "b"}, {"event": "c"}, {"event": "d"}, {"event": "e"}] + + await generic_logger.async_send_batch() + + assert generic_logger.log_queue == [{"event": "c"}, {"event": "d"}, {"event": "e"}] + + @pytest.fixture(autouse=True) def _vcr_outcome_gate(request, vcr): install_live_call_probe(request, vcr) From 9cd0bbbcad8cba8ac1c22eaac7f0f82639c0d17b Mon Sep 17 00:00:00 2001 From: sun04222 <3242054475@qq.com> Date: Sat, 24 Oct 2026 09:47:39 +0800 Subject: [PATCH 2/7] fix(generic_api): keep the status-code raise inside the try so retries still apply Moving it to `else` (to satisfy the ruff strict gate's TRY300 ceiling) let a 503 response escape the retry loop; only the `return` belongs in `else`. --- litellm/integrations/generic_api/generic_api_callback.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/litellm/integrations/generic_api/generic_api_callback.py b/litellm/integrations/generic_api/generic_api_callback.py index 429762c00df..224bee2a34f 100644 --- a/litellm/integrations/generic_api/generic_api_callback.py +++ b/litellm/integrations/generic_api/generic_api_callback.py @@ -269,7 +269,6 @@ class GenericAPILogger(CustomBatchLogger): # retry transient statuses instead of treating the batch as delivered. if response.status_code >= 400: response.raise_for_status() - return response except Exception as e: is_last_attempt = attempt == self.max_retries should_retry = self._should_retry_exception(e) @@ -284,6 +283,8 @@ class GenericAPILogger(CustomBatchLogger): total_attempts, ) await self._sleep_before_retry(attempt) + else: + return response raise RuntimeError("Generic API Logger retry loop exited unexpectedly") From e0d4931d31274c812c14de243e162c08308a82e6 Mon Sep 17 00:00:00 2001 From: sun04222 <3242054475@qq.com> Date: Sat, 24 Oct 2026 09:56:28 +0800 Subject: [PATCH 3/7] fix(generic_api): move the success return out of the try block (TRY300) ruff_strict_gate enforces a zero-growth ceiling on strict-rule counts against the merge base; the success-path `return response` inside the `try` in `_post_with_retries` added one at generic_api_callback.py:272. Moving it into the `else:` clause keeps the retry semantics identical (same return value, same exception path) and restores the base count. From 07c1bb69fb4ea5b0c0c6a98cae12be66a4e7408d Mon Sep 17 00:00:00 2001 From: sun04222 <3242054475@qq.com> Date: Sat, 24 Oct 2026 09:58:43 +0800 Subject: [PATCH 4/7] fix(generic_api): satisfy the type-discipline gate in _send_batch `failed_events` becomes a bare `Final` local (matching the `tasks` pattern in the same function) and the failure path extends it in place instead of rebinding, so no new LIT001/LIT010 violations are added. --- litellm/integrations/generic_api/generic_api_callback.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/litellm/integrations/generic_api/generic_api_callback.py b/litellm/integrations/generic_api/generic_api_callback.py index 224bee2a34f..93efdfb9637 100644 --- a/litellm/integrations/generic_api/generic_api_callback.py +++ b/litellm/integrations/generic_api/generic_api_callback.py @@ -376,7 +376,7 @@ class GenericAPILogger(CustomBatchLogger): batch: Final = tuple(self.log_queue) if not batch: return - failed_events: list[Mapping[str, object] | StandardLoggingPayload] = [] + failed_events: Final = [] try: verbose_logger.debug( "Generic API Logger - about to flush %s events in '%s' format", len(batch), self.log_format @@ -430,7 +430,7 @@ class GenericAPILogger(CustomBatchLogger): # Transient failures (timeouts, transport errors, 5xx) are re-queued so the next # flush retries them; permanent failures (4xx, payload errors) are dropped. if self._should_retry_exception(e): - failed_events = list(batch) + failed_events.extend(batch) finally: del self.log_queue[: len(batch)] if failed_events: From 383d8e2ab13e116210d18575153ead069842302b Mon Sep 17 00:00:00 2001 From: sun04222 <3242054475@qq.com> Date: Sat, 24 Oct 2026 10:14:19 +0800 Subject: [PATCH 5/7] fix(generic_api): type failed_events so basedpyright stays at its ceiling The bare `Final` left the list unparameterized and its member accesses surfaced two new reportUnknownMemberType errors; annotate it fully and keep the mutability documented with `# mutable-ok:`. --- litellm/integrations/generic_api/generic_api_callback.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/litellm/integrations/generic_api/generic_api_callback.py b/litellm/integrations/generic_api/generic_api_callback.py index 93efdfb9637..4e2df7dc014 100644 --- a/litellm/integrations/generic_api/generic_api_callback.py +++ b/litellm/integrations/generic_api/generic_api_callback.py @@ -376,7 +376,7 @@ class GenericAPILogger(CustomBatchLogger): batch: Final = tuple(self.log_queue) if not batch: return - failed_events: Final = [] + failed_events: Final[list[Mapping[str, object] | StandardLoggingPayload]] = [] # mutable-ok: filled while the send runs and drained into the queue in finally try: verbose_logger.debug( "Generic API Logger - about to flush %s events in '%s' format", len(batch), self.log_format From bf0be8b237e922f539fb008b444768a94a1376cd Mon Sep 17 00:00:00 2001 From: sun04222 <3242054475@qq.com> Date: Sat, 24 Oct 2026 10:21:15 +0800 Subject: [PATCH 6/7] fix(generic_api): ruff-format the restored failed_events annotation The annotation restore left the line in a form `ruff format --check` wants to reformat (CI exit 123). Applying ruff format; no semantic change. --- litellm/integrations/generic_api/generic_api_callback.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/litellm/integrations/generic_api/generic_api_callback.py b/litellm/integrations/generic_api/generic_api_callback.py index 4e2df7dc014..cbdc3ecd205 100644 --- a/litellm/integrations/generic_api/generic_api_callback.py +++ b/litellm/integrations/generic_api/generic_api_callback.py @@ -376,7 +376,9 @@ class GenericAPILogger(CustomBatchLogger): batch: Final = tuple(self.log_queue) if not batch: return - failed_events: Final[list[Mapping[str, object] | StandardLoggingPayload]] = [] # mutable-ok: filled while the send runs and drained into the queue in finally + failed_events: Final[ + list[Mapping[str, object] | StandardLoggingPayload] + ] = [] # mutable-ok: filled while the send runs and drained into the queue in finally try: verbose_logger.debug( "Generic API Logger - about to flush %s events in '%s' format", len(batch), self.log_format From 8f24094a108ccf2f10f27f778098076ddf265f87 Mon Sep 17 00:00:00 2001 From: sun04222 <3242054475@qq.com> Date: Sat, 24 Oct 2026 10:24:40 +0800 Subject: [PATCH 7/7] fix(generic_api): keep the mutable-ok suppression on the annotated line ruff format wraps the long annotation; the suppression comment belongs on the first line of the statement, where the LIT001 violation is emitted. --- litellm/integrations/generic_api/generic_api_callback.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/litellm/integrations/generic_api/generic_api_callback.py b/litellm/integrations/generic_api/generic_api_callback.py index cbdc3ecd205..17a6ae6bc79 100644 --- a/litellm/integrations/generic_api/generic_api_callback.py +++ b/litellm/integrations/generic_api/generic_api_callback.py @@ -376,9 +376,9 @@ class GenericAPILogger(CustomBatchLogger): batch: Final = tuple(self.log_queue) if not batch: return - failed_events: Final[ + failed_events: Final[ # mutable-ok: filled while the send runs and drained into the queue in finally list[Mapping[str, object] | StandardLoggingPayload] - ] = [] # mutable-ok: filled while the send runs and drained into the queue in finally + ] = [] try: verbose_logger.debug( "Generic API Logger - about to flush %s events in '%s' format", len(batch), self.log_format