diff --git a/litellm/integrations/generic_api/generic_api_callback.py b/litellm/integrations/generic_api/generic_api_callback.py index 5f9851f595f..17a6ae6bc79 100644 --- a/litellm/integrations/generic_api/generic_api_callback.py +++ b/litellm/integrations/generic_api/generic_api_callback.py @@ -259,12 +259,16 @@ 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() except Exception as e: is_last_attempt = attempt == self.max_retries should_retry = self._should_retry_exception(e) @@ -279,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") @@ -370,6 +376,9 @@ class GenericAPILogger(CustomBatchLogger): batch: Final = tuple(self.log_queue) if not batch: return + failed_events: Final[ # mutable-ok: filled while the send runs and drained into the queue in finally + 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 +398,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 +429,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.extend(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)