Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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**
Expand Down
48 changes: 39 additions & 9 deletions backend/core/event_bus.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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:
Expand All @@ -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)
Expand Down
94 changes: 47 additions & 47 deletions frontend/src/hooks/useAppData.js
Original file line number Diff line number Diff line change
@@ -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';
Expand All @@ -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';

/**
Expand Down Expand Up @@ -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({
Expand All @@ -212,23 +208,27 @@ 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;
} catch (e) {}
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);
Comment thread
coderabbitai[bot] marked this conversation as resolved.
};
loadAll();
// Restore local UI state
Expand Down Expand Up @@ -290,7 +290,7 @@ export default function useAppData() {
setOmniUiRestoreComplete(true);
}
return () => {
cancelled = true;
cancelledRef.cancelled = true;
};
}, []);

Expand Down
67 changes: 67 additions & 0 deletions frontend/src/test/initialLoadRetry.test.js
Original file line number Diff line number Diff line change
@@ -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;');
});
});
29 changes: 29 additions & 0 deletions frontend/src/utils/initialLoadRetry.js
Original file line number Diff line number Diff line change
@@ -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);
}
}
78 changes: 78 additions & 0 deletions tests/test_event_bus_thread_emit.py
Original file line number Diff line number Diff line change
@@ -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)
Loading