diff --git a/CHANGELOG.md b/CHANGELOG.md index 3921cd84..72862191 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,8 @@ the frozen-backend fallback mirror it for their toolchains. - "Ready" now requires the deep health probe (a working database-backed route), not just the identity probe — a backend whose install broke underneath can no longer be announced up while every real request fails (#1548) - Supervisor restarts after repeat crashes now back off (immediate, then 5s, then 15s) instead of respawning back-to-back, so a tight crash loop can't burn the whole restart budget in seconds (#1548) +- Renaming, deleting, or revoking consent on a voice (and starring/clearing history, recording exports) now live-updates every open tab again — the sync routes' WebSocket events were silently dropped, which could look like "all my voices are gone" (#1561) — thanks @paoloantinori! + ## [0.5.0] — 2026-08-13 **Highlights** diff --git a/backend/core/event_bus.py b/backend/core/event_bus.py index baa3b995..74fd662c 100644 --- a/backend/core/event_bus.py +++ b/backend/core/event_bus.py @@ -23,9 +23,17 @@ _listeners: list[asyncio.Queue] = [] _lock = asyncio.Lock() +# The loop that serves /ws/events, captured on first use. Sync FastAPI +# endpoints (rename/delete profile, revoke consent) run in threadpool workers +# where `asyncio.get_running_loop()` raises, which used to silently drop their +# events — the UI then never refetched the voice list (#1158 class). +_serving_loop: asyncio.AbstractEventLoop | None = None + async def subscribe() -> asyncio.Queue: """Register a new listener. Returns a Queue that receives event dicts.""" + global _serving_loop + _serving_loop = asyncio.get_running_loop() q: asyncio.Queue = asyncio.Queue(maxsize=64) async with _lock: _listeners.append(q) @@ -56,12 +64,34 @@ def emit(kind: str, payload: dict[str, Any] | None = None) -> None: **(payload or {}), } event_str = json.dumps(event) + loop = asyncio.get_running_loop() if _on_event_loop() else _serving_loop + if loop is None: + # No serving loop yet — nobody to notify; dropping is correct. + logger.debug("No event loop — event dropped: %s", kind) + return try: - loop = asyncio.get_running_loop() - loop.create_task(_broadcast(event_str)) + if _on_event_loop(): + loop.create_task(_broadcast(event_str)) + else: + # Threadpool worker (sync endpoint body): the only thread-safe way in. + loop.call_soon_threadsafe(_schedule_broadcast, event_str) except RuntimeError: - # No event loop running (unlikely in FastAPI context but safe) - logger.debug("No event loop — event dropped: %s", kind) + # The serving loop closed between capture and use (app shutdown). + logger.debug("Event loop closed — event dropped: %s", kind) + + +def _on_event_loop() -> bool: + """True when called from the running event loop (async-context emit).""" + try: + asyncio.get_running_loop() + except RuntimeError: + return False + return True + + +def _schedule_broadcast(event_str: str) -> None: + """Run `_broadcast` on the serving loop; called via call_soon_threadsafe.""" + asyncio.get_running_loop().create_task(_broadcast(event_str)) async def _broadcast(event_str: str) -> None: @@ -73,11 +103,11 @@ async def _broadcast(event_str: str) -> None: q.put_nowait(event_str) except asyncio.QueueFull: # Slow consumer — drop oldest, then push. Not a race (#1163): - # every queue op runs on the single event loop, and there is - # no await between the QueueFull and this get_nowait/put_nowait - # pair — no consumer can interleave, so get_nowait cannot raise - # QueueEmpty here. emit() from a foreign thread drops the event - # before ever touching a queue (see the RuntimeError branch). + # every queue op runs on the single event loop (a foreign + # thread's emit() hands off via call_soon_threadsafe first), + # and there is no await between the QueueFull and this + # get_nowait/put_nowait pair — no consumer can interleave, so + # get_nowait cannot raise QueueEmpty here. try: q.get_nowait() q.put_nowait(event_str) diff --git a/frontend/src/hooks/useAppData.js b/frontend/src/hooks/useAppData.js index 815cf1cc..f4fe496d 100644 --- a/frontend/src/hooks/useAppData.js +++ b/frontend/src/hooks/useAppData.js @@ -1,4 +1,4 @@ -import { useState, useEffect, useLayoutEffect, useCallback, useRef } from 'react'; +import { useState, useEffect, useLayoutEffect, useRef } from 'react'; import { useAppStore } from '../store'; import { listProfiles } from '../api/profiles'; import { listHistory } from '../api/generate'; @@ -10,6 +10,7 @@ import { useModelStatus } from '../api/hooks'; import useRealtimeEvents from './useRealtimeEvents'; import { mergeDescribedAttrs } from '../utils/voiceInstruct'; import { sanitizeOmniUi } from '../utils/omniUiSchema'; +import { retryInitialLoad } from '../utils/initialLoadRetry'; import { queueJsonWrite } from '../utils/coalescedJsonStorage'; /** @@ -163,43 +164,38 @@ export default function useAppData() { }, [modelStatus, modelSubStage, modelDetail, modelError, modelProgress]); // ── Data loading callbacks ── - // Failures keep the previous list (better than blanking the UI), but are - // logged so "my voices/history vanished" reports carry a cause (#1158). - const loadProfiles = useCallback(async () => { - try { - setProfiles(await listProfiles()); - } catch (e) { - console.warn('Failed to load voice profiles:', e); - } - }, []); - const loadHistory = useCallback(async () => { - try { - setHistory(await listHistory()); - } catch (e) { - console.warn('Failed to load generation history:', e); - } - }, []); - const loadDubHistory = useCallback(async () => { - try { - setDubHistory(await listDubHistory()); - } catch (e) { - console.warn('Failed to load dub history:', e); - } - }, []); - const loadProjects = useCallback(async () => { - try { - setStudioProjects(await listProjects()); - } catch (e) { - console.warn('Failed to load projects:', e); - } - }, []); - const loadExportHistory = useCallback(async () => { - try { - setExportHistory(await listExportHistory()); - } catch (e) { - console.warn('Failed to load export history:', e); - } - }, []); + // WS-triggered reloads swallow failures: keeping the previous list is + // better than blanking the UI, and the warn gives "my voices vanished" + // reports a cause (#1158). The INITIAL load passes `{ rethrow: true }` so + // retryInitialLoad can retry — there is no previous list to keep yet. + // Each loader is last-write-wins by invocation order: a slow in-flight + // request (the initial retry loop overlaps freely with WS reloads) must + // not overwrite the fresher list a later reload already applied. Plain + // per-render closures over stable imports/setters — a useRef-free module + // would need hooks inside a helper, which rules-of-hooks forbids. + const loadersRef = useRef({ profiles: 0, history: 0, dub: 0, projects: 0, exports: 0 }); + const makeLoader = + (key, fetch, set, label) => + async ({ rethrow } = {}) => { + const gen = ++loadersRef.current[key]; + try { + const data = await fetch(); + if (gen === loadersRef.current[key]) set(data); + } catch (e) { + console.warn(`Failed to load ${label}:`, e); + if (rethrow) throw e; + } + }; + const loadProfiles = makeLoader('profiles', listProfiles, setProfiles, 'voice profiles'); + const loadHistory = makeLoader('history', listHistory, setHistory, 'generation history'); + const loadDubHistory = makeLoader('dub', listDubHistory, setDubHistory, 'dub history'); + const loadProjects = makeLoader('projects', listProjects, setStudioProjects, 'projects'); + const loadExportHistory = makeLoader( + 'exports', + listExportHistory, + setExportHistory, + 'export history', + ); // ── WebSocket real-time updates ── useRealtimeEvents({ @@ -212,10 +208,10 @@ export default function useAppData() { // ── Initial data load with backend retry ── useEffect(() => { - let cancelled = false; + const cancelledRef = { cancelled: false }; const loadAll = async () => { let delay = 1000; - while (!cancelled) { + while (!cancelledRef.cancelled) { try { await apiModelStatus(); break; @@ -223,12 +219,16 @@ export default function useAppData() { await new Promise((r) => setTimeout(r, delay)); delay = Math.min(delay * 2, 4000); } - if (cancelled) return; - loadProfiles(); - loadHistory(); - loadDubHistory(); - loadProjects(); - loadExportHistory(); + if (cancelledRef.cancelled) return; + // Initial loads retry until FIRST success (#1158 class): a later + // (WS-triggered) reload failure keeps the previous list, but the first + // load has no previous list to keep — one transient failure used to + // leave the panel empty, which read as "my voices are gone". + retryInitialLoad(() => loadProfiles({ rethrow: true }), cancelledRef); + retryInitialLoad(() => loadHistory({ rethrow: true }), cancelledRef); + retryInitialLoad(() => loadDubHistory({ rethrow: true }), cancelledRef); + retryInitialLoad(() => loadProjects({ rethrow: true }), cancelledRef); + retryInitialLoad(() => loadExportHistory({ rethrow: true }), cancelledRef); }; loadAll(); // Restore local UI state @@ -290,7 +290,7 @@ export default function useAppData() { setOmniUiRestoreComplete(true); } return () => { - cancelled = true; + cancelledRef.cancelled = true; }; }, []); diff --git a/frontend/src/test/initialLoadRetry.test.js b/frontend/src/test/initialLoadRetry.test.js new file mode 100644 index 00000000..3cd0dd63 --- /dev/null +++ b/frontend/src/test/initialLoadRetry.test.js @@ -0,0 +1,67 @@ +import { describe, it, expect, vi } from 'vitest'; +import { retryInitialLoad } from '../utils/initialLoadRetry'; + +// #1158 class: the initial data loads (profiles/history/…) ran exactly once +// after the backend became reachable. A transient failure on that single call +// (backend restarting, WS reconnect window, a 502 from the LAN gate) left the +// voices panel empty with no retry — which reads to users as "all my voices +// are gone". The load itself already keeps the previous list on later (WS) +// reloads; only the INITIAL load must retry until first success. + +describe('retryInitialLoad (#1158 class)', () => { + it('retries a failing loader until it succeeds', async () => { + let calls = 0; + const loader = vi.fn(async () => { + calls += 1; + if (calls < 3) throw new Error('transient'); + }); + await retryInitialLoad(loader, { baseDelayMs: 1 }); + expect(calls).toBe(3); + }); + + it('does not retry after first success', async () => { + const loader = vi.fn(async () => {}); + await retryInitialLoad(loader, { baseDelayMs: 1 }); + expect(loader).toHaveBeenCalledTimes(1); + }); + + it('stops retrying when cancelled', async () => { + const loader = vi.fn(async () => { + throw new Error('down'); + }); + const opts = { baseDelayMs: 1 }; + const p = retryInitialLoad(loader, opts); + opts.cancelled = true; + await p; + // At most one attempt plus whatever was already in flight; the point is + // it terminates instead of looping forever after cancellation. + expect(loader.mock.calls.length).toBeLessThan(5); + }); + + // #1158 wiring contract: the initial load must pass loaders that REJECT on + // failure. useAppData's loadProfiles-style loaders swallow errors by design + // (WS reloads keep the previous list); if the initial-load call site ever + // stops using { rethrow: true }, the retry helper resolves on attempt one + // and the fix is silently inert again (skeptic finding F1). + it('integration: useAppData wires rethrowing loaders into the retry', async () => { + const { readFileSync } = await import('node:fs'); + // jsdom replaces the global URL; Node's fileURLToPath needs a REAL node: + // URL instance, so build one from the specifier's pathname directly. + const nodeUrl = await import('node:url'); + const NodeURL = nodeUrl.URL; + const hookPath = new NodeURL('../hooks/useAppData.js', import.meta.url).pathname; + const src = readFileSync(hookPath, 'utf8'); + const initialBlock = src.slice(src.indexOf('retryInitialLoad(')); + for (const loader of [ + 'loadProfiles', + 'loadHistory', + 'loadDubHistory', + 'loadProjects', + 'loadExportHistory', + ]) { + expect(initialBlock).toContain(`${loader}({ rethrow: true })`); + } + // and the loader factory must actually rethrow when asked + expect(src).toContain('if (rethrow) throw e;'); + }); +}); diff --git a/frontend/src/utils/initialLoadRetry.js b/frontend/src/utils/initialLoadRetry.js new file mode 100644 index 00000000..a300eb1c --- /dev/null +++ b/frontend/src/utils/initialLoadRetry.js @@ -0,0 +1,29 @@ +/** + * retryInitialLoad — run an initial data load until first success. + * + * The INITIAL app load differs from later (WS-triggered) reloads: on reload a + * failure keeps the previous list, but on first load there IS no previous + * list, so a single transient failure left the panel empty forever and read + * to users as "my voices are gone" (#1158 class). Retry with backoff until + * the first success; afterwards the caller's normal keep-previous-list + * behavior takes over. + * + * `opts.cancelled` is the caller's mount-cancellation flag — set it to true + * and any pending wait resolves without another attempt. + */ +export async function retryInitialLoad(loader, opts = {}) { + const baseDelayMs = opts.baseDelayMs ?? 500; + const maxDelayMs = opts.maxDelayMs ?? 4000; + let delay = 0; // first attempt is immediate; only failures pay the backoff + for (;;) { + if (opts.cancelled) return; + try { + await loader(); + return; + } catch { + // transient — retry + } + await new Promise((r) => setTimeout(r, delay || baseDelayMs)); + delay = Math.min((delay || baseDelayMs) * 2, maxDelayMs); + } +} diff --git a/tests/test_event_bus_thread_emit.py b/tests/test_event_bus_thread_emit.py new file mode 100644 index 00000000..49b9960c --- /dev/null +++ b/tests/test_event_bus_thread_emit.py @@ -0,0 +1,78 @@ +"""event_bus.emit must deliver from threadpool threads (sync FastAPI endpoints). + +PUT /profiles/{id} (rename), DELETE /profiles/{id}, and DELETE .../consent are +sync endpoints, so Starlette runs their bodies in a threadpool worker with no +running event loop. event_bus.emit() used to call asyncio.get_running_loop() +and silently drop the event (DEBUG log only) from exactly those threads, which +reached users as "I renamed a voice and the list went stale / looked empty" +(the frontend only refetches the voice list on the WS "profiles" event). + +The test drives the real failure shape: subscribe on the serving loop, call +emit() from a plain thread (as the threadpool does), and assert the event +arrives in the listener queue. +""" + +from __future__ import annotations + +import asyncio +import json +import threading + +import pytest + + +@pytest.fixture +def bus(): + """Resolve the module per test — binding it at collection lets another + suite's `sys.modules` rebinding make this exercise a different object.""" + return __import__("core.event_bus", fromlist=["emit"]) + + +def test_emit_from_thread_reaches_serving_loop(bus, tmp_path): + loop = asyncio.new_event_loop() + received: list[str] = [] + started = threading.Event() + done = threading.Event() + + def run_loop(): + asyncio.set_event_loop(loop) + loop.run_until_complete(_serve(bus, received, started, done)) + + t = threading.Thread(target=run_loop, name="test-serving-loop") + t.start() + try: + assert started.wait(2.0), "serving loop never subscribed" + + # The bug's exact context: no running loop in this thread, like a + # Starlette threadpool worker executing a sync endpoint body. + def sync_endpoint_body(): + bus.emit("profiles", {"action": "updated", "id": "abc123"}) + + worker = threading.Thread(target=sync_endpoint_body, name="threadpool-worker") + worker.start() + worker.join(2.0) + + assert done.wait(2.0), ( + "emit() from a threadpool thread was dropped — the WS 'profiles' " + "event never reached the serving loop's listener queue" + ) + payload = json.loads(received[0]) + assert payload["kind"] == "profiles" + assert payload["action"] == "updated" + assert payload["id"] == "abc123" + finally: + loop.call_soon_threadsafe(done.set) + t.join(2.0) + loop.close() + + +async def _serve(bus, received: list[str], started: threading.Event, done: threading.Event): + q = await bus.subscribe() + started.set() + # Await the event itself (no sleep-polling): a failure surfaces as + # asyncio.TimeoutError, which fails the test with a clear traceback. + try: + received.append(await asyncio.wait_for(q.get(), 2.0)) + finally: + done.set() + await bus.unsubscribe(q)