Merge 8f24094a10 into 40be7fa1a8

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:

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)