From 2d2c182a0cdb734e406bc950b362a78531e835dd Mon Sep 17 00:00:00 2001 From: duffyduck Date: Sat, 19 Sep 2026 04:12:16 +0200 Subject: [PATCH] =?UTF-8?q?fix(fleet):=20worker=5Fhello=20periodisch=20wie?= =?UTF-8?q?derholen=20(wie=20Satellit)=20=E2=80=94=20Box=20wird=20nach=20D?= =?UTF-8?q?iagnostic/Bridge-Neustart=20wieder=20sichtbar?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Worker sendeten worker_hello nur EINMAL beim Connect. Startet das Diagnostic oder die Bridge NACH der Box neu, verpassen sie das hello (RVS spielt es nicht nach) und sehen nur noch worker_ping → die Box taucht nicht in der Compute- Flotte auf. Der Satellit ist genau deshalb zuverlaessig sichtbar: er wiederholt sat_hello periodisch. - alle vier Worker (f5tts/whisper/voxtral/llm-adapter): worker_hello wird jetzt zusaetzlich alle ~30s (jeder 3. Ping-Zyklus) wiederholt → ein neu gestartetes Diagnostic/Bridge lernt die Box innerhalb von 30s, ohne Box-Neustart. - diagnostic/server.js: Voice-Reconcile (f5tts) laeuft nur beim ERSTEN/erneuten Auftauchen der Box, nicht bei jedem 30s-hello-Resend (sonst Push/Pull-Dauerlauf). Co-Authored-By: Claude Opus 4.8 --- diagnostic/server.js | 8 ++++++-- xtts/f5tts/bridge.py | 18 ++++++++++++------ xtts/llm-adapter/adapter.py | 4 ++++ xtts/voxtral/bridge.py | 12 ++++++++---- xtts/whisper/bridge.py | 16 +++++++++++----- 5 files changed, 41 insertions(+), 17 deletions(-) diff --git a/diagnostic/server.js b/diagnostic/server.js index b56965e..c044887 100644 --- a/diagnostic/server.js +++ b/diagnostic/server.js @@ -1076,6 +1076,10 @@ function connectRVS(forcePlain) { const p = msg.payload || {}; if (p.instanceId) { const prev = workers.get(p.instanceId) || {}; + // War die Box vor diesem hello schon frisch gesehen? worker_hello wird + // jetzt alle ~30s wiederholt — Reconcile nur beim ERSTEN/erneuten + // Auftauchen, nicht bei jedem Resend. + const wasFresh = prev.last_seen && (Date.now() - prev.last_seen < WORKER_OFFLINE_MS); workers.set(p.instanceId, { instanceId: p.instanceId, service: p.service || "", node: p.node || "", gpus: p.gpus || "", model: p.model || "", @@ -1083,8 +1087,8 @@ function connectRVS(forcePlain) { busy: !!prev.busy, last_seen: Date.now(), }); broadcastWorkers(); - // Voice-Flotte: f5tts-Box gerade online → Stimmen abgleichen/provisionieren. - if ((p.service || "") === "f5tts") reconcileVoices(p.instanceId, p.voices); + // Voice-Flotte: f5tts-Box NEU online → Stimmen abgleichen/provisionieren. + if ((p.service || "") === "f5tts" && !wasFresh) reconcileVoices(p.instanceId, p.voices); } } else if (msg.type === "worker_ping") { // Heartbeat eines Workers (traegt busy-Status). diff --git a/xtts/f5tts/bridge.py b/xtts/f5tts/bridge.py index 21be3e6..4e4c01c 100644 --- a/xtts/f5tts/bridge.py +++ b/xtts/f5tts/bridge.py @@ -842,17 +842,23 @@ async def _broadcast_status(ws, state: str, **extra) -> None: async def _worker_register(ws, *, model: str = "", busy_fn=None) -> None: """Meldet diesen Worker bei der aria-bridge an (worker_hello) und haelt die - Flotten-Registry per periodischem worker_ping (mit busy-Status) frisch.""" + Flotten-Registry per periodischem worker_ping frisch. worker_hello wird + zusaetzlich alle ~30s WIEDERHOLT (wie der Satellit sat_hello), damit auch ein + neu gestartetes Diagnostic/Bridge uns lernt — RVS spielt hellos nicht nach.""" + def _hello(): + return {"instanceId": INSTANCE_ID, "service": WORKER_SERVICE, + "node": NODE_NAME, "gpus": GPU_IDS, "model": model, + "voices": _local_voice_names()} try: - await _send(ws, "worker_hello", { - "instanceId": INSTANCE_ID, "service": WORKER_SERVICE, - "node": NODE_NAME, "gpus": GPU_IDS, "model": model, - "voices": _local_voice_names(), # fuer Voice-Provisioning-Reconciliation - }) + await _send(ws, "worker_hello", _hello()) + n = 0 while True: await asyncio.sleep(WORKER_PING_INTERVAL_S) + n += 1 busy = bool(busy_fn()) if busy_fn else False await _send(ws, "worker_ping", {"instanceId": INSTANCE_ID, "busy": busy}) + if n % 3 == 0: + await _send(ws, "worker_hello", _hello()) except asyncio.CancelledError: raise except Exception: diff --git a/xtts/llm-adapter/adapter.py b/xtts/llm-adapter/adapter.py index eebbbee..91937e8 100644 --- a/xtts/llm-adapter/adapter.py +++ b/xtts/llm-adapter/adapter.py @@ -260,10 +260,14 @@ async def _worker_register(ws) -> None: Flotten-Registry per periodischem worker_ping (mit busy-Status) frisch.""" try: await _announce(ws) + n = 0 while True: await asyncio.sleep(WORKER_PING_INTERVAL_S) + n += 1 await _send(ws, "worker_ping", {"instanceId": INSTANCE_ID, "busy": _inflight > 0}) + if n % 3 == 0: # ~30s worker_hello wiederholen (wie der Satellit) → + await _announce(ws) # auch neu gestartetes Diagnostic/Bridge lernt uns except asyncio.CancelledError: raise except Exception: diff --git a/xtts/voxtral/bridge.py b/xtts/voxtral/bridge.py index 29dc5f6..79ee42a 100644 --- a/xtts/voxtral/bridge.py +++ b/xtts/voxtral/bridge.py @@ -713,15 +713,19 @@ async def _broadcast_status(ws, state: str, **extra) -> None: async def _worker_register(ws, *, model: str = "", busy_fn=None) -> None: """Meldet diesen Worker bei der aria-bridge an (worker_hello) und haelt die Flotten-Registry per periodischem worker_ping (mit busy-Status) frisch.""" + def _hello(): + return {"instanceId": INSTANCE_ID, "service": WORKER_SERVICE, + "node": NODE_NAME, "gpus": GPU_IDS, "model": model} try: - await _send(ws, "worker_hello", { - "instanceId": INSTANCE_ID, "service": WORKER_SERVICE, - "node": NODE_NAME, "gpus": GPU_IDS, "model": model, - }) + await _send(ws, "worker_hello", _hello()) + n = 0 while True: await asyncio.sleep(WORKER_PING_INTERVAL_S) + n += 1 busy = bool(busy_fn()) if busy_fn else False await _send(ws, "worker_ping", {"instanceId": INSTANCE_ID, "busy": busy}) + if n % 3 == 0: # ~30s worker_hello wiederholen (wie der Satellit) + await _send(ws, "worker_hello", _hello()) except asyncio.CancelledError: raise except Exception: diff --git a/xtts/whisper/bridge.py b/xtts/whisper/bridge.py index 2fb4ba5..e0a50ed 100644 --- a/xtts/whisper/bridge.py +++ b/xtts/whisper/bridge.py @@ -837,16 +837,22 @@ async def _broadcast_status(ws, state: str, **extra) -> None: async def _worker_register(ws, *, model: str = "", busy_fn=None) -> None: """Meldet diesen Worker bei der aria-bridge an (worker_hello) und haelt die - Flotten-Registry per periodischem worker_ping (mit busy-Status) frisch.""" + Flotten-Registry per periodischem worker_ping frisch. worker_hello wird alle + ~30s WIEDERHOLT (wie der Satellit), damit ein neu gestartetes Diagnostic/ + Bridge uns lernt — RVS spielt hellos nicht nach.""" + def _hello(): + return {"instanceId": INSTANCE_ID, "service": WORKER_SERVICE, + "node": NODE_NAME, "gpus": GPU_IDS, "model": model} try: - await _send(ws, "worker_hello", { - "instanceId": INSTANCE_ID, "service": WORKER_SERVICE, - "node": NODE_NAME, "gpus": GPU_IDS, "model": model, - }) + await _send(ws, "worker_hello", _hello()) + n = 0 while True: await asyncio.sleep(WORKER_PING_INTERVAL_S) + n += 1 busy = bool(busy_fn()) if busy_fn else False await _send(ws, "worker_ping", {"instanceId": INSTANCE_ID, "busy": busy}) + if n % 3 == 0: + await _send(ws, "worker_hello", _hello()) except asyncio.CancelledError: raise except Exception: