Repository navigation
fix(responses): run stream failure and success hooks on the iterating loop instead of blocking it - #43270
Conversation
… loop instead of blocking it A dropped provider stream on native /v1/responses ran the failure logging through run_async_function from inside the async iterator, which parks the event loop thread on a helper-thread future until every failure callback returns, and never returns when a callback waits on state only that loop can advance. With a running loop the failure handlers (and the completed-stream success deployment hook) are now scheduled as tasks on it, the way chat streaming already does; the sync iterator keeps its blocking path
|
I'll fix CI failures and address comments from users with write access. I'll skip comments containing "(aside)".
|
 |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
Codecov Report✅ All modified and coverable lines are covered by tests. 📢 Thoughts on this report? Let us know! |
…ting loop before propagating Keep the merge-base hook set for the native Responses stream: async_failure_handler plus the executor-thread failure_handler on failure, and the post-call success deployment hook on completion. Inside a running loop the async handler is scheduled as a task on that loop and the async iterator awaits it before re-raising, so the loop is never blocked on a foreign-loop future and the failure is attributed before the router's fallback wrapper re-enters the same logging object. The sync iterator inside a running loop keeps the task fire-and-forget with a strong reference.
| except Exception: | ||
| await self._await_pending_logging() | ||
| raise |
There was a problem hiding this comment.
StopAsyncIteration, making a successful stream wait for its success hook before ending. The event loop remains responsive, but the affected client can still be left waiting.
There was a problem hiding this comment.
Base blocked the whole loop for the same duration, so this is parity. Awaiting also keeps failed-deployment attribution correct during router fallbacks
There was a problem hiding this comment.
You're right. The important distinction is that this await yields to the iterating event loop; it no longer blocks the loop as the old helper-thread path did.
Waiting before propagating the exception is intentional: it preserves the existing ordering and ensures the primary deployment's failure is logged before router fallback can reuse or mutate the logging object. The added regression test also codifies that attribution requirement.
This can still delay the affected consumer if a callback never completes, but that is a per-stream delay rather than a proxy-wide event-loop stall. I’m withdrawing the original finding.
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
1 similar comment
|
bugbot run |
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
There was a problem hiding this comment.
✅ Bugbot reviewed your changes and found no new issues!
Comment @cursor review or bugbot run to trigger another review on this PR
Reviewed by Cursor Bugbot for commit 0be4a64. Configure here.
2 participants
TLDR
Problem this solves:
/v1/responsesstalls the whole proxyHow it solves it:
User Flow
Before: a developer streaming from
/v1/responsesthrough a proxy with an async logging callback (Datadog here) freezes the proxy for everyone the moment one provider stream drops"stream": trueand getresponse.createdplus the first output deltasGET https://litellm-domain/health/livelinessincluded, times out for as long as the failure logging takes (20s with a slow Datadog agent), and the pod can stay frozen until it is restarted when the drop lands while another request's log is still being sentresponse.failedAfter: the same drop reaches their client as
response.failedwhile the proxy keeps serving"stream": trueand getresponse.createdplus the first output deltasGET https://litellm-domain/health/livelinesskeeps answering 200 throughout and the failure event still reaches Datadogresponse.failedonce the drop has been logged and endsLinear ticket
Resolves LIT-8678
Pre-Submission checklist
Please complete all items before asking a LiteLLM maintainer to review your PR
uv run pytest tests/test_litellm/<your_test_file>.py -v. Leave the suites (make test-unit-*,make test-unit) to CI: it finishes in ~15 minutes where a laptop takes an hour or more@greptileaito re-request a review after pushing changes)Screenshots / Proof of Fix
Both legs ran the same rig on one laptop and differ only in the commit the proxy was booted from: the Before leg at the PR's merge base 4adbc13 and the After leg at the PR tip 0be4a64, each in its own worktree (
$WT) with its own venv ($PYis$WT/.venv/bin/python,$MKthe proxy master key). Every case boots one fresh proxy with--num_workers 2on a random free port fromrig/config.yaml, which carries three deployments:gpt-5.4-mini(openai/gpt-5.4-ministraight to api.openai.com),gpt-5.4-mini-flaky-upstream(the same model withapi_basepointing at a local relay that forwards every request to https://api.openai.com and, once armed withtouch rig/out/cut.signal, aborts the proxy's connection mid-body at the next chunk after 20 output deltas) andgpt-5.4-mini-bridge-flaky-upstream(openai/responses/gpt-5.4-minithrough that relay, so/v1/chat/completionsand/v1/messagesride the Responses API stream).litellm_settings.callbacks: ["datadog"]runs in agent mode (LITELLM_DD_AGENT_HOST/LITELLM_DD_AGENT_PORT) against a local stand-in that accepts everyPOST /api/v2/logsand holds its 202 for 20s, a slow Datadog agent, withDD_BATCH_SIZE=1so every event flushes right away. Throughout each case six keep-alive HTTP/1.1 connections each sendGET /health/livelinessonce a second (what a Kubernetes liveness probe and every other caller of that pod see); a connection with no answer for 2s or more is listed as hung with how long it waited. The clients are the OpenAI Python SDK (client.responses.create(..., stream=True)on/v1/responses,client.chat.completions.create(..., stream=True)on/v1/chat/completions, a non-streamclient.responses.create(...)for the requests whose log flush holds the logger) and the Anthropic Python SDK (client.messages.create(..., stream=True)on/v1/messages), all against the real gpt-5.4-mini. Response ids in the non-stream output are shortened toresp_<id>. Every timestamp is local wall clock, cases ran back to backBefore (4adbc13)
Mid-stream provider drop on /v1/responses
cd $WT && .venv/bin/python litellm/proxy/proxy_cli.py --config rig/config.yaml --port 49660 --num_workers 2(pid 22071, /health/liveliness answering)touch rig/out/cut.signal: the relay drops the upstream connection after 20 output deltas)$PY rig/client_stream.py http://127.0.0.1:49660/v1 $MK gpt-5.4-mini-flaky-upstreamclient.responses.create(..., stream=True)):GET http://127.0.0.1:49660/health/livelinessonce a second throughout:Drop while six other requests' log flushes hold the logger
cd $WT && .venv/bin/python litellm/proxy/proxy_cli.py --config rig/config.yaml --port 25322 --num_workers 2(pid 24987, /health/liveliness answering)$PY rig/client_stream.py http://127.0.0.1:25322/v1 $MK gpt-5.4-mini-flaky-upstream(no cut armed yet)for i in 1 2 3 4 5 6; do $PY rig/client_once.py http://127.0.0.1:25322/v1 $MK gpt-5.4-mini & done; wait(each one's success log flush holds the Datadog logger's lock on its worker for 20s)touch rig/out/cut.signal)client.responses.create(..., stream=True)):GET http://127.0.0.1:25322/health/livelinessonce a second throughout:Healthy stream completes
cd $WT && .venv/bin/python litellm/proxy/proxy_cli.py --config rig/config.yaml --port 54144 --num_workers 2(pid 28870, /health/liveliness answering)$PY rig/client_stream.py http://127.0.0.1:54144/v1 $MK gpt-5.4-mini-flaky-upstreamclient.responses.create(..., stream=True)):GET http://127.0.0.1:54144/health/livelinessonce a second throughout:Mid-stream drop on /v1/chat/completions through the Responses bridge
cd $WT && .venv/bin/python litellm/proxy/proxy_cli.py --config rig/config.yaml --port 56906 --num_workers 2(pid 31663, /health/liveliness answering)touch rig/out/cut.signal: the relay drops the upstream connection after 20 output deltas)$PY rig/client_stream_chat.py http://127.0.0.1:56906/v1 $MK gpt-5.4-mini-bridge-flaky-upstreamclient.chat.completions.create(..., stream=True)on a model served through the Responses API):GET http://127.0.0.1:56906/health/livelinessonce a second throughout:Mid-stream drop on /v1/messages through the Responses bridge
cd $WT && .venv/bin/python litellm/proxy/proxy_cli.py --config rig/config.yaml --port 21091 --num_workers 2(pid 36798, /health/liveliness answering)touch rig/out/cut.signal: the relay drops the upstream connection after 20 output deltas)$PY rig/client_stream_messages.py http://127.0.0.1:21091 $MK gpt-5.4-mini-bridge-flaky-upstreamclient.messages.create(..., stream=True)on a model served through the Responses API):GET http://127.0.0.1:21091/health/livelinessonce a second throughout:Drop while one earlier request's log flush holds the logger (the repro's frozen-pod sequence, three attempts)
Attempt 1
cd $WT && .venv/bin/python litellm/proxy/proxy_cli.py --config rig/config.yaml --port 33495 --num_workers 2(pid 61942, /health/liveliness answering)$PY rig/client_stream.py http://127.0.0.1:33495/v1 $MK gpt-5.4-mini-flaky-upstream(no cut armed yet)$PY rig/client_once.py http://127.0.0.1:33495/v1 $MK gpt-5.4-mini(its success log flush holds the Datadog logger's lock on its worker for 20s)touch rig/out/cut.signal); the stream client gets up to 150s from hereclient.responses.create(..., stream=True)):GET http://127.0.0.1:33495/health/livelinessonce a second throughout:Attempt 2
cd $WT && .venv/bin/python litellm/proxy/proxy_cli.py --config rig/config.yaml --port 21627 --num_workers 2(pid 73627, /health/liveliness answering)$PY rig/client_stream.py http://127.0.0.1:21627/v1 $MK gpt-5.4-mini-flaky-upstream(no cut armed yet)$PY rig/client_once.py http://127.0.0.1:21627/v1 $MK gpt-5.4-mini(its success log flush holds the Datadog logger's lock on its worker for 20s)touch rig/out/cut.signal); the stream client gets up to 150s from hereclient.responses.create(..., stream=True)):GET http://127.0.0.1:21627/health/livelinessonce a second throughout:Attempt 3
cd $WT && .venv/bin/python litellm/proxy/proxy_cli.py --config rig/config.yaml --port 38155 --num_workers 2(pid 80554, /health/liveliness answering)$PY rig/client_stream.py http://127.0.0.1:38155/v1 $MK gpt-5.4-mini-flaky-upstream(no cut armed yet)$PY rig/client_once.py http://127.0.0.1:38155/v1 $MK gpt-5.4-mini(its success log flush holds the Datadog logger's lock on its worker for 20s)touch rig/out/cut.signal); the stream client gets up to 150s from hereclient.responses.create(..., stream=True)):GET http://127.0.0.1:38155/health/livelinessonce a second throughout:After (0be4a64)
Mid-stream provider drop on /v1/responses
cd $WT && .venv/bin/python litellm/proxy/proxy_cli.py --config rig/config.yaml --port 23575 --num_workers 2(pid 2497, /health/liveliness answering)touch rig/out/cut.signal: the relay drops the upstream connection after 20 output deltas)$PY rig/client_stream.py http://127.0.0.1:23575/v1 $MK gpt-5.4-mini-flaky-upstreamclient.responses.create(..., stream=True)):GET http://127.0.0.1:23575/health/livelinessonce a second throughout:Drop while six other requests' log flushes hold the logger
cd $WT && .venv/bin/python litellm/proxy/proxy_cli.py --config rig/config.yaml --port 44750 --num_workers 2(pid 10818, /health/liveliness answering)$PY rig/client_stream.py http://127.0.0.1:44750/v1 $MK gpt-5.4-mini-flaky-upstream(no cut armed yet)for i in 1 2 3 4 5 6; do $PY rig/client_once.py http://127.0.0.1:44750/v1 $MK gpt-5.4-mini & done; wait(each one's success log flush holds the Datadog logger's lock on its worker for 20s)touch rig/out/cut.signal)client.responses.create(..., stream=True)):GET http://127.0.0.1:44750/health/livelinessonce a second throughout:Healthy stream completes
cd $WT && .venv/bin/python litellm/proxy/proxy_cli.py --config rig/config.yaml --port 32256 --num_workers 2(pid 23446, /health/liveliness answering)$PY rig/client_stream.py http://127.0.0.1:32256/v1 $MK gpt-5.4-mini-flaky-upstreamclient.responses.create(..., stream=True)):GET http://127.0.0.1:32256/health/livelinessonce a second throughout:Mid-stream drop on /v1/chat/completions through the Responses bridge
cd $WT && .venv/bin/python litellm/proxy/proxy_cli.py --config rig/config.yaml --port 46622 --num_workers 2(pid 28697, /health/liveliness answering)touch rig/out/cut.signal: the relay drops the upstream connection after 20 output deltas)$PY rig/client_stream_chat.py http://127.0.0.1:46622/v1 $MK gpt-5.4-mini-bridge-flaky-upstreamclient.chat.completions.create(..., stream=True)on a model served through the Responses API):GET http://127.0.0.1:46622/health/livelinessonce a second throughout:Mid-stream drop on /v1/messages through the Responses bridge
cd $WT && .venv/bin/python litellm/proxy/proxy_cli.py --config rig/config.yaml --port 57576 --num_workers 2(pid 36647, /health/liveliness answering)touch rig/out/cut.signal: the relay drops the upstream connection after 20 output deltas)$PY rig/client_stream_messages.py http://127.0.0.1:57576 $MK gpt-5.4-mini-bridge-flaky-upstreamclient.messages.create(..., stream=True)on a model served through the Responses API):GET http://127.0.0.1:57576/health/livelinessonce a second throughout:Drop while one earlier request's log flush holds the logger (the repro's frozen-pod sequence, three attempts)
Attempt 1
cd $WT && .venv/bin/python litellm/proxy/proxy_cli.py --config rig/config.yaml --port 23430 --num_workers 2(pid 95090, /health/liveliness answering)$PY rig/client_stream.py http://127.0.0.1:23430/v1 $MK gpt-5.4-mini-flaky-upstream(no cut armed yet)$PY rig/client_once.py http://127.0.0.1:23430/v1 $MK gpt-5.4-mini(its success log flush holds the Datadog logger's lock on its worker for 20s)touch rig/out/cut.signal); the stream client gets up to 150s from hereclient.responses.create(..., stream=True)):GET http://127.0.0.1:23430/health/livelinessonce a second throughout:Attempt 2
cd $WT && .venv/bin/python litellm/proxy/proxy_cli.py --config rig/config.yaml --port 47366 --num_workers 2(pid 9250, /health/liveliness answering)$PY rig/client_stream.py http://127.0.0.1:47366/v1 $MK gpt-5.4-mini-flaky-upstream(no cut armed yet)$PY rig/client_once.py http://127.0.0.1:47366/v1 $MK gpt-5.4-mini(its success log flush holds the Datadog logger's lock on its worker for 20s)touch rig/out/cut.signal); the stream client gets up to 150s from hereclient.responses.create(..., stream=True)):GET http://127.0.0.1:47366/health/livelinessonce a second throughout:Attempt 3
cd $WT && .venv/bin/python litellm/proxy/proxy_cli.py --config rig/config.yaml --port 25140 --num_workers 2(pid 19871, /health/liveliness answering)$PY rig/client_stream.py http://127.0.0.1:25140/v1 $MK gpt-5.4-mini-flaky-upstream(no cut armed yet)$PY rig/client_once.py http://127.0.0.1:25140/v1 $MK gpt-5.4-mini(its success log flush holds the Datadog logger's lock on its worker for 20s)touch rig/out/cut.signal); the stream client gets up to 150s from hereclient.responses.create(..., stream=True)):GET http://127.0.0.1:25140/health/livelinessonce a second throughout:Observations from both legs, with whether this PR causes each, makes it worse, or leaves it alone:
response.failedlands ~40s after the cut on/v1/responses, both legs; PR leaves it aloneType
🐛 Bug Fix
Decisions
asyncio.get_running_loop()rather than by iterator class, so a sync iterator driven from inside a live loop takes the non-blocking path as well, and with no loop the sync path is byte for byte what it wasfailure_handleron the executor); only where and when they run changed. An earlier revision of this PR went throughdispatch_failure_handlers(prefer_async_handlers=True), which only submits the sync handler when sync failure callbacks are configured; it was dropped because it changed which callbacks fire, not just when__anext__before the exception propagates, instead of being fired and forgotten the way chat streaming's_log_stream_failure_and_raisedoes it. That keeps the base's guarantee that logging has completed by the time the caller sees the error, holds only this one request while it waits (every other request keeps being served), and means a client that goes away right after the error cannot cancel the logging out from under a failed-deployment attribution. A bounded wait was considered and rejected as a knob nobody asked forfailure_handleris executor-submitted in afinallyafter the awaited async handler, so the base's order (async first, then sync) holds even when the async handler raises__next__under a loop) cannot await, so it schedules the same task fire-and-forget; tasks live in a module-level strong-reference set with a done callback that logs a failed task at ERROR, where the base swallowed the exceptionend_timeis captured once and handed to both handlers instead of each computing its ownStopAsyncIterationpropagates, since that is anExceptionand goes through the sameexceptCaveats (if any)
Low
__next__path with no running loop still blocks its own thread__next__needs a loop we do not ownresponse.failedwaits for a stalled logging callback with no timeouttest_implemented_namespace_case_modules_remain_importableocr params, main job 2218790, fixed by test: fix stale and state-leaking tests red on scheduled CircleCI #43266test_get_model_infotests, main job 2218778, fixed by test: fix stale and state-leaking tests red on scheduled CircleCI #43266test_e2e_langfuse_callbacks_in_db, main job 2218794test_passthrough_worker_sigkill_leaves_sibling_serving_and_logging, main job 2218749, rewritten by fix(ci): stop stale CI reds, keep unit tests off the host env, retry CyberArk policy conflicts #43294test_price_map_has_no_duplicate_keys, fixed by fix(cost-map): remove duplicate openrouter/perceptron/perceptron-mk1.5 entry #43273Live A/B risk check
Verdict: no dependent path breaks. Three legs ran the same 33 scenarios (28 proxy runs plus 5 SDK runs) against the real gpt-5.4-mini through a forwarding relay, each as one proxy with
--num_workers 2in its own worktree and venv against one Postgres database: Before at the merge base 4adbc13 (proxy port 35950, relay 21618), After at the PR tip 0be4a64 (33108, 52916), and Merged at dcc6517d71, the tip merged into main 5856a36307 (47492, 25310). Every leg printed its worktree, its loadedlitellm.__file__, and whether the head-only symbols_spawn_logging_taskand_await_pending_loggingimport (absent on Before, present on After and Merged). The rig loaded one consumer of every hook kind through the documented extension points: aCustomLoggerwith only sync hooks, one with only async hooks, aCustomGuardrailwhoseasync_post_call_success_deployment_hookraises on purpose, a legacyfailure_callbackcallable, and a legacysuccess_callbackcallable, each appending a row (hook, thread, loop, request keys) per call; spend rows were read back throughGET /spend/logsby request tagBreaking: none observed. Per scenario the hook call counts are identical on all three legs (87 callback rows per leg; a mid-stream cut on
/v1/responsesrecordsasync_log_failure_eventonce and the legacy failure callable once on each leg, an early cut with router fallback records the failure hooks plus the three success hooks on each leg, and the SDK sync iterator under a running loop records the async, legacy, and sync failure hooks once each on each leg, including the variant that exits the interpreter right after the failure). The relay forwarded and cut the same requests on every leg (6GET /v1/models, 29POST /v1/responses, the same cut mode per marker, request bodies the same size Before vs Merged and 1 byte smaller on After from the marker text). Spend rows are 1 per marker per run on every leg. Client verdicts differ only in call ids, marker text, and output lengthBackward incompatible:
response.failedlater: QA case 2 at 59s after the cut and case 6 at 59s, 59s, 60s, against 40s and 39s, 40s, 40s on the base, because the failure logging is now awaited before the error propagates (Decision 3, commit f1407ac); shipped on purpose so the error never outruns its logThreadPoolExecutor-N_0on Before andMainThreadon After and Merged for every async hook. This is the fix itself (commit a2ac961); a callback bound to the serving loop now works, and one that blocks its thread blocks a loop it already blocked beforeLiteLLM:ERROR: streaming_iterator.py:169 - Responses stream post-call success hook failed: ...appears once on After and Merged for the raising-guardrail scenario and never on Before, which swallowed it (commit a2ac961, Decision 5)Regression risk (dependents this run did not reach):
litellm.responses(stream=True)): its code path is the base's, unchanged by the diff, but no leg drove it; it takes a scripted SDK run outside asyncio with the relay cut armedDependency graph (dependents of
_handle_failure,_run_post_success_hooks,ResponsesAPIStreamingIterator.__anext__,__next__, and the new task helpers):/v1/responsesstreaming through the proxy (mid-stream cut, early cut with fallback, clean EOF without a terminal event, healthy stream, healthy stream with a raising guardrail): verified-live, and tested by the five unit tests intests/unit/responses/test_streaming_iterator.py/v1/chat/completionsthrough the Responses bridge (openai/responses/...), mid-stream cut: verified-live (QA case 4 as well)/v1/messagesthrough the Responses bridge, mid-stream cut: verified-live (QA case 5 as well)/v1/responses: verified-live, unchangedlitellm.aresponses(stream=True)underasyncio.run, exiting right after the failure and with a 3s settle: verified-livelitellm.responses(stream=True)iterated inside a running loop, exiting right after, with a 3s settle and a legacy failure callback, and with onlyCustomLoggerinstances: verified-liveCustomLogger, sync-onlyCustomLogger, legacy failure and success callables, guardrail post-call success deployment hook, the Datadog integration underflush_lockcontention in QA cases 2 and 6): verified-liveGET /health/livelinesson six keep-alive connections during a failure: verified-live (0 hung on After in every case; 3, 5, and 4 hung on Before in cases 1, 4, and 5)Not verified:
inputormessages, and the bridge kwargs carry neither on any leg, so the chat and messages bridge scenarios were compared on client verdict, spend rows, and relay counts onlyFinal Attestation
The tests check the right things, including the edge cases, and regressions in the respective real-world customer use-cases are not possible after this PR
0be4a64 passes /live-pr-risk
Note
Medium Risk
Changes async hook scheduling and error propagation timing on the proxy’s hot
/v1/responsesstreaming path; incorrect ordering or awaiting could regress logging, fallbacks, or health under load.Overview
Fixes LIT-8678, where mid-stream
/v1/responsesfailures could block the proxy event loop because failure and post-success deployment hooks ran viarun_async_functionon a helper thread while the iterating loop waited.When
asyncio.get_running_loop()is active, those hooks arecreate_task'd on that loop, tracked on the iterator, and the asyncResponsesAPIStreamingIteratorawaits them in__anext__before the error propagates. Async failure logging still runs before syncfailure_handler(executor), and a singleend_timeis shared. The completed-stream success deployment hook uses the same scheduling; with no running loop, behavior stays on the blocking path.Unit tests assert hooks run on the test loop, failure attribution finishes before the consumer sees the error (router fallback), and sync-only / ordering guarantees when the sync iterator runs under a loop.
Reviewed by Cursor Bugbot for commit 0be4a64. Bugbot is set up for automated code reviews on this repo. Configure here.