From 5480ddf3d727d0bcf2b160fa4d15f6933eb7bccc Mon Sep 17 00:00:00 2001 From: Charan Rathore Date: Mon, 28 Sep 2026 10:49:15 +0530 Subject: [PATCH 1/3] fix(proxy): requeue spend logs on local pre-write failures --- litellm/proxy/utils.py | 15 +++- .../test_proxy_update_spend.py | 74 +++++++++++++++++++ 2 files changed, 87 insertions(+), 2 deletions(-) diff --git a/litellm/proxy/utils.py b/litellm/proxy/utils.py index b336ce1fa27..082552c58e1 100644 --- a/litellm/proxy/utils.py +++ b/litellm/proxy/utils.py @@ -7288,7 +7288,12 @@ class ProxyUpdateSpend: if not base_url.endswith("/"): base_url += "/" verbose_proxy_logger.debug("base_url: %s", base_url) - json_data = json.dumps(logs_to_process) + try: + json_data = json.dumps(logs_to_process) + except (TypeError, ValueError): + # No external request has been sent. The batch is safe to replay. + await requeue_spend_logs(prisma_client, proxy_logging_obj, logs_to_process) + raise response = await db_writer_client.post( url=base_url + "spend/update", data=json_data, @@ -7301,7 +7306,13 @@ class ProxyUpdateSpend: else: for j in range(0, len(logs_to_process), BATCH_SIZE): batch = logs_to_process[j : j + BATCH_SIZE] - batch_with_dates = [prisma_client.jsonify_object({**entry}) for entry in batch] + try: + batch_with_dates = [prisma_client.jsonify_object({**entry}) for entry in batch] + except (TypeError, ValueError): + # This batch has not reached Prisma. Earlier batches may + # already have committed, so replay only this tail. + await requeue_spend_logs(prisma_client, proxy_logging_obj, logs_to_process[j:]) + raise isolation_budget = MAX_SPEND_LOG_ISOLATION_FAILURES_PER_BATCH for statement_rows in spend_log_write_batches( batch_with_dates, diff --git a/tests/test_litellm/proxy/utils/prisma_and_spend/test_proxy_update_spend.py b/tests/test_litellm/proxy/utils/prisma_and_spend/test_proxy_update_spend.py index 7099101db1c..89afe852427 100644 --- a/tests/test_litellm/proxy/utils/prisma_and_spend/test_proxy_update_spend.py +++ b/tests/test_litellm/proxy/utils/prisma_and_spend/test_proxy_update_spend.py @@ -917,3 +917,77 @@ async def test_update_spend_logs_parks_failed_batch_in_redis_with_wire_safe_date parked = await buffer.get_spend_logs_from_redis_buffer(limit=10) assert mock_prisma_client.spend_log_transactions == [] assert [(row["request_id"], row["startTime"]) for row in parked] == [("a", started.isoformat())] + + +@pytest.mark.asyncio +async def test_spend_log_serialization_failure_requeues_only_unwritten_tail( + mock_prisma_client: Any, make_spend_log_row: Any, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.delenv("SPEND_LOGS_URL", raising=False) + rows = [make_spend_log_row(request_id=f"committed-{i}") for i in range(1000)] + rows.append(make_spend_log_row(request_id="unwritten")) + original_jsonify = mock_prisma_client.jsonify_object + + def jsonify(row: Any) -> Any: + if row["request_id"] == "unwritten": + raise TypeError("bad local serialization") + return original_jsonify(row) + + mock_prisma_client.jsonify_object = jsonify + mock_prisma_client.db.litellm_spendlogs.create_many = AsyncMock() + proxy_logging = MagicMock() + proxy_logging.failure_handler = AsyncMock() + with pytest.raises(TypeError, match="bad local serialization"): + await ProxyUpdateSpend.update_spend_logs( + n_retry_times=0, + prisma_client=mock_prisma_client, + db_writer_client=None, + proxy_logging_obj=proxy_logging, + logs_to_process=rows, + ) + assert mock_prisma_client.db.litellm_spendlogs.create_many.await_count >= 1 + assert [row["request_id"] for row in mock_prisma_client.spend_log_transactions] == ["unwritten"] + + +@pytest.mark.asyncio +async def test_external_spend_log_preflight_failure_requeues_without_post( + mock_prisma_client: Any, make_spend_log_row: Any, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("SPEND_LOGS_URL", "http://writer.invalid") + rows = [make_spend_log_row(request_id="unwritten")] + rows[0]["unserializable"] = object() + writer = MagicMock() + writer.post = AsyncMock() + proxy_logging = MagicMock() + proxy_logging.failure_handler = AsyncMock() + with pytest.raises(TypeError): + await ProxyUpdateSpend.update_spend_logs( + n_retry_times=0, + prisma_client=mock_prisma_client, + db_writer_client=writer, + proxy_logging_obj=proxy_logging, + logs_to_process=rows, + ) + writer.post.assert_not_awaited() + assert [row["request_id"] for row in mock_prisma_client.spend_log_transactions] == ["unwritten"] + + +@pytest.mark.asyncio +async def test_external_spend_log_post_error_is_not_replayed( + mock_prisma_client: Any, make_spend_log_row: Any, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("SPEND_LOGS_URL", "http://writer.invalid") + writer = MagicMock() + writer.post = AsyncMock(side_effect=ValueError("uncertain delivery")) + proxy_logging = MagicMock() + proxy_logging.failure_handler = AsyncMock() + with pytest.raises(ValueError, match="uncertain delivery"): + await ProxyUpdateSpend.update_spend_logs( + n_retry_times=0, + prisma_client=mock_prisma_client, + db_writer_client=writer, + proxy_logging_obj=proxy_logging, + logs_to_process=[make_spend_log_row(request_id="maybe-committed")], + ) + writer.post.assert_awaited_once() + assert mock_prisma_client.spend_log_transactions == [] From 023d82f23ffc41ce02ad7d49cc948d49e7529257 Mon Sep 17 00:00:00 2001 From: Charan Rathore Date: Mon, 28 Sep 2026 10:51:12 +0530 Subject: [PATCH 2/3] fix(proxy): do not retry ambiguous spend writer POSTs --- litellm/proxy/utils.py | 6 ++++++ .../test_proxy_update_spend.py | 20 +++++++++++++++++++ 2 files changed, 26 insertions(+) diff --git a/litellm/proxy/utils.py b/litellm/proxy/utils.py index 082552c58e1..a5a9e997677 100644 --- a/litellm/proxy/utils.py +++ b/litellm/proxy/utils.py @@ -7282,6 +7282,7 @@ class ProxyUpdateSpend: start_time: Final = time.time() try: for i in range(n_retry_times + 1): + external_post_attempted = False try: base_url = os.getenv("SPEND_LOGS_URL", None) if len(logs_to_process) > 0 and base_url is not None and db_writer_client is not None: @@ -7294,6 +7295,7 @@ class ProxyUpdateSpend: # No external request has been sent. The batch is safe to replay. await requeue_spend_logs(prisma_client, proxy_logging_obj, logs_to_process) raise + external_post_attempted = True response = await db_writer_client.post( url=base_url + "spend/update", data=json_data, @@ -7336,6 +7338,10 @@ class ProxyUpdateSpend: ) break except Exception as e: + if external_post_attempted: + # Even a transport error can arrive after the remote writer + # committed. Retrying or requeueing could duplicate spend. + raise if not _is_transient_spend_log_write_error(e): if PrismaDBExceptionHandler.is_prisma_error(e): await requeue_spend_logs(prisma_client, proxy_logging_obj, logs_to_process) diff --git a/tests/test_litellm/proxy/utils/prisma_and_spend/test_proxy_update_spend.py b/tests/test_litellm/proxy/utils/prisma_and_spend/test_proxy_update_spend.py index 89afe852427..9e1ec346b51 100644 --- a/tests/test_litellm/proxy/utils/prisma_and_spend/test_proxy_update_spend.py +++ b/tests/test_litellm/proxy/utils/prisma_and_spend/test_proxy_update_spend.py @@ -991,3 +991,23 @@ async def test_external_spend_log_post_error_is_not_replayed( ) writer.post.assert_awaited_once() assert mock_prisma_client.spend_log_transactions == [] + + +@pytest.mark.asyncio +async def test_external_spend_log_transport_error_is_not_retried_or_requeued( + mock_prisma_client: Any, make_spend_log_row: Any, monkeypatch: pytest.MonkeyPatch +) -> None: + import httpx + + monkeypatch.setenv("SPEND_LOGS_URL", "http://writer.invalid") + writer = MagicMock() + writer.post = AsyncMock(side_effect=httpx.ReadError("response lost after send")) + proxy_logging = MagicMock() + proxy_logging.failure_handler = AsyncMock() + with pytest.raises(httpx.ReadError): + await ProxyUpdateSpend.update_spend_logs( + n_retry_times=2, prisma_client=mock_prisma_client, db_writer_client=writer, + proxy_logging_obj=proxy_logging, logs_to_process=[make_spend_log_row(request_id="maybe-committed")], + ) + writer.post.assert_awaited_once() + assert mock_prisma_client.spend_log_transactions == [] From d55efe1e882d594c4c416238d440dc226b688360 Mon Sep 17 00:00:00 2001 From: Charan Rathore Date: Mon, 28 Sep 2026 11:07:04 +0530 Subject: [PATCH 3/3] test(proxy): run spend-log failure cases in CI shard --- .../test_proxy_update_spend.py | 94 ------------------- tests/unit/proxy/test_update_spend.py | 69 ++++++++++++++ 2 files changed, 69 insertions(+), 94 deletions(-) diff --git a/tests/test_litellm/proxy/utils/prisma_and_spend/test_proxy_update_spend.py b/tests/test_litellm/proxy/utils/prisma_and_spend/test_proxy_update_spend.py index 9e1ec346b51..7099101db1c 100644 --- a/tests/test_litellm/proxy/utils/prisma_and_spend/test_proxy_update_spend.py +++ b/tests/test_litellm/proxy/utils/prisma_and_spend/test_proxy_update_spend.py @@ -917,97 +917,3 @@ async def test_update_spend_logs_parks_failed_batch_in_redis_with_wire_safe_date parked = await buffer.get_spend_logs_from_redis_buffer(limit=10) assert mock_prisma_client.spend_log_transactions == [] assert [(row["request_id"], row["startTime"]) for row in parked] == [("a", started.isoformat())] - - -@pytest.mark.asyncio -async def test_spend_log_serialization_failure_requeues_only_unwritten_tail( - mock_prisma_client: Any, make_spend_log_row: Any, monkeypatch: pytest.MonkeyPatch -) -> None: - monkeypatch.delenv("SPEND_LOGS_URL", raising=False) - rows = [make_spend_log_row(request_id=f"committed-{i}") for i in range(1000)] - rows.append(make_spend_log_row(request_id="unwritten")) - original_jsonify = mock_prisma_client.jsonify_object - - def jsonify(row: Any) -> Any: - if row["request_id"] == "unwritten": - raise TypeError("bad local serialization") - return original_jsonify(row) - - mock_prisma_client.jsonify_object = jsonify - mock_prisma_client.db.litellm_spendlogs.create_many = AsyncMock() - proxy_logging = MagicMock() - proxy_logging.failure_handler = AsyncMock() - with pytest.raises(TypeError, match="bad local serialization"): - await ProxyUpdateSpend.update_spend_logs( - n_retry_times=0, - prisma_client=mock_prisma_client, - db_writer_client=None, - proxy_logging_obj=proxy_logging, - logs_to_process=rows, - ) - assert mock_prisma_client.db.litellm_spendlogs.create_many.await_count >= 1 - assert [row["request_id"] for row in mock_prisma_client.spend_log_transactions] == ["unwritten"] - - -@pytest.mark.asyncio -async def test_external_spend_log_preflight_failure_requeues_without_post( - mock_prisma_client: Any, make_spend_log_row: Any, monkeypatch: pytest.MonkeyPatch -) -> None: - monkeypatch.setenv("SPEND_LOGS_URL", "http://writer.invalid") - rows = [make_spend_log_row(request_id="unwritten")] - rows[0]["unserializable"] = object() - writer = MagicMock() - writer.post = AsyncMock() - proxy_logging = MagicMock() - proxy_logging.failure_handler = AsyncMock() - with pytest.raises(TypeError): - await ProxyUpdateSpend.update_spend_logs( - n_retry_times=0, - prisma_client=mock_prisma_client, - db_writer_client=writer, - proxy_logging_obj=proxy_logging, - logs_to_process=rows, - ) - writer.post.assert_not_awaited() - assert [row["request_id"] for row in mock_prisma_client.spend_log_transactions] == ["unwritten"] - - -@pytest.mark.asyncio -async def test_external_spend_log_post_error_is_not_replayed( - mock_prisma_client: Any, make_spend_log_row: Any, monkeypatch: pytest.MonkeyPatch -) -> None: - monkeypatch.setenv("SPEND_LOGS_URL", "http://writer.invalid") - writer = MagicMock() - writer.post = AsyncMock(side_effect=ValueError("uncertain delivery")) - proxy_logging = MagicMock() - proxy_logging.failure_handler = AsyncMock() - with pytest.raises(ValueError, match="uncertain delivery"): - await ProxyUpdateSpend.update_spend_logs( - n_retry_times=0, - prisma_client=mock_prisma_client, - db_writer_client=writer, - proxy_logging_obj=proxy_logging, - logs_to_process=[make_spend_log_row(request_id="maybe-committed")], - ) - writer.post.assert_awaited_once() - assert mock_prisma_client.spend_log_transactions == [] - - -@pytest.mark.asyncio -async def test_external_spend_log_transport_error_is_not_retried_or_requeued( - mock_prisma_client: Any, make_spend_log_row: Any, monkeypatch: pytest.MonkeyPatch -) -> None: - import httpx - - monkeypatch.setenv("SPEND_LOGS_URL", "http://writer.invalid") - writer = MagicMock() - writer.post = AsyncMock(side_effect=httpx.ReadError("response lost after send")) - proxy_logging = MagicMock() - proxy_logging.failure_handler = AsyncMock() - with pytest.raises(httpx.ReadError): - await ProxyUpdateSpend.update_spend_logs( - n_retry_times=2, prisma_client=mock_prisma_client, db_writer_client=writer, - proxy_logging_obj=proxy_logging, logs_to_process=[make_spend_log_row(request_id="maybe-committed")], - ) - writer.post.assert_awaited_once() - assert mock_prisma_client.spend_log_transactions == [] diff --git a/tests/unit/proxy/test_update_spend.py b/tests/unit/proxy/test_update_spend.py index ebe505b3d60..39b285be21d 100644 --- a/tests/unit/proxy/test_update_spend.py +++ b/tests/unit/proxy/test_update_spend.py @@ -321,3 +321,72 @@ async def test_update_spend_logs_multiple_batches_with_failure(): # Verify all logs were cleared from transactions assert len(prisma_client.spend_log_transactions) == 0 + + +# These tests live in the proxy-db-db-and-spend CI shard, unlike the separate +# prisma_and_spend suite, so their branch coverage contributes to codecov. +def _spend_log_lifetime_fakes(): + client = MockPrismaClient() + logging = create_mock_proxy_logging() + logging.db_spend_update_writer.redis_update_buffer.store_spend_logs_in_redis = AsyncMock(return_value=False) + return client, logging + + +@pytest.mark.asyncio +async def test_spend_log_pre_write_failure_requeues_only_unwritten_tail(monkeypatch): + from litellm.proxy.utils import ProxyUpdateSpend + + monkeypatch.delenv("SPEND_LOGS_URL", raising=False) + client, logging = _spend_log_lifetime_fakes() + rows = [{"request_id": f"committed-{i}"} for i in range(1000)] + rows.append({"request_id": "unwritten"}) + + def jsonify(row): + if row["request_id"] == "unwritten": + raise TypeError("local conversion failed") + return row + + client.jsonify_object = jsonify + with pytest.raises(TypeError, match="local conversion failed"): + await ProxyUpdateSpend.update_spend_logs( + n_retry_times=0, prisma_client=client, db_writer_client=None, + proxy_logging_obj=logging, logs_to_process=rows, + ) + assert client.db.litellm_spendlogs.create_many.await_count >= 1 + assert client.spend_log_transactions == [{"request_id": "unwritten"}] + + +@pytest.mark.asyncio +async def test_spend_log_external_preflight_requeues_without_post(monkeypatch): + from litellm.proxy.utils import ProxyUpdateSpend + + monkeypatch.setenv("SPEND_LOGS_URL", "http://writer.invalid") + client, logging = _spend_log_lifetime_fakes() + rows = [{"request_id": "unwritten", "bad": object()}] + writer = MagicMock() + writer.post = AsyncMock() + with pytest.raises(TypeError): + await ProxyUpdateSpend.update_spend_logs( + n_retry_times=0, prisma_client=client, db_writer_client=writer, + proxy_logging_obj=logging, logs_to_process=rows, + ) + writer.post.assert_not_awaited() + assert client.spend_log_transactions == rows + + +@pytest.mark.asyncio +@pytest.mark.parametrize("failure", [ValueError("uncertain delivery"), httpx.ReadError("response lost after send")]) +async def test_spend_log_external_post_failure_never_retries_or_requeues(monkeypatch, failure): + from litellm.proxy.utils import ProxyUpdateSpend + + monkeypatch.setenv("SPEND_LOGS_URL", "http://writer.invalid") + client, logging = _spend_log_lifetime_fakes() + writer = MagicMock() + writer.post = AsyncMock(side_effect=failure) + with pytest.raises(type(failure)): + await ProxyUpdateSpend.update_spend_logs( + n_retry_times=2, prisma_client=client, db_writer_client=writer, + proxy_logging_obj=logging, logs_to_process=[{"request_id": "maybe-committed"}], + ) + writer.post.assert_awaited_once() + assert client.spend_log_transactions == []