Repository navigation
(feat): add 0 downtime support during pod restarts - #964
Conversation
|
Important Review skippedAuto incremental reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
WalkthroughThe PR adds graceful-drain handling for DragonTTS, including readiness and drain endpoints, request tracking, Clairvoyance notifications, Slack alerts, and shutdown waiting. It also updates voice parameter forwarding, ElevenLabs speed settings, and Gemini resolution tests. ChangesGraceful drain integration
Voice provider updates
Gemini resolution coverage
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Kubernetes
participant FastAPI
participant Clairvoyance
participant Slack
participant ShutdownLifespan
Kubernetes->>FastAPI: call /drain
FastAPI->>Clairvoyance: request kill_switch
Clairvoyance-->>FastAPI: return acknowledgement
FastAPI->>Slack: send drain alert
ShutdownLifespan->>FastAPI: stop accepting work
ShutdownLifespan->>ShutdownLifespan: wait for in-flight requests
ShutdownLifespan-->>ShutdownLifespan: close cache and providers
Possibly related PRs
Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1🛠️ Fix failing CI checks 💡
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 8
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@app/ai/voice/tts/dragontts.py`:
- Around line 58-69: The DragonTTS fallback must not send enable_ssml_parsing
through the unsupported HTTP synth path. Update the synth() fallback routing to
use an SSML-capable fallback when SSML is enabled, or bypass the HTTP fallback
entirely for those requests, while preserving the existing fallback behavior for
non-SSML synthesis.
In `@dragontts/app/api/v1/health.py`:
- Around line 22-38: Add explicit return type annotations to ready and drain in
dragontts/app/api/v1/health.py, and annotate call_next plus the return type of
_track_inflight in dragontts/app/main.py using the project’s established types.
- Around line 32-37: Restrict the `drain` endpoint in `health.router` to
authenticated service callers or the pod-local preStop caller before invoking
`engage_drain()`. Add the route-level authorization or request-origin validation
within `drain`, while preserving its existing drain behavior for authorized
requests.
In `@dragontts/app/core/config.py`:
- Around line 243-251: Update the graceful-drain documentation comment in
config.py to state that startup does not automatically send action=restore and
that restoring clairvoyance caching is a manual operation. Keep the existing
shutdown sequence and configuration behavior documented accurately.
In `@dragontts/app/drain.py`:
- Around line 139-167: Update engage_drain() to check enable_graceful_drain
before any drain side effects, returning without calling mark_draining(),
notify_clairvoyance(), or send_kill_switch_slack() when the feature is disabled;
preserve the existing ordered drain flow when enabled.
In `@dragontts/app/main.py`:
- Around line 192-200: Replace the _track_inflight middleware’s
call_next/finally tracking with ASGI send-wrapper logic that keeps _inflight
incremented until the final http.response.body message has more_body=False.
Ensure aborted or exceptional responses still decrement exactly once, and add a
regression test confirming an open streaming response remains counted after
headers are sent.
In `@tests/test_llm_gemini_resolution.py`:
- Line 20: Annotate every function signature in
tests/test_llm_gemini_resolution.py, including _valid_config and all test
functions, with explicit parameter and return types. Type each monkeypatch
fixture and parametrized argument/value, using appropriate existing typing
symbols while preserving the current test behavior.
- Line 22: Update the Gemini test setup around LLMProvider.GEMINI by either
adding the GEMINI enum member and wiring the resolver’s dispatch to
_resolve_gemini, or replacing the test with a currently supported provider
contract; ensure the test only references a provider that exists and is handled
by the resolver.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: d12bc5e1-7fcc-40fb-859e-18ec4820508d
📒 Files selected for processing (8)
app/ai/voice/tts/dragontts.pydragontts/.env.exampledragontts/app/api/v1/health.pydragontts/app/core/config.pydragontts/app/drain.pydragontts/app/main.pydragontts/app/providers/elevenlabs.pytests/test_llm_gemini_resolution.py
| # enable_ssml_parsing MUST be here: without it, the cached/dragontts path drops | ||
| # the flag and ElevenLabs parses <break/> as plain text (SSML only "worked" on the | ||
| # direct ElevenLabs path, which forwards it separately). It selects a distinct | ||
| # warm SSML socket + cache key inside DragonTTS. | ||
| _PARAM_FIELDS = ( | ||
| "speed", | ||
| "volume", | ||
| "emotion", | ||
| "pitch", | ||
| "style_prompt", | ||
| "enable_ssml_parsing", | ||
| ) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/usr/bin/env bash
set -euo pipefail
rg -n -C 5 \
'enable_ssml_parsing|async def synth|async def stream_synth|self\.synth\(' \
app/ai/voice/tts/dragontts.py \
dragontts/app/providers/elevenlabs.py
for url in \
'https://elevenlabs.io/docs/api-reference/text-to-speech/convert' \
'https://elevenlabs.io/docs/api-reference/text-to-speech/v-1-text-to-speech-voice-id-multi-stream-input'; do
printf '\n=== %s ===\n' "$url"
curl -fsSL "$url" | rg -n 'enable_ssml_parsing|voice_settings' || true
doneRepository: juspay/clairvoyance
Length of output: 50378
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '=== dragontts outline ===\n'
ast-grep outline app/ai/voice/tts/dragontts.py --view expanded || true
printf '\n=== dragontts relevant lines ===\n'
sed -n '1,140p' app/ai/voice/tts/dragontts.py | nl -ba
printf '\n=== elevenlabs outline ===\n'
ast-grep outline dragontts/app/providers/elevenlabs.py --view expanded || true
printf '\n=== elevenlabs relevant synth/voice sections ===\n'
sed -n '160,320p' dragontts/app/providers/elevenlabs.py | nl -ba
printf '\n=== focused source search ===\n'
rg -n -C 4 \
'_PARAM_FIELDS|_collect_params|enable_ssml_parsing|stream_synth|self\.synth\(' \
app/ai/voice/tts/dragontts.py \
dragontts/app/providers/elevenlabs.py || true
printf '\n=== ElevenLabs references in tracked files ===\n'
rg -n -C 3 'elevenlabs|enable_ssml_parsing|stream-input|multi-stream|residency.elevenlabs|api.in\.residency' \
$(git ls-files '*.py' '*.yaml' '*.yml' '*.md' '*.json') || trueRepository: juspay/clairvoyance
Length of output: 1313
🌐 Web query:
ElevenLabs text-to-speech convert API enable_ssml_parsing create speech endpoint documentation
💡 Result:
In the ElevenLabs API, the enable_ssml_parsing parameter is primarily associated with the WebSocket streaming endpoint rather than the standard REST /v1/text-to-speech/{voice_id} (Create speech) endpoint [1][2]. When using the WebSocket stream input for text-to-speech, enabling enable_ssml_parsing allows the API to process SSML tags, such as those used for breaks or custom pronunciations [3]. Note that using this feature may increase latency during audio generation [3]. For the standard REST "Create speech" (/v1/text-to-speech/{voice_id}) endpoint, documentation does not explicitly list enable_ssml_parsing as a supported parameter [4][5][6]. However, ElevenLabs has indicated in support documentation that certain SSML-like tags, such as the tag, can still be used to create pauses in speech when generating audio via the API [7]. If you require advanced SSML support, the WebSocket interface is the documented method for enabling formal SSML parsing [2][3].
Citations:
- 1: https://elevenlabs.io/docs/api-reference/text-to-speech.md
- 2: https://elevenlabs.io/docs/api-reference/text-to-speech/v-1-text-to-speech-voice-id-stream-input
- 3: https://elevenlabslib.readthedocs.io/en/latest/source/api/helpers.html
- 4: https://elevenlabs.io/docs/api-reference/text-to-speech/convert
- 5: https://elevenlabs.io/docs/api-reference/text-to-speech/convert.mdx
- 6: https://elevenlabs.io/docs/api-reference/text-to-speech/convert/llms.txt
- 7: https://help.elevenlabs.io/hc/en-us/sections/14163158308369-API
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '=== dragontts relevant lines ===\n'
sed -n '1,340p' app/ai/voice/tts/dragontts.py
printf '\n=== elevenlabs synth/voice sections ===\n'
sed -n '160,320p' dragontts/app/providers/elevenlabs.py
printf '\n=== focused source search ===\n'
rg -n -C 4 \
'_PARAM_FIELDS|_collect_params|enable_ssml_parsing|stream_synth|self\.synth\(' \
app/ai/voice/tts/dragontts.py dragontts/app/providers/elevenlabs.py
printf '\n=== ElevenLabs references in tracked files ===\n'
rg -n -C 3 'elevenlabs|enable_ssml_parsing|stream-input|multi-stream|residency.elevenlabs|api.in\.residency' \
$(git ls-files '*.py' '*.yaml' '*.yml' '*.md' '*.json') || trueRepository: juspay/clairvoyance
Length of output: 50376
Keep SSML requests off the unsupported HTTP fallback.
enable_ssml_parsing is now included in DragonTTS params, so ElevenLabs SSML streams can fall back to synth() and send it on /v1/text-to-speech/{voice_id}. That schema does not list enable_ssml_parsing as a supported parameter, while the WebSocket multi-stream-input path does. Route SSML requests to a supported fallback, or skip the HTTP fallback when SSML is enabled.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@app/ai/voice/tts/dragontts.py` around lines 58 - 69, The DragonTTS fallback
must not send enable_ssml_parsing through the unsupported HTTP synth path.
Update the synth() fallback routing to use an SSML-capable fallback when SSML is
enabled, or bypass the HTTP fallback entirely for those requests, while
preserving the existing fallback behavior for non-SSML synthesis.
| @router.get("/ready") | ||
| async def ready(): | ||
| """Readiness probe. 503 while draining so k8s stops routing NEW traffic to | ||
| this pod (complements endpoint removal on termination). Stays 200 on the | ||
| liveness probe (/health) so k8s doesn't restart the pod mid-drain.""" | ||
| if is_draining(): | ||
| raise HTTPException(status_code=503, detail="draining") | ||
| return {"status": "ready"} | ||
|
|
||
|
|
||
| @router.get("/drain") | ||
| async def drain(): | ||
| """k8s preStop hook. Marks the pod draining, fires the kill-switch Slack | ||
| alert, and tells clairvoyance to bypass us BEFORE SIGTERM arrives — then | ||
| the lifespan shutdown waits for in-flight requests to finish. See app.drain.""" | ||
| await engage_drain() | ||
| return {"status": "draining"} |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Add required type annotations to new function signatures.
dragontts/app/api/v1/health.py#L22-L38: add return annotations toreadyanddrain.dragontts/app/main.py#L192-L200: annotatecall_nextand the return type of_track_inflight.
As per coding guidelines, “Include required type hints on all function signatures.”
📍 Affects 2 files
dragontts/app/api/v1/health.py#L22-L38(this comment)dragontts/app/main.py#L192-L200
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@dragontts/app/api/v1/health.py` around lines 22 - 38, Add explicit return
type annotations to ready and drain in dragontts/app/api/v1/health.py, and
annotate call_next plus the return type of _track_inflight in
dragontts/app/main.py using the project’s established types.
Source: Coding guidelines
| @router.get("/drain") | ||
| async def drain(): | ||
| """k8s preStop hook. Marks the pod draining, fires the kill-switch Slack | ||
| alert, and tells clairvoyance to bypass us BEFORE SIGTERM arrives — then | ||
| the lifespan shutdown waits for in-flight requests to finish. See app.drain.""" | ||
| await engage_drain() |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect router-level dependencies and application router registration.
ast-grep outline dragontts/app/api/v1/health.py --items all
rg -n -C 3 'APIRouter\(|include_router|dependencies=' dragontts/app --glob '*.py'
# Inspect deployment manifests for the preStop caller and network exposure.
fd -HI -t f -e yaml -e yml . | xargs -r rg -n -C 4 'preStop|/drain|ingress|NetworkPolicy|dragontts'Repository: juspay/clairvoyance
Length of output: 5451
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== health.py =="
cat -n dragontts/app/api/v1/health.py
echo
echo "== app include_router section =="
sed -n '160,205p' dragontts/app/main.py | cat -n
echo
echo "== auth/dependencies and drain usages =="
rg -n -C 3 'Security|HTTPAuthorizationCredentials|BearerToken|oauth2|authenticate|dependencies|engage_drain|/drain|preStop|NetworkPolicy|NetworkPolic|Service|Ingress|port'|rg --glob '*.py|*.yaml|*.yml' .
echo
echo "== likely manifest files =="
fd -HI -t f -e yaml -e yml . | sortRepository: juspay/clairvoyance
Length of output: 3437
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== auth-related file candidates =="
fd -HI 'auth|security|middleware|dependencies|gateway|ingress|service|deployment|network' . -E .github -E node_modules -e py -e yaml -e yml -e json -e txt
echo
echo "== targeted auth/security searches =="
rg -n -C 3 'HTTPBearer|HTTPBasic|OAuth2|[A-Z][a-zA-Z0-9_]*Credentials|Bearer|Authorization|Security\(|dependencies=|include_router|middleware|ingress|Ingress|NetworkPolicy|Service|preStop|/drain|port:|containerPort|app\.include_router|fastapi' --glob '*.py' --glob '*.yaml' --glob '*.yml' --glob '*.json' --glob '*.txt' .
echo
echo "== deployment/manifest manifests if any =="
for f in $(fd -HI -t f -e yaml -e yml .); do
echo "-- $f --"
rg -n -C 4 'preStop|/drain|ingress|Ingress|NetworkPolicy|Service|containerPort|port:|dragontts|clairvoyance|breeze|production|prod' "$f" || true
doneRepository: juspay/clairvoyance
Length of output: 50376
Restrict access to GET /drain.
GET /drain is registered through health.router without route-level authentication, and it activates engage_drain(), which can force traffic away from DragonTTS. Restrict this endpoint to the pod-local preStop caller or add service authentication before release.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@dragontts/app/api/v1/health.py` around lines 32 - 37, Restrict the `drain`
endpoint in `health.router` to authenticated service callers or the pod-local
preStop caller before invoking `engage_drain()`. Add the route-level
authorization or request-origin validation within `drain`, while preserving its
existing drain behavior for authorized requests.
| # --- Graceful drain / clairvoyance kill switch --- | ||
| # On shutdown (k8s preStop -> GET /drain) dragontts FIRST tells clairvoyance | ||
| # to bypass it (so enable_tts_caching templates fall back to their upstream | ||
| # TTS provider), THEN drains its in-flight requests, THEN exits. On the next | ||
| # pod's startup it calls clairvoyance back with action=restore to resume | ||
| # caching. Clairvoyance's admin endpoint is HTTPBearer + require_admin, so a | ||
| # clairvoyance admin JWT must be supplied (env-injected in prod, never baked | ||
| # into the image). Empty CLAIRVOYANCE_URL => the notify calls are skipped | ||
| # (no-op), so the feature degrades cleanly when unconfigured. |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Correct the restore lifecycle documentation.
Line 247 states that startup sends action=restore. dragontts/app/drain.py only sends action="kill_switch". The restore action is manual.
Update this comment to prevent operators from expecting automatic recovery after a restart.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@dragontts/app/core/config.py` around lines 243 - 251, Update the
graceful-drain documentation comment in config.py to state that startup does not
automatically send action=restore and that restoring clairvoyance caching is a
manual operation. Keep the existing shutdown sequence and configuration behavior
documented accurately.
| async def engage_drain() -> bool: | ||
| """Run the /drain sequence, strictly ordered: | ||
|
|
||
| 1. enable the kill switch — mark draining (readiness 503) + tell clairvoyance | ||
| to bypass us so no new TTS calls arrive, | ||
| 2. send the Slack alert, | ||
| 3. (SIGTERM, in lifespan shutdown) drain in-flight requests, | ||
| 4. shutdown (pool/db cleanup). | ||
|
|
||
| The alert is AWAITED (bounded) so it's guaranteed sent before draining | ||
| begins, but a slow/down webhook can't stall preStop past the grace budget. | ||
| """ | ||
| if _is_draining: | ||
| logger.info("/drain already engaged — no-op") | ||
| return False | ||
| # 1) Kill switch FIRST. | ||
| mark_draining() | ||
| clairvoyance_ok = await notify_clairvoyance() | ||
| # 2) Alert (after the kill switch is enabled). | ||
| try: | ||
| await asyncio.wait_for( | ||
| send_kill_switch_slack( | ||
| clairvoyance_ok=clairvoyance_ok, inflight_n=inflight() | ||
| ), | ||
| timeout=2.0, | ||
| ) | ||
| except asyncio.TimeoutError: | ||
| logger.warning("kill-switch Slack alert timed out — proceeding with drain") | ||
| return clairvoyance_ok |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Honor enable_graceful_drain before drain side effects.
When ENABLE_GRACEFUL_DRAIN=false, wait_for_inflight_drain() returns early. However, engage_drain() still marks the worker draining, activates the Clairvoyance kill switch, and sends the Slack alert. This makes the feature-disable setting bypass DragonTTS traffic anyway.
Add the feature gate before mark_draining().
Proposed fix
async def engage_drain() -> bool:
+ if not settings.enable_graceful_drain:
+ logger.info("graceful drain disabled")
+ return False
if _is_draining:
logger.info("/drain already engaged — no-op")
return False📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| async def engage_drain() -> bool: | |
| """Run the /drain sequence, strictly ordered: | |
| 1. enable the kill switch — mark draining (readiness 503) + tell clairvoyance | |
| to bypass us so no new TTS calls arrive, | |
| 2. send the Slack alert, | |
| 3. (SIGTERM, in lifespan shutdown) drain in-flight requests, | |
| 4. shutdown (pool/db cleanup). | |
| The alert is AWAITED (bounded) so it's guaranteed sent before draining | |
| begins, but a slow/down webhook can't stall preStop past the grace budget. | |
| """ | |
| if _is_draining: | |
| logger.info("/drain already engaged — no-op") | |
| return False | |
| # 1) Kill switch FIRST. | |
| mark_draining() | |
| clairvoyance_ok = await notify_clairvoyance() | |
| # 2) Alert (after the kill switch is enabled). | |
| try: | |
| await asyncio.wait_for( | |
| send_kill_switch_slack( | |
| clairvoyance_ok=clairvoyance_ok, inflight_n=inflight() | |
| ), | |
| timeout=2.0, | |
| ) | |
| except asyncio.TimeoutError: | |
| logger.warning("kill-switch Slack alert timed out — proceeding with drain") | |
| return clairvoyance_ok | |
| async def engage_drain() -> bool: | |
| """Run the /drain sequence, strictly ordered: | |
| 1. enable the kill switch — mark draining (readiness 503) + tell clairvoyance | |
| to bypass us so no new TTS calls arrive, | |
| 2. send the Slack alert, | |
| 3. (SIGTERM, in lifespan shutdown) drain in-flight requests, | |
| 4. shutdown (pool/db cleanup). | |
| The alert is AWAITED (bounded) so it's guaranteed sent before draining | |
| begins, but a slow/down webhook can't stall preStop past the grace budget. | |
| """ | |
| if not settings.enable_graceful_drain: | |
| logger.info("graceful drain disabled") | |
| return False | |
| if _is_draining: | |
| logger.info("/drain already engaged — no-op") | |
| return False | |
| # 1) Kill switch FIRST. | |
| mark_draining() | |
| clairvoyance_ok = await notify_clairvoyance() | |
| # 2) Alert (after the kill switch is enabled). | |
| try: | |
| await asyncio.wait_for( | |
| send_kill_switch_slack( | |
| clairvoyance_ok=clairvoyance_ok, inflight_n=inflight() | |
| ), | |
| timeout=2.0, | |
| ) | |
| except asyncio.TimeoutError: | |
| logger.warning("kill-switch Slack alert timed out — proceeding with drain") | |
| return clairvoyance_ok |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@dragontts/app/drain.py` around lines 139 - 167, Update engage_drain() to
check enable_graceful_drain before any drain side effects, returning without
calling mark_draining(), notify_clairvoyance(), or send_kill_switch_slack() when
the feature is disabled; preserve the existing ordered drain flow when enabled.
| @app.middleware("http") | ||
| async def _track_inflight(request: Request, call_next): | ||
| # Per-process in-flight gauge so the SIGTERM shutdown drain can wait for | ||
| # live requests to finish before pools/db close. See app.drain. | ||
| incr_inflight() | ||
| try: | ||
| return await call_next(request) | ||
| finally: | ||
| decr_inflight() |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Locate current streaming tests and middleware implementations without executing code.
rg -n -C 4 '`@app`\.middleware\("http"\)|BaseHTTPMiddleware|StreamingResponse|/tts/stream' \
dragontts --glob '*.py'
fd -HI -t f . | rg '(^|/)(test|tests)/.*\.py$' | while IFS= read -r file; do
rg -n -C 3 'stream|StreamingResponse|TestClient|AsyncClient' "$file" || true
doneRepository: juspay/clairvoyance
Length of output: 35640
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== app/main.py relevant middleware/drain areas =="
sed -n '140,215p' dragontts/app/main.py
echo
echo "== app/drain.py =="
cat -n dragontts/app/drain.py
echo
echo "== app/api/v1/tts.py tts_stream implementation =="
sed -n '74,136p' dragontts/app/api/v1/tts.py
echo
echo "== package versions/framework =="
if [ -f dragontts/pyproject.toml ]; then
rg -n 'fastapi|starlette|uvicorn|httpx' dragontts/pyproject.toml -C 2 || true
fi
if [ -f dragontts/requirements.txt ]; then
rg -n 'fastapi|starlette|uvicorn|httpx' dragontts/requirements.txt -C 2 || true
fi
echo
echo "== inspect existing test files for app main drain/middleware =="
rg -n -C 3 'wait_for_inflight_drain|_track_inflight|drain|BaseHTTPMiddleware|StreamingResponse|tts/stream' dragontts --glob '*/test*.py' --glob '*test*.py' || trueRepository: juspay/clairvoyance
Length of output: 13725
🌐 Web query:
Starlette BaseHTTPMiddleware StreamingResponse call_next returns after body started documentation
💡 Result:
When using BaseHTTPMiddleware in Starlette, the call_next function is designed to handle the response from the underlying application and return a StreamingResponse object [1][2]. The issues regarding streaming responses and body generation often stem from the architecture of BaseHTTPMiddleware, which wraps the application's response to enable custom middleware logic [1][3]. Key technical points regarding this behavior include: 1. Response Wrapping: BaseHTTPMiddleware uses a memory object stream to capture the ASGI messages (http.response.start and http.response.body) sent by the underlying application [1][4]. Because it must wait to receive and potentially inspect these messages before returning the response to the middleware's dispatch method, it essentially converts all responses into a StreamingResponse [1][2]. 2. Body Consumption and Streaming: Because the response is internally buffered or streamed through an ASGI-compatible generator, attempts to consume the response body after call_next has been awaited—or trying to modify the body in ways that conflict with the stream already in progress—can lead to issues [5][3]. The middleware waits for the application's response to complete to ensure it can correctly pass on headers and content [1][6]. 3. Known Limitations and Risks: The use of BaseHTTPMiddleware has been widely criticized and discussed within the Starlette community due to several inherent limitations [3]: - Streaming responses are effectively forced to be loaded or managed through the middleware's internal stream, which can impact performance or memory usage [6][3]. - Background tasks associated with the response may behave differently, as the middleware must orchestrate their execution relative to the response completion [6][3]. - Exceptions occurring during the generation of the streaming body were historically prone to being silently ignored, though various PRs have aimed to improve exception propagation [7][8]. Due to these complexities, it is often recommended to use lighter-weight middleware patterns (such as those simply implementing call) when possible, rather than subclassing BaseHTTPMiddleware, especially for endpoints that rely heavily on streaming or complex background task logic [3][9].
Citations:
- 1: https://github.com/encode/starlette/blob/b8ea367b4304a98653ec8ce9c794ad0ba6dcaf4b/starlette/middleware/base.py
- 2: https://stackoverflow.com/questions/74174298/fastapi-starlette-middleware-returns-a-streamingresponse-instead-of-jsonresponse
- 3: Document
BaseHTTPMiddlewarebugs Kludex/starlette#1640 - 4: https://github.com/encode/starlette/blob/040d8c86b09f34be49e8c253d97a588973bc7308/starlette/middleware/base.py
- 5: Reuse Request's body buffer for call_next in BaseHTTPMiddleware Kludex/starlette#1692
- 6: Streaming with BaseHTTPMiddleware: force background to run after response completes Kludex/starlette#1017
- 7: Prevent BaseHTTPMiddleware from hiding errors of StreamingResponse Kludex/starlette#1459
- 8: Prevent BaseHTTPMiddleware from hiding errors of StreamingResponse Kludex/starlette#1459
- 9: https://starlette.dev/middleware/
🌐 Web query:
Starlette BaseHTTPMiddleware dispatch yield receives messages call_next StreamingResponse await call_next
💡 Result:
In Starlette's BaseHTTPMiddleware, the dispatch method uses an internal call_next function to execute the application stack [1][2]. When you await call_next(request) inside your dispatch method, it returns a StreamingResponse object, regardless of what the underlying endpoint originally returned (e.g., JSONResponse, HTMLResponse) [3]. Key Technical Details: 1. call_next Mechanism: The call_next function internally executes the application by setting up a task group and a memory object stream [1][4]. It captures the ASGI messages sent by the application and reconstructs them into a StreamingResponse [1]. 2. Response Body Access: Because call_next returns a StreamingResponse, the response body is provided as an asynchronous generator (response.body_iterator) [5][6]. If you need to read or modify the body within your middleware, you must consume this iterator [5][6]. A common approach involves converting the generator into a list, processing the bytes, and then re-assigning a new iterator back to the response (often using iterate_in_threadpool from starlette.concurrency to maintain compatibility) [5][6]. 3. Streaming and Memory Usage: Using BaseHTTPMiddleware forces the response to be handled as a stream [1][3]. Note that accessing the full body by iterating through response.body_iterator loads the entire response into memory, which can be dangerous or inefficient for large streaming responses or file responses [7][8][6]. 4. Warning on "Yield": While some newer ASGI middleware patterns (often proposed in experimental pull requests) might use yield to separate pre- and post-processing logic [9], the standard BaseHTTPMiddleware utilizes an awaitable dispatch method that requires you to return a response object explicitly [2]. Due to the internal implementation of BaseHTTPMiddleware, it is generally discouraged for performance-sensitive applications or those heavily relying on StreamingResponse/FileResponse, as it adds overhead and can complicate the ASGI lifecycle [8]. For advanced use cases, implementing raw ASGI middleware (by defining a call method that handles scope, receive, and send directly) is recommended as a more efficient and flexible alternative [8].
Citations:
- 1: https://github.com/encode/starlette/blob/b8ea367b4304a98653ec8ce9c794ad0ba6dcaf4b/starlette/middleware/base.py
- 2: https://starlette.dev/middleware/
- 3: https://stackoverflow.com/questions/74174298/fastapi-starlette-middleware-returns-a-streamingresponse-instead-of-jsonresponse
- 4: https://github.com/encode/starlette/blob/040d8c86b09f34be49e8c253d97a588973bc7308/starlette/middleware/base.py
- 5: Get response body in Middleware Kludex/starlette#874
- 6: https://stackoverflow.com/questions/71882419/fastapi-how-to-get-the-response-body-in-middleware
- 7: Improve documentation on
BaseHTTPMiddlewareKludex/starlette#1655 - 8: Background tasks don't work with middleware that subclasses
BaseHTTPMiddlewareKludex/starlette#919 - 9: Experiment a high-level HTTPMiddleware Kludex/starlette#1691
🌐 Web query:
Starlette 0.42.0 BaseHTTPMiddleware base.py call_next dispatch StreamingResponse yield response_body message body_iterator
💡 Result:
In Starlette (including version 0.42.0), the BaseHTTPMiddleware class operates by intercepting the ASGI lifecycle to provide a high-level request/response interface [1][2][3]. When you call call_next(request) within your dispatch method, the middleware internally manages the ASGI message stream [1]. It uses a memory object stream to capture the response from the underlying application [1]. Specifically, it defines an internal body_stream asynchronous generator that reads from this stream, yielding chunks of the response body as they arrive [1]. Key technical implications regarding the response body and iterator: 1. Memory Evaluation: BaseHTTPMiddleware consumes the response body by iterating over the stream [1]. This effectively means that for any request passing through this middleware, the entire response body is evaluated and buffered in memory [4][5]. Because of this, it is generally discouraged for use with large streaming responses (e.g., FileResponse or long-lived StreamingResponse), as it negates the benefits of streaming by loading data into memory [4][5]. 2. Modifying the Body: If you need to read or modify the response body, you must be aware that the response returned by call_next is typically a StreamingResponse [6]. Because the original body_iterator is consumed during the middleware's internal process, simply attempting to iterate over response.body_iterator after call_next returns will yield nothing [6]. To access or modify the body, you must extract the chunks, perform your operations, and then re-assign a new iterator (often using starlette.concurrency.iterate_in_threadpool) to response.body_iterator before returning the response [6]. 3. Known Issues and Recommendations: The community and Starlette maintainers have long identified that BaseHTTPMiddleware's abstraction model creates limitations—specifically regarding background tasks, streaming, and context variables [5][7]. Consequently, it is often recommended to use native ASGI middleware (by implementing a call method) if your use case requires complex handling of response streams or performance-sensitive streaming [5][7]. Top results: [1], [6], [4], [5]
Citations:
- 1: https://github.com/encode/starlette/blob/b8ea367b4304a98653ec8ce9c794ad0ba6dcaf4b/starlette/middleware/base.py
- 2: https://starlette.dev/middleware/
- 3: https://github.com/encode/starlette/blob/a766a58d/docs/release-notes.md
- 4: Improve documentation on
BaseHTTPMiddlewareKludex/starlette#1655 - 5: Document
BaseHTTPMiddlewarebugs Kludex/starlette#1640 - 6: Get response body in Middleware Kludex/starlette#874
- 7: Deprecating BaseHTTPMiddleware Kludex/starlette#1678
🌐 Web query:
Starlette 0.42.0 BaseHTTPMiddleware base.py call_next dispatch yie
💡 Result:
In Starlette 0.42.0, BaseHTTPMiddleware provides an abstract class to implement ASGI middleware using a request/response interface rather than handling raw ASGI messages [1][2]. The dispatch method is the core of this class, where you must implement your logic [2]. In your request, you mentioned call_next and yield in the context of BaseHTTPMiddleware. It is important to distinguish between how BaseHTTPMiddleware works and how alternative (often proposed) high-level middleware abstractions work: 1. BaseHTTPMiddleware (Current): You call await call_next(request) to trigger the rest of the application stack and receive a response object [2]. It does not use yield within the dispatch method to control the flow [3]. 2. Alternative/Proposed High-Level Middleware: Proposals and discussions (such as in Starlette repository issues) have suggested "HTTPMiddleware" patterns that use yield to allow middleware to handle request/response cycles more cleanly, such as response = yield, but this is not the current implementation of BaseHTTPMiddleware [4]. Known Limitations of BaseHTTPMiddleware: - Context Variables: BaseHTTPMiddleware can disrupt the propagation of context variables to subsequent middleware or the application, as the downstream application is often run in a separate task [5][6]. - Request Body Consumption: If your middleware reads the request body, it may not be available for the downstream application via call_next, potentially causing issues with how the request is processed [5]. - Streaming/Background Tasks: Because BaseHTTPMiddleware wraps the request/response cycle, it can sometimes force streaming responses to be fully buffered in memory or force background tasks to be evaluated in the foreground before the response is returned, which may not be the intended behavior for all applications [7][8]. Given these limitations, Starlette maintainers and the community often recommend using pure ASGI middleware (implementing the call(self, scope, receive, send) method) when you need precise control over streaming, background tasks, or context propagation, as it avoids the abstractions that create these issues [2][7][8].
Citations:
- 1: https://github.com/encode/starlette/blob/a766a58d/docs/release-notes.md
- 2: https://starlette.dev/middleware/
- 3: https://github.com/encode/starlette/blob/b8ea367b4304a98653ec8ce9c794ad0ba6dcaf4b/starlette/middleware/base.py
- 4: Experiment a high-level HTTPMiddleware Kludex/starlette#1691
- 5: Remaining bugs/limitations of `BaseHTTPMiddleware` - and what to do about them Kludex/starlette#1729
- 6: docs: clarify BaseHTTPMiddleware impact on contextvars for subsequent middleware Kludex/starlette#2943
- 7: Document
BaseHTTPMiddlewarebugs Kludex/starlette#1640 - 8: Background tasks don't work with middleware that subclasses
BaseHTTPMiddlewareKludex/starlette#919
🌐 Web query:
github encode starlette v0.42.0 middleware/base.py call_next StreamingResponse response_body
💡 Result:
In Starlette v0.42.0, the BaseHTTPMiddleware class uses call_next to execute the application and return a StreamingResponse [1][2]. Because BaseHTTPMiddleware operates by intercepting the ASGI lifecycle, the call_next function must return a StreamingResponse regardless of the response type originally returned by the underlying endpoint [3][4]. This architecture creates a specific behavior when accessing the response body in middleware: 1. Streaming Nature: The response object returned by call_next is a StreamingResponse, and its body is contained within an asynchronous iterator response.body_iterator [3]. 2. Consuming the Body: Because response.body_iterator is an asynchronous generator, reading it directly consumes the stream [3]. If you consume the stream in your middleware, it will be empty by the time the response reaches the client [3]. 3. Restoring the Body: To both inspect the body and ensure the client receives it, you must collect the chunks from the response.body_iterator, and then re-assign a new asynchronous iterator (often created using starlette.concurrency.iterate_in_threadpool) back to response.body_iterator [3]. Common implementation pattern: from starlette.concurrency import iterate_in_threadpool from starlette.middleware.base import BaseHTTPMiddleware class MyMiddleware(BaseHTTPMiddleware): async def dispatch(self, request, call_next): response = await call_next(request) # 1. Consume and collect the body response_body = [chunk async for chunk in response.body_iterator] # 2. Do something with the body (e.g., logging) # print(response_body.decode) # 3. Restore the body iterator so the response can be sent to the client response.body_iterator = iterate_in_threadpool(iter(response_body)) return response Because of the complexity and performance overhead associated with this pattern, the Starlette documentation often recommends using pure ASGI middleware instead of BaseHTTPMiddleware when granular control over the response body is required [5][6].
Citations:
- 1: https://github.com/encode/starlette/blob/b8ea367b4304a98653ec8ce9c794ad0ba6dcaf4b/starlette/middleware/base.py
- 2: https://github.com/encode/starlette/blob/a766a58d/docs/release-notes.md
- 3: Get response body in Middleware Kludex/starlette#874
- 4: https://stackoverflow.com/questions/74174298/fastapi-starlette-middleware-returns-a-streamingresponse-instead-of-jsonresponse
- 5: https://starlette.dev/middleware/
- 6: "Task was destroyed but it is pending!" error when middleware dispatch func discards response from call_next Kludex/starlette#1022
🌐 Web query:
Starlette middleware base ResponseWrapper StreamingResponse start_body_if_started send response
💡 Result:
In the context of Starlette middleware, "start_body_if_started" is not a standard method in the public API or core implementations like BaseHTTPMiddleware or StreamingResponse. It appears you may be referring to the internal mechanisms used to handle ASGI response lifecycles when wrapping middleware. When writing middleware in Starlette, the recommended pattern for inspecting or modifying a response is to wrap the send ASGI callable [1][2]. The core challenge, which internal components address, is that ASGI requires an http.response.start message to be sent before any http.response.body messages [1]. Key technical points regarding response wrapping: 1. Wrapping the send callable: Middleware often implements a custom send function that intercepts http.response.start messages to inject headers or modify status codes before passing the message to the actual send callable [1]. 2. StreamingResponse handling: StreamingResponse objects manage the conversion of a body iterator into a series of http.response.body ASGI messages [3][4]. If you are implementing middleware that must interact with these streams, you generally need to consume the ASGI stream from the underlying app, process it, and then yield or send it forward [5][6]. 3. BaseHTTPMiddleware: This utility class simplifies middleware creation by allowing you to work with Request and Response objects directly [5]. Internally, it manages the transformation between the raw ASGI interface and these high-level objects [5]. It uses a MemoryObjectStream to capture the response from the inner application, which allows it to reconstruct a Response object (often a StreamingResponse) that the user-provided dispatch function can then return [5]. 4. Best practices: To avoid common pitfalls such as consuming request bodies prematurely or breaking the ASGI lifecycle, developers are encouraged to keep middleware stateless [2]. When state is required, patterns like using a responder helper class—which buffers the initial response start until the final response structure is known—are preferred [2]. If you encountered "start_body_if_started" in a specific codebase or older version of Starlette, it is likely a custom helper function intended to check whether the response headers have already been sent before attempting to send the body, a common requirement when managing response state in ASGI middleware [1].
Citations:
- 1: https://starlette.dev/middleware/
- 2: Document how to create ASGI middlewares Kludex/starlette#1656
- 3: https://github.com/encode/starlette/blob/e54be85aba48201b0f1c061c892bdc380efb754d/starlette/responses.py
- 4: https://github.com/Kludex/starlette/blob/main/starlette/responses.py
- 5: https://github.com/encode/starlette/blob/040d8c86b09f34be49e8c253d97a588973bc7308/starlette/middleware/base.py
- 6: https://github.com/Kludex/starlette/blob/main/starlette/middleware/base.py
🌐 Web query:
github encode starlette BaseHTTPMiddleware dispatch asyncio.create_task call_next response_body message generator
💡 Result:
The use of BaseHTTPMiddleware in Starlette often leads to complications when attempting to access the response body or manage background tasks due to how the dispatch method and call_next interact with ASGI streams [1][2][3]. Accessing the Response Body Because BaseHTTPMiddleware processes responses as a StreamingResponse (where body_iterator is a generator), you cannot simply read the body without consuming the generator [1][4]. If you iterate over response.body_iterator to read the data, that generator is exhausted, causing the response sent to the client to be empty or incomplete [1][5]. To safely access and preserve the body, you must store the chunks in a list, then replace the body_iterator with a new iterator created from that list [1][4]. A common approach uses starlette.concurrency.iterate_in_threadpool [1][4]: python from starlette.concurrency import iterate_in_threadpool # Inside your dispatch method response = await call_next(request) response_body = [chunk async for chunk in response.body_iterator] response.body_iterator = iterate_in_threadpool(iter(response_body)) # You can now use response_body Asyncio Tasks and Background Processing The call_next function inside BaseHTTPMiddleware uses an anyio task group to manage the lifecycle of the downstream application [6][7]. Issues frequently arise (such as h11 protocol errors or deadlocks) because the middleware expects the StreamingResponse to finish before completing the task group [3][8][5]. If you attempt to use asyncio.create_task for background work within dispatch, it may be prematurely cancelled or cause timing issues because the middleware's internal call_next implementation relies on specific anyio task group lifecycles to coordinate the response stream [2][6][3]. Recommendations 1. Prefer Pure ASGI Middleware: If your middleware requires low-level control over request/response bodies or reliable background task management, the official documentation recommends writing pure ASGI middleware rather than using BaseHTTPMiddleware [9][5]. Pure ASGI middleware allows you to wrap the send callable directly, giving you complete control over the ASGI messages (like http.response.body) without the limitations of the StreamingResponse abstraction [5]. 2. Avoid Modifying State: Do not mutate the middleware instance or its state outside of init, as per Starlette's design principles [10]. 3. Background Tasks: If using FastAPI or Starlette's BackgroundTasks, ensure they are properly registered on the Response object itself rather than trying to spawn them manually inside BaseHTTPMiddleware if you encounter unexpected connection termination [9].
Citations:
- 1: Get response body in Middleware Kludex/starlette#874
- 2: Background tasks don't work with middleware that subclasses
BaseHTTPMiddlewareKludex/starlette#919 - 3: Streaming with BaseHTTPMiddleware: force background to run after response completes Kludex/starlette#1017
- 4: Get response body in Middleware Kludex/starlette#874
- 5: Accessing Response Body inside of middleware throwing error `Too little data for declared Content-Length` fastapi/fastapi#11064
- 6: https://github.com/encode/starlette/blob/b8ea367b4304a98653ec8ce9c794ad0ba6dcaf4b/starlette/middleware/base.py
- 7: https://github.com/encode/starlette/blob/master/starlette/middleware/base.py
- 8: Replace task cancellation in
BaseHTTPMiddlewarewithhttp.disconnect+recv_stream.closeKludex/starlette#1715 - 9:
HTTPMiddlewarebreaksBackgroundTasksin the same connection Kludex/starlette#2093 - 10: https://starlette.dev/middleware/
🌐 Web query:
Starlette BaseHTTPMiddleware dispatch response_body message await app_stream response_body body_stream call_next
💡 Result:
In Starlette, the BaseHTTPMiddleware class allows developers to intercept and modify request/response lifecycles using an ASGI-compatible approach [1][2]. The core mechanism involves the dispatch method and the call_next function [3][1]. When you implement BaseHTTPMiddleware, you define a dispatch method that receives the request and the call_next endpoint [3][1]. Calling await call_next(request) triggers the rest of the application stack and returns a StreamingResponse [3][4]. Internally, BaseHTTPMiddleware manages the response lifecycle through the following process: 1. call_next: When you call this function, it creates a memory object stream to capture the ASGI messages (like http.response.start and http.response.body) sent by the underlying application [3]. 2. Response Streaming: The application's response is converted into a StreamingResponse [3]. The body is managed by an internal body_stream generator, which iterates over the recv_stream to collect and yield chunks of the response body [3]. 3. Consuming the Body: Because the response returned by call_next is a StreamingResponse, the body is not immediately available in memory [4]. If you need to access or modify the response body within your middleware, you must consume the stream [4]. A common pattern to read the body involves iterating over the body_iterator and then replacing it so the response can be sent correctly [4]: async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response: response = await call_next(request) # Consume the response body response_body = [section async for section in response.body_iterator] # Reconstruct the body iterator for the response to be sent response.body_iterator = iterate_in_threadpool(iter(response_body)) # Process the body (e.g., decode and log) full_body = b"".join(response_body) return response Note: BaseHTTPMiddleware has known limitations regarding its implementation, specifically concerning how it interacts with the underlying ASGI scope and context variables, which can sometimes lead to unexpected behavior compared to pure ASGI middleware [2][5]. [3][4]
Citations:
- 1: https://starlette.dev/middleware/
- 2: https://www.starlette.io/middleware/
- 3: https://github.com/encode/starlette/blob/b8ea367b4304a98653ec8ce9c794ad0ba6dcaf4b/starlette/middleware/base.py
- 4: Get response body in Middleware Kludex/starlette#874
- 5: https://github.com/encode/starlette/blob/master/docs/middleware.md
🏁 Script executed:
#!/bin/bash
set -euo pipefail
tmp="$(mktemp -d)"
python3 - <<'PY'
import urllib.request
url = "https://raw.githubusercontent.com/encode/starlette/v0.46.2/middleware/base.py"
try:
urllib.request.urlretrieve(url, f"{tmp}/base.py")
except Exception as e:
print("download failed", e)
raise SystemExit(0)
with open(f"{tmp}/base.py", encoding="utf-8", errors="replace") as f:
s=f.read()
for needle in [
"class BaseHTTPMiddleware",
"async def __call__",
"dispatch",
"async def call_next",
"response_body",
"Response",
"StreamingResponse",
"await app_stream",
]:
i=s.find(needle)
print(f"\n--- {snippet := needle[:120]} at {max(0,i-30)}")
if i>=0:
print(s[max(0,i-300):i+1800])
PY
python3 - <<'PY'
# A read-only semantics probe based on the Starlette middleware pattern found above.
# It models that BaseHTTPMiddleware awaits the downstream response iterator before
# returning from the user dispatch, so a try/finally after `await call_next`
# can run before client body sending completes.
class ModelStreamResponse:
def __init__(self):
self.sent = []
async def body_iterator(self):
try:
for i in range(5):
yield f"{i}\n".encode()
finally:
print("body iterator cleaned up after all chunks sent")
def invoke_via_middleware_with_finally(self):
async def body_iterator():
try:
async for chunk in self.body_iterator():
self.sent.append(chunk)
finally:
print("middleware after call_next finally -> _inflight decremented")
return body_iterator()
async def run():
r = ModelStreamResponse()
gen = r.invoke_via_middleware_with_finally()
await gen.__anext__() # fastapi/start response headers/first chunk ready
print(f"after first middleware frame: sent={len(r.sent)}")
first = await gen.__anext__()
print(f"first chunk consumed before rest: {first}")
while True:
try:
first = await gen.__anext__()
except StopAsyncIteration:
break
await gen.aclose()
print(f"final: sent={len(r.sent)}")
import asyncio
asyncio.run(run())
PYRepository: juspay/clairvoyance
Length of output: 722
🏁 Script executed:
#!/bin/bash
set -euo pipefail
tmp="$(mktemp -d)"
python3 - <<'PY'
import urllib.request
url = "https://raw.githubusercontent.com/encode/starlette/v0.46.2/middleware/base.py"
try:
urllib.request.urlretrieve(url, f"{tmp}/base.py")
except Exception as e:
print("download failed", e)
raise SystemExit(0)
with open(f"{tmp}/base.py", encoding="utf-8", errors="replace") as f:
s=f.read()
for needle in [
"class BaseHTTPMiddleware",
"async def __call__",
"def dispatch",
"async def call_next",
"response_body",
"Response",
"StreamingResponse",
"await app_stream",
]:
i=s.find(needle)
print(f"\n--- {needle[:120]} at {max(0, i-30)}")
if i >= 0:
print(s[max(0, i-300): i+2200])
PY
python3 - <<'PY'
# A read-only semantics probe that models Starlette BaseHTTPMiddleware's effect:
# the response iterator is consumed with await, not passed through unchanged.
# Therefore an async try/finally immediately after `await call_next()` runs
# before the rest of the response body reaches the client.
sent = []
async def body_iterator():
try:
for i in range(5):
sent.append(f"{i}\n".encode())
# simulate network yield while final chunks remain unsent
await asyncio.sleep(0)
finally:
print("body iterator cleanup only runs after all body chunks are sent")
async def run():
# consume the "response" iterator exactly as BaseHTTPMiddleware does,
# so the generator stays open until the last chunk is sent.
final = None
try:
async for chunk in body_iterator():
if final is None:
print(f"first chunk delivered: {chunk!r}")
finally:
print("decorator try/finally runs: _inflight decremented while body may still be sending")
print(f"sent after first chunk: {sent}")
await asyncio.sleep(0)
print(f"sent at event loop quiescence: {sent}")
import asyncio
asyncio.run(run())
PYRepository: juspay/clairvoyance
Length of output: 834
Track streaming requests until the final response body is sent.
await call_next(request) buffers the streamed body, so the middleware finally runs while /tts/stream still sends StreamingResponse chunks. The _inflight counter can reach 0 before shutdown must wait for the live response, allowing cache.stop() or registry cleanup to run early.
Use ASGI middleware that decrements the gauge only after the final http.response.body message with more_body=False, and keep a fallback for aborted responses. Add a regression test that an open stream remains counted after headers are sent.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@dragontts/app/main.py` around lines 192 - 200, Replace the _track_inflight
middleware’s call_next/finally tracking with ASGI send-wrapper logic that keeps
_inflight incremented until the final http.response.body message has
more_body=False. Ensure aborted or exceptional responses still decrement exactly
once, and add a regression test confirming an open streaming response remains
counted after headers are sent.
| BUNDLE = "app.ai.voice.agents.breeze_buddy.llm" | ||
|
|
||
|
|
||
| def _valid_config(**overrides) -> LLMConfiguration: |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Add type annotations to all test function signatures.
The repository requires type hints on all function signatures. Annotate _valid_config, each monkeypatch parameter, parametrized values, and each return type.
Also applies to: 32-32, 44-44, 52-52, 69-71, 79-79, 90-90
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/test_llm_gemini_resolution.py` at line 20, Annotate every function
signature in tests/test_llm_gemini_resolution.py, including _valid_config and
all test functions, with explicit parameter and return types. Type each
monkeypatch fixture and parametrized argument/value, using appropriate existing
typing symbols while preserving the current test behavior.
Source: Coding guidelines
|
|
||
| def _valid_config(**overrides) -> LLMConfiguration: | ||
| base = dict( | ||
| provider=LLMProvider.GEMINI, |
There was a problem hiding this comment.
🎯 Functional Correctness | 🔴 Critical | ⚡ Quick win
Add the Gemini provider contract before using it in this test.
LLMProvider has no GEMINI member. Pyrefly fails at this line. The supplied resolver also has no GEMINI dispatch branch. Add the enum member and its _resolve_gemini dispatch path, or change this test to the supported provider contract.
🧰 Tools
🪛 GitHub Actions: PR Build Check / 0_build.txt
[error] 22-22: Pyrefly check failed: Class LLMProvider has no class attribute GEMINI. (missing-attribute)
🪛 GitHub Actions: PR Build Check / build
[error] 22-22: Pyrefly error [missing-attribute]: Class LLMProvider has no class attribute GEMINI.
🪛 GitHub Check: build
[failure] 22-22: Pyrefly missing-attribute
Class LLMProvider has no class attribute GEMINI
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/test_llm_gemini_resolution.py` at line 22, Update the Gemini test setup
around LLMProvider.GEMINI by either adding the GEMINI enum member and wiring the
resolver’s dispatch to _resolve_gemini, or replacing the test with a currently
supported provider contract; ensure the test only references a provider that
exists and is handled by the resolver.
Sources: Linters/SAST tools, Pipeline failures
8d1f0cc to
97629e7
Compare
There was a problem hiding this comment.
Pull request overview
This PR aims to reduce/avoid downtime during DragonTTS pod restarts by introducing a drain flow that (1) signals Clairvoyance to bypass DragonTTS before termination and (2) lets in-flight requests complete before caches/pools are torn down.
Changes:
- Added graceful-drain coordination for DragonTTS:
/drainpreStop hook,/readyreadiness probe behavior, and per-request in-flight tracking to delay shutdown cleanup. - Added new DragonTTS config/env knobs for Clairvoyance kill-switch notification and drain timing.
- Updated TTS parameter forwarding/caching behavior (e.g., SSML parsing flag in DragonTTS param collection; ElevenLabs voice_settings now folds in speed).
Reviewed changes
Copilot reviewed 8 out of 8 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| tests/test_llm_gemini_resolution.py | Adds tests for a “Gemini Developer API” text-LLM resolver. |
| dragontts/app/providers/elevenlabs.py | Extends default ElevenLabs voice_settings to include speed when provided. |
| dragontts/app/main.py | Adds HTTP middleware to track in-flight requests and waits for drain during lifespan shutdown. |
| dragontts/app/drain.py | Introduces drain orchestration (kill switch notify, Slack alert, in-flight drain wait). |
| dragontts/app/core/config.py | Adds settings for Clairvoyance kill-switch + graceful drain behavior. |
| dragontts/app/api/v1/health.py | Adds /ready readiness probe and /drain preStop hook endpoint. |
| dragontts/.env.example | Documents new graceful-drain / Clairvoyance kill-switch environment variables. |
| app/ai/voice/tts/dragontts.py | Adds enable_ssml_parsing to DragonTTS param collection for cache-key + provider forwarding. |
|
|
||
|
|
||
| def is_draining() -> bool: | ||
| return _is_draining | ||
|
|
||
|
|
||
| def mark_draining() -> None: | ||
| global _is_draining | ||
| _is_draining = True | ||
|
|
||
|
|
| # On shutdown (k8s preStop -> GET /drain) dragontts FIRST tells clairvoyance | ||
| # to bypass it (so enable_tts_caching templates fall back to their upstream | ||
| # TTS provider), THEN drains its in-flight requests, THEN exits. On the next | ||
| # pod's startup it calls clairvoyance back with action=restore to resume | ||
| # caching. Clairvoyance's admin endpoint is HTTPBearer + require_admin, so a |
| @router.get("/drain") | ||
| async def drain(): | ||
| """k8s preStop hook. Marks the pod draining, fires the kill-switch Slack | ||
| alert, and tells clairvoyance to bypass us BEFORE SIGTERM arrives — then | ||
| the lifespan shutdown waits for in-flight requests to finish. See app.drain.""" | ||
| await engage_drain() |
| def _valid_config(**overrides) -> LLMConfiguration: | ||
| base = dict( | ||
| provider=LLMProvider.GEMINI, | ||
| model="gemini-2.5-flash-lite", |
|
Thanks @coderabbitai + Copilot — all comments addressed below. Fixes are mirrored across both the standalone ✅ Fixed
⏭️ Won't fix (justified)
⏭️ Separate work (not from this PR's drain/SSML changes)
🛠️ Operational (manifest, not code)The graceful-drain feature needs the Deployment to wire it — |
|
Tip For best results, initiate chat on the files or code changes.
The streaming middleware fix addresses the response-body lifetime issue. The I accept the stated rationale for the unauthenticated The Gemini resolver test remains a separate blocking concern for this PR if the test file stays committed. Either include the required resolver implementation or remove the unrelated test from this PR. The Deployment wiring is required before graceful drain operates in production. Do you want me to open follow-up GitHub issues for the Gemini work and the Deployment configuration? 🐇 ✅ ✏️ Learnings added
If you found this review helpful, would you consider giving us a shout-out on X? Thank you for using CodeRabbit! You are interacting with an AI system. |
031830a to
ab2776d
Compare
| 3. **During the restart window**, clairvoyance keeps serving TTS from upstream — | ||
| callers still get audio (just uncached: marginally higher latency/cost, **no | ||
| silence, no errors**). |
There was a problem hiding this comment.
Calls already in progress when the pod drains don't get this fallback — they go silent for the whole restart window. get_tts_service reads is_dragontts_healthy() exactly once, when the pipeline is built at call start, and DragonTTSService.run_tts has no per-utterance fallback: an unreachable pod just yields an ErrorFrame, and nothing in breeze_buddy handles it. The kill switch only protects calls that start after the flag flips.
|
Adds a graceful-drain path to the DragonTTS caching proxy: a 1 new issue found. |
ab2776d to
36d9121
Compare
…gracefull shutdown
36d9121 to
0f6b7b4
Compare
thats why we have 2 mins time for the running requests to end |
Summary by CodeRabbit
New Features
Bug Fixes