From f8ad8778810734077fb77cef38bccdadb8236330 Mon Sep 17 00:00:00 2001 From: duffyduck Date: Wed, 28 Oct 2026 20:17:40 +0100 Subject: [PATCH] fix(llm): fertige Antwort geht bei RVS-Abbruch nicht mehr verloren MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 28.10., 17:51:42: alle Clients verloren gleichzeitig die RVS-Verbindung (RVS/Proxy kurz weg) — waehrend Gemma 31 s an einer Anfrage von Marina rechnete. Nach 2 s war alles wieder verbunden, aber die fertige Antwort ging ueber die tote Verbindung raus und wurde verworfen ("Send fehlgeschlagen (llm_response)"); Marina wartete bis zu ihrer Frist (660 s) ins Leere. llm-adapter: llm_response wird ueber die aktuelle Verbindung geschickt, ist gerade keine offen, aufgehoben (5 min) und nach dem Wiederverbinden nachgeschickt. RVS: gezielte Antwort, deren Fragender nur neu verbunden ist (alte Verbindung weg), geht bis 2 MB an den Raum statt verworfen zu werden — der Fragende erkennt sie an der requestId. Dateien bleiben gezielt. Co-Authored-By: Claude Opus 5.5 --- rvs/server.js | 14 +++++++++- xtts/llm-adapter/adapter.py | 54 +++++++++++++++++++++++++++++++------ 2 files changed, 59 insertions(+), 9 deletions(-) diff --git a/rvs/server.js b/rvs/server.js index c1ef06f..e1c9e9c 100644 --- a/rvs/server.js +++ b/rvs/server.js @@ -188,6 +188,9 @@ const TARGETED_REPLIES = new Set(TARGETED.values()); // Nachrichten-ID) — darum ein Set, sonst bekaeme nur der letzte die Antwort. const pendingReplies = new Map(); const PENDING_TTL_MS = 180_000; +// Antworten bis zu dieser Groesse gehen an den Raum, wenn der Fragende nur +// neu verbunden ist (siehe Weiterleitung). Dateien sind groesser. +const REJOIN_BROADCAST_MAX = 2 * 1024 * 1024; function pendingKey(replyType, requestId) { return `${replyType}\u0000${requestId}`; @@ -696,7 +699,16 @@ function registerClient(ws, token) { rememberRequester(ws, msg); // Gezielte Antwort? Dann NUR an den Fragenden — nicht an den ganzen Raum. - const targets = targetedRecipients(msg); + let targets = targetedRecipients(msg); + // Fragender inzwischen neu verbunden (RVS-/Proxy-Neustart waehrend die + // Antwort noch lief): seine alte Verbindung gibt es nicht mehr, die neue + // kennen wir nicht. Kleine Antworten (LLM-Antwort, Status) dann an den + // Raum — der Fragende erkennt sie an der requestId. Grosse (Dateien) + // bleiben verworfen, sonst gingen hunderte MB an alle. + if (targets && targets.length === 0 && raw.length <= REJOIN_BROADCAST_MAX) { + log(`${msg.type}: Fragender neu verbunden — ${(raw.length / 1024).toFixed(0)}KB an den Raum`); + targets = null; + } if (targets) { const data = raw.toString(); let sent = 0; diff --git a/xtts/llm-adapter/adapter.py b/xtts/llm-adapter/adapter.py index 4fd3248..e6bdd96 100644 --- a/xtts/llm-adapter/adapter.py +++ b/xtts/llm-adapter/adapter.py @@ -332,15 +332,49 @@ async def _reload_llama() -> None: await asyncio.sleep(3) # Poll-Intervall + Puffer, bevor jemand fragt +# Antworten, die nicht verloren gehen duerfen: Bricht die RVS-Verbindung ab, +# waehrend eine Anfrage noch rechnet (z.B. RVS-/Proxy-Neustart), wuerde die +# fertige Antwort sonst ueber die tote Verbindung geschickt und verworfen — der +# Fragende (Schmiede, ARIA) wartet dann bis zu seiner Frist ins Leere (am +# 28.10. 11 Minuten). Jetzt: ueber die NEUE Verbindung senden, notfalls bis zum +# Wiederverbinden aufheben. +_RELIABLE_TYPES = {"llm_response"} +_OUTBOX_TTL_S = 300 +_outbox: list = [] # [(ts, mtype, payload)] +_current_ws = None # die aktuell offene RVS-Verbindung + + async def _send(ws, mtype: str, payload: dict) -> None: - try: - await ws.send(json.dumps({ - "type": mtype, - "payload": payload, - "timestamp": int(time.time() * 1000), - })) - except Exception as e: - logger.warning("Send fehlgeschlagen (%s): %s", mtype, e) + data = json.dumps({"type": mtype, "payload": payload, "timestamp": int(time.time() * 1000)}) + targets = [ws] if ws is _current_ws or _current_ws is None else [ws, _current_ws] + last_err = None + for t in targets: + try: + await t.send(data) + return + except Exception as e: + last_err = e + if mtype in _RELIABLE_TYPES: + _outbox.append((time.time(), mtype, payload)) + logger.warning("Send fehlgeschlagen (%s): %s — aufgehoben, geht nach dem Wiederverbinden raus", + mtype, last_err) + else: + logger.warning("Send fehlgeschlagen (%s): %s", mtype, last_err) + + +async def _flush_outbox(ws) -> None: + """Nach dem (Wieder-)Verbinden: aufgehobene Antworten nachschicken.""" + now = time.time() + keep = [(t, m, p) for (t, m, p) in _outbox if now - t <= _OUTBOX_TTL_S] + _outbox.clear() + for t, m, p in keep: + try: + await ws.send(json.dumps({"type": m, "payload": p, "timestamp": int(time.time() * 1000)})) + logger.info("Aufgehobene Antwort nachgeschickt (%s, %.0f s alt)", m, now - t) + except Exception as e: + _outbox.append((t, m, p)) + logger.warning("Nachschicken fehlgeschlagen (%s): %s", m, e) + return def _sanitize_messages(messages: list) -> list: @@ -1364,6 +1398,7 @@ async def _handle_models_request(ws, payload: dict) -> None: async def _run() -> None: + global _current_ws if not RVS_HOST: logger.error("RVS_HOST nicht gesetzt — Abbruch") return @@ -1401,6 +1436,9 @@ async def _run() -> None: f" (TCP {RVS_HOST})" if (use_tls and RVS_SNI) else "", LLAMA_URL) async with websockets.connect(url, **connect_kwargs) as ws: logger.info("RVS verbunden — llm-adapter online") + _current_ws = ws + if _outbox: + await _flush_outbox(ws) retry_s = 2 tls_fallback_tried = False ping_task = asyncio.create_task(_worker_register(ws))