From 9bb8f119b52f0cc2abd08d1cfadd62f2ff7b8f37 Mon Sep 17 00:00:00 2001 From: Janis Date: Wed, 24 Jun 2026 09:22:31 +0200 Subject: [PATCH] fix(spark-analytics): stream NIM AI responses to kill webview 504s The /explain endpoint blocked on a synchronous NIM call (llama-3.3-70b, 30-60s+) while nginx's default 60s proxy-read-timeout fired first, surfacing every AI request as a 504 Gateway Timeout. Switch the NIM proxy to streaming (stream:true) and relay its SSE to the browser via a new _emit() helper; the client now reads resp.body incrementally and renders tokens as they arrive. With chunks flowing, nginx's read-timeout resets per chunk so the 60s wall is gone, and the frozen spinner is replaced by live text. Ingress: disable proxy-buffering (else nginx buffers the whole response, defeating streaming) and set proxy-read-timeout to 120s as the inter-chunk gap budget. Server urlopen timeout 60->110s. Co-Authored-By: Claude Opus 4.8 --- spark-analytics/base/webview.yaml | 119 +++++++++++++----- .../eu-central-1-staging/webview-ingress.yaml | 4 + 2 files changed, 91 insertions(+), 32 deletions(-) diff --git a/spark-analytics/base/webview.yaml b/spark-analytics/base/webview.yaml index 20db08d..51cc94a 100755 --- a/spark-analytics/base/webview.yaml +++ b/spark-analytics/base/webview.yaml @@ -187,37 +187,52 @@ data: if (btn) btn.disabled = !hasKey(); } - async function nimFetch(payload) { + async function nimStream(payload, onDelta) { const ctrl = new AbortController(); - const tid = setTimeout(() => ctrl.abort(), 65000); - let resp; + const tid = setTimeout(() => ctrl.abort(), 180000); try { - resp = await fetch('/explain', { + const resp = await fetch('/explain', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(payload), signal: ctrl.signal, }); + if (resp.status === 429) { + const retryAfter = parseInt(resp.headers.get('Retry-After') || '10', 10); + const err = new Error('Rate limited — retry in ' + retryAfter + 's'); + err.retryAfter = retryAfter; + throw err; + } + if (!resp.ok) { + const msg = await resp.text(); + throw new Error('Proxy ' + resp.status + ': ' + msg); + } + const reader = resp.body.getReader(); + const dec = new TextDecoder(); + const SEP = String.fromCharCode(10, 10); // "\n\n" — avoids YAML backslash escaping + let buf = '', full = ''; + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + buf += dec.decode(value, { stream: true }); + let idx; + while ((idx = buf.indexOf(SEP)) >= 0) { + const frame = buf.slice(0, idx).trim(); + buf = buf.slice(idx + 2); + if (!frame.startsWith('data:')) continue; + const obj = JSON.parse(frame.slice(5).trim()); + if (obj.error) throw new Error(obj.error); + if (obj.delta) { full += obj.delta; onDelta(full); } + } + } + return full || '(no response)'; } finally { clearTimeout(tid); } - if (resp.status === 429) { - const retryAfter = parseInt(resp.headers.get('Retry-After') || '10', 10); - const err = new Error('Rate limited — retry in ' + retryAfter + 's'); - err.retryAfter = retryAfter; - throw err; - } - if (!resp.ok) { - const msg = await resp.text(); - throw new Error('Proxy ' + resp.status + ': ' + msg); - } - const data = await resp.json(); - if (data.error) throw new Error(data.error); - return data.text || '(no response)'; } - function queueNim(payload) { - const next = nimQueue.then(() => nimFetch(payload)); + function queueNim(payload, onDelta) { + const next = nimQueue.then(() => nimStream(payload, onDelta)); nimQueue = next.catch(() => {}); return next; } @@ -237,7 +252,10 @@ data: textEl.textContent = ''; try { - const text = await queueNim({ key: key, mode: 'dataset', dataset: window.DATASET }); + const text = await queueNim( + { key: key, mode: 'dataset', dataset: window.DATASET }, + partial => { textEl.textContent = partial; }, + ); textEl.textContent = text; } catch (e) { textEl.className = 'explain-error'; @@ -275,10 +293,13 @@ data: textEl.textContent = 'Analyzing…'; try { - const text = await queueNim({ - key: key, mode: 'row', - dataset: { label: dataset.label, headers: dataset.headers, row: row }, - }); + const text = await queueNim( + { + key: key, mode: 'row', + dataset: { label: dataset.label, headers: dataset.headers, row: row }, + }, + partial => { textEl.textContent = partial; }, + ); rowCache[idx] = text; textEl.textContent = text; btn.innerHTML = '✓'; @@ -480,6 +501,7 @@ data: self.end_headers() return length = int(self.headers.get("Content-Length", 0)) + headers_sent = False try: body = json.loads(self.rfile.read(length)) api_key = body.get("key", "") @@ -538,7 +560,7 @@ data: ], "max_tokens": max_tokens, "temperature": 0.4, - "stream": False, + "stream": True, }).encode("utf-8") req = urllib.request.Request( "https://integrate.api.nvidia.com/v1/chat/completions", @@ -550,10 +572,7 @@ data: method="POST", ) try: - with urllib.request.urlopen(req, timeout=60) as r: - nim_data = json.loads(r.read()) - text = nim_data.get("choices", [{}])[0].get("message", {}).get("content", "(no response)") - self._send_json({"text": text}) + upstream = urllib.request.urlopen(req, timeout=110) except urllib.error.HTTPError as he: if he.code == 429: retry_after = he.headers.get("Retry-After", "10") @@ -567,10 +586,46 @@ data: self.wfile.write(resp_body) except BrokenPipeError: pass - else: - self._send_json({"error": f"NIM API error {he.code}: {he.reason}"}) + return + raise RuntimeError(f"NIM API error {he.code}: {he.reason}") + + # Relay NIM's SSE stream to the browser as our own SSE. + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.send_header("Cache-Control", "no-cache") + self.send_header("X-Accel-Buffering", "no") + self.end_headers() + headers_sent = True + for raw in upstream: + line = raw.decode("utf-8", "replace").strip() + if not line.startswith("data:"): + continue + data = line[5:].strip() + if data == "[DONE]": + break + try: + obj = json.loads(data) + except json.JSONDecodeError: + continue + piece = obj.get("choices", [{}])[0].get("delta", {}).get("content", "") + if piece and not self._emit({"delta": piece}): + return + self._emit({"done": True}) except Exception as e: - self._send_json({"error": str(e)}) + if not headers_sent: + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.send_header("Cache-Control", "no-cache") + self.end_headers() + self._emit({"error": str(e)}) + + def _emit(self, obj: dict) -> bool: + try: + self.wfile.write(("data: " + json.dumps(obj) + "\n\n").encode("utf-8")) + self.wfile.flush() + return True + except (BrokenPipeError, ConnectionResetError): + return False def _send(self, body: str): data = body.encode("utf-8") diff --git a/spark-analytics/eu-central-1-staging/webview-ingress.yaml b/spark-analytics/eu-central-1-staging/webview-ingress.yaml index 6fb51b5..9833d22 100755 --- a/spark-analytics/eu-central-1-staging/webview-ingress.yaml +++ b/spark-analytics/eu-central-1-staging/webview-ingress.yaml @@ -9,6 +9,10 @@ metadata: cert-manager.io/cluster-issuer: "letsencrypt-prod" nginx.ingress.kubernetes.io/ssl-redirect: "true" nginx.ingress.kubernetes.io/force-ssl-redirect: "true" + # AI /explain streams Server-Sent Events — disable response buffering so + # chunks flush to the client immediately, and allow gaps between chunks. + nginx.ingress.kubernetes.io/proxy-buffering: "off" + nginx.ingress.kubernetes.io/proxy-read-timeout: "120" spec: ingressClassName: nginx rules: