openvibers/openvibe.chat Commits
Order audio queue start and stop: recover gates pumps, stop awaits work (#6)
Changes
@@@ -37,6 +37,16 @@* as above, fails rows queued too long ago to still matter, and plays the rest once the room has* listeners again: a second after the first socket rejoins it (clients come back 1–4 s after a* restart notice), or after RESUME_GRACE_MS whatever happens.++ *++ * Start and stop are ordered, so a restart is the same every time:++ * - `recover()` closes the pump gate before its first await and opens it when the queue is settled and held:++ * a pump asked for meanwhile (a socket's request, the heartbeat) is parked and runs after, so a fresh claim of++ * this process is never read by recover() as a row left over from before the restart, and nothing plays out of order;++ * - `stop()` is async: it cancels the pacing timers, then waits for the work in flight (a pump making a clip, a++ * finish, a tick, a recover) before it resolves, and what that work does after the stop is limited to what a++ * restart settles anyway: a clip made but not yet delivered is left claimed and undelivered (replayed once), a++ * clip delivered is left with its started_at (played, 'delivered before a restart') and gets no pacing timer.++ * The chat server awaits it before it closes its sockets and the database.* A keyed request (a TTS of chat message m<id>) is queued once per room, so a retried bridge call* or a double delivery never reads it twice, even across a restart.*@@@ -134,7 +144,7 @@ async function getRequest(id) { return await db.get('SELECT * FROM audio_request* Move one request along the state machine. Returns true when it moved (the row was in a state* that allows `to`); false when something else moved it first.*/--async function transition(id, to, { error = null, actor = null, durationMs = null, now = Date.now() } = {}) {++async function transition(id, to, { error = null, actor = null, durationMs = null, now = clock() } = {}) {const from = Object.keys(TRANSITIONS).filter((s) => TRANSITIONS[s].includes(to));if (!from.length) return false;const final = to !== 'playing';@@@ -154,7 +164,11 @@ const performers = {}; // kind → async (row, payload) → { frame, durationMlet deliver = null; // (row, frame) → voidlet instanceId = process.env.CHAT_AUDIO_INSTANCE_ID || `${os.hostname()}:${process.env.PORT || 4400}`;let clock = () => Date.now();++let timers = { setTimeout, clearTimeout }; // the pacing, hold and deferral timers (tests drive them by hand)let heartbeat = null;++const inflight = new Set(); // pumps, finishes, ticks and recovers still running: stop() waits for them++const deferred = new Set(); // rooms a pump was asked for while recover() ran: pumped when it is done++let recovering = false;const waiting = new Set(); // rooms whose claim lost to another process's playing row: pumped again by the heartbeatconst rooms = new Map(); // room → { busy, timer, playingId }let stopped = false;@@@ -166,32 +180,37 @@ function roomState(room) {/** Hold a room's queue until `until` (restart recovery), then play. A shorter hold replaces a longer one. */function holdRoom(room, until) {++ if (stopped) return;const st = roomState(room);if (st.holdUntil && st.holdUntil <= until) return;st.holdUntil = until;-- if (st.holdTimer) clearTimeout(st.holdTimer);-- st.holdTimer = setTimeout(() => { st.holdUntil = 0; st.holdTimer = null; pump(room).catch((err) => console.warn('[AudioQueue] pump:', err.message)); }, Math.max(0, until - Date.now()));++ if (st.holdTimer) timers.clearTimeout(st.holdTimer);++ st.holdTimer = timers.setTimeout(() => { st.holdUntil = 0; st.holdTimer = null; pump(room).catch((err) => console.warn('[AudioQueue] pump:', err.message)); }, Math.max(0, until - clock()));if (st.holdTimer.unref) st.holdTimer.unref();}/** A socket joined a room: a queue held since a restart plays shortly. */function roomJoined(room) {const st = room && rooms.get(room);-- if (st && st.holdUntil > Date.now()) holdRoom(room, Date.now() + RESUME_SETTLE_MS);++ if (st && st.holdUntil > clock()) holdRoom(room, clock() + RESUME_SETTLE_MS);}/*** Wire the queue to the chat server: how each kind is made, and how a frame reaches a room.* `performers[kind](row, payload)` resolves { frame, durationMs } or null (nothing to play).* `instanceId` names this process's claims (default: host and port); `clock` (ms) is for tests.-- * `heartbeat: false` leaves lease renewal and the sweep to the caller (tests call tick()).++ * `heartbeat: false` leaves lease renewal and the sweep to the caller (tests call tick()); `timers`++ * ({ setTimeout, clearTimeout }) replaces the pacing, hold and deferral timers (tests fire them by hand).*/--function init({ performers: p, deliver: d, instanceId: id = null, clock: c = null, heartbeat: beat = true }) {++function init({ performers: p, deliver: d, instanceId: id = null, clock: c = null, timers: t = null, heartbeat: beat = true }) {Object.assign(performers, p || {});deliver = d;if (id) instanceId = String(id);if (c) clock = c;++ if (t) timers = { ...timers, ...t };stopped = false;++ recovering = false;++ deferred.clear();if (heartbeat) clearInterval(heartbeat);heartbeat = null;if (beat) {@@@ -200,17 +219,37 @@ function init({ performers: p, deliver: d, instanceId: id = null, clock: c = nul}}--/** Stop pacing timers (shutdown). States stay as they are; recover() picks them up at boot. */--function stop() {++/** Run `work` as something stop() and idle() wait for. */++function track(work) {++ inflight.add(work);++ const done = () => inflight.delete(work);++ work.then(done, done);++ return work;++}++/** Resolves once nothing is in flight (what a timer or a request started has finished). */++async function idle() {++ while (inflight.size) await Promise.allSettled([...inflight]);++}++++/**++ * Stop (shutdown): cancel the pacing timers, then wait (at most `graceMs`) for the work in flight, so nothing++ * of this process's queue is still touching the database or the sockets once it resolves. States stay as they are;++ * recover() picks them up at boot. Never rejects.++ */++async function stop({ graceMs = 3000 } = {}) {stopped = true;if (heartbeat) clearInterval(heartbeat);heartbeat = null;waiting.clear();++ deferred.clear();for (const st of rooms.values()) {-- if (st.timer) clearTimeout(st.timer);-- if (st.holdTimer) clearTimeout(st.holdTimer);++ if (st.timer) timers.clearTimeout(st.timer);++ if (st.holdTimer) timers.clearTimeout(st.holdTimer);st.timer = null; st.holdTimer = null;}++ let bound;++ await Promise.race([idle(), new Promise((resolve) => { bound = setTimeout(resolve, graceMs); })]);++ clearTimeout(bound);rooms.clear();}@@@ -243,7 +282,7 @@ async function enqueue({ kind, streamId = null, channelUserId = null, requestedBif (!r.changes) return { queued: false, reason: 'duplicate' };return { queued: true, id: Number(r.lastInsertRowid), room };});-- if (result.queued) setImmediate(() => { pump(result.room).catch((err) => console.warn('[AudioQueue] pump:', err.message)); });++ if (result.queued) timers.setTimeout(() => { pump(result.room).catch((err) => console.warn('[AudioQueue] pump:', err.message)); }, 0);return result;}@@@ -278,17 +317,20 @@ async function claimOrBusy(id, now) {}}--/** Play the next request of a room if none is playing. */--async function pump(room) {++/** Play the next request of a room if none is playing (not while recover() runs: the room is pumped when it is done). */++function pump(room) { return track(pumpRoom(room)); }++async function pumpRoom(room) {if (stopped || !room) return;++ if (recovering) { deferred.add(room); return; }const st = roomState(room);-- if (st.busy || st.playingId || st.holdUntil > Date.now()) return;++ if (st.busy || st.playingId || st.holdUntil > clock()) return;st.busy = true;try {for (;;) {if (stopped) return;let row;try { row = await db.get("SELECT * FROM audio_requests WHERE room = ? AND state = 'queued' ORDER BY id LIMIT 1", [room]); } catch { return; }++ if (stopped) return;if (!row) { waiting.delete(room); return; }if (row.attempts >= MAX_ATTEMPTS) { await transition(row.id, 'failed', { error: 'gave up after repeated attempts', actor: 'system' }); continue; }const { row: claimed, busy } = await claimOrBusy(row.id, clock());@@@ -310,12 +352,12 @@ async function pump(room) {if (error) { await transition(row.id, 'failed', { error: String(error).slice(0, 300), actor: 'system' }); continue; }const durationMs = Math.round(Math.min(MAX_PLAY_MS, Math.max(MIN_PLAY_MS, Number(made.durationMs) || estimatePlayMs(made.frame))));// Skipped or cleared while it was being made: nothing goes out.-- const now = Date.now();-- const started = await db.run("UPDATE audio_requests SET started_at = ?, duration_ms = ? WHERE id = ? AND state = 'playing'", [now, durationMs, row.id]);++ const started = await db.run("UPDATE audio_requests SET started_at = ?, duration_ms = ? WHERE id = ? AND state = 'playing'", [clock(), durationMs, row.id]);if (!started.changes) continue;try { if (deliver) deliver(row, { ...made.frame, request_id: row.id }); } catch (err) { console.warn('[AudioQueue] deliver:', err.message); }++ if (stopped) return; // delivered (started_at is set: a restart counts it played); stop() cleared the timers, so no window to pacest.playingId = row.id;-- st.timer = setTimeout(() => { finish(room, row.id, 'played').catch((err) => console.warn('[AudioQueue] finish:', err.message)); }, durationMs + GAP_MS);++ st.timer = timers.setTimeout(() => { finish(room, row.id, 'played').catch((err) => console.warn('[AudioQueue] finish:', err.message)); }, durationMs + GAP_MS);if (st.timer.unref) st.timer.unref();return;}@@@ -323,15 +365,16 @@ async function pump(room) {}/** End the playing request of a room (its window ended, it was skipped, or the client reported). */--async function finish(room, id, to, opts = {}) {++function finish(room, id, to, opts = {}) { return track(finishRoom(room, id, to, opts)); }++async function finishRoom(room, id, to, opts) {const st = roomState(room);let moved = false;try { moved = await transition(id, to, opts); } catch { moved = false; }if (st.playingId === id) {-- if (st.timer) clearTimeout(st.timer);++ if (st.timer) timers.clearTimeout(st.timer);st.timer = null;st.playingId = null;-- setImmediate(() => { pump(room).catch((err) => console.warn('[AudioQueue] pump:', err.message)); });++ if (!stopped) timers.setTimeout(() => { pump(room).catch((err) => console.warn('[AudioQueue] pump:', err.message)); }, 0);}return moved;}@@@ -411,7 +454,8 @@ async function reclaimDead({ now = clock(), mine = false } = {}) {* The heartbeat (every HEARTBEAT_MS, or by hand in tests): renew this process's leases, settle dead owners' rows* and pump the rooms that freed, and retry rooms whose claim lost to another process.*/--async function tick({ now = clock() } = {}) {++function tick(opts) { return track(tickOnce(opts)); }++async function tickOnce({ now = clock() } = {}) {if (stopped) return { renewed: 0, rooms: [] };const renewed = (await db.run("UPDATE audio_requests SET lease_until = ? WHERE state = 'playing' AND claimed_by = ?", [now + LEASE_MS, instanceId])).changes;const swept = await reclaimDead({ now });@@@ -426,7 +470,19 @@ async function tick({ now = clock() } = {}) {* queued too long ago fail as expired; every room with queued rows starts playing again. Another live process's* playing rows stay as they are.*/--async function recover({ now = clock(), graceMs = RESUME_GRACE_MS } = {}) {++function recover(opts) { return track(recoverOnce(opts)); }++async function recoverOnce({ now = clock(), graceMs = RESUME_GRACE_MS } = {}) {++ recovering = true; // before the first await: no pump of this process runs until the queue is settled and held++ try {++ return await settleAtBoot(now, graceMs);++ } finally {++ recovering = false;++ const asked = [...deferred];++ deferred.clear();++ for (const room of asked) pump(room).catch((err) => console.warn('[AudioQueue] pump:', err.message));++ }++}++async function settleAtBoot(now, graceMs) {const settled = await reclaimDead({ now, mine: true });const played = settled.played;const expired = (await db.run("UPDATE audio_requests SET state = 'failed', finished_at = ?, error = 'expired', actor = 'system' WHERE state = 'queued' AND created_at < ?", [now, now - STALE_MS])).changes;@@@ -442,6 +498,6 @@ async function recover({ now = clock(), graceMs = RESUME_GRACE_MS } = {}) {module.exports = {STATES, TRANSITIONS, KINDS, STALE_MS, GAP_MS, RESUME_GRACE_MS, RESUME_SETTLE_MS, MAX_ATTEMPTS, LEASE_MS, HEARTBEAT_MS,canTransition, roomKey, estimatePlayMs, mp3Bitrate,-- init, stop, enqueue, pump, claim, skip, clear, report, list, recover, tick, reclaimDead, roomJoined, getRequest, transition,++ init, stop, idle, enqueue, pump, claim, skip, clear, report, list, recover, tick, reclaimDead, roomJoined, getRequest, transition,get instanceId() { return instanceId; },};@@@ -2767,7 +2767,7 @@ class ChatServer {}}-- close() {++ async close() {if (this.wss) {if (this.heartbeatInterval) {clearInterval(this.heartbeatInterval);@@@ -2781,7 +2781,8 @@ class ChatServer {clearInterval(this._autoDeleteSweepInterval);this._autoDeleteSweepInterval = null;}-- audioQueue.stop();++ // The queue's work in flight (a clip being made or delivered) ends before the sockets and the database close.++ await audioQueue.stop();this.wss.clients.forEach(ws => ws.close());this.wss.close();}@@@ -3,10 +3,12 @@* The persisted TTS and sound queue (server/chat/audio-queue.js, roadmap Wave 6 D4): requests are* rows that go queued → playing → played, or skipped / failed; one plays at a time per room; the* broadcaster and moderators skip and clear (/skiptts, /cleartts, /api/tts/queue); failures are-- * recorded and the queue moves on; a keyed request is queued once; and a Chat restart (a real-- * process, SIGTERM) keeps the queue: what was playing is finished, what waited plays, in order,-- * once. The frames clients already know (tts-audio, soundboard-audio) are unchanged apart from a-- * new request_id. The restart boundary across processes runs on an injected clock and instance ids:++ * recorded and the queue moves on; a keyed request is queued once; and a Chat restart keeps the++ * queue: what was playing is finished, what waited plays, in order, once. The restart tests run on an++ * injected clock, timers, instance ids and performer gates (no sleeps): stop() waits for the work in++ * flight, and a pump asked for while recover() runs waits for it. The frames clients already know++ * (tts-audio, soundboard-audio) are unchanged apart from a new request_id. The restart boundary across++ * processes runs the same way:* a live owner's playing row is never touched by another process's recover(); a dead owner's row* is replayed once if it was never delivered, finished if it was.*/@@@ -18,7 +20,6 @@ const { boot, suite } = require('./helpers');const t = suite('audio-queue');let h, aq, tts, streamer, mod, viewer, listenerUser, chatters, channelId, streamId, listener;const sockets = [];--const realSynth = {};/** 8 kHz 8-bit mono WAV of `seconds`, base64 (1 s = 8000 bytes of audio). */function wav(seconds) {@@@ -61,8 +62,6 @@ t('boot', async () => {h = await boot();aq = require('../server/chat/audio-queue');tts = require('../server/chat/tts-engine');-- realSynth.synthesize = tts.synthesize;-- realSynth.synthesizeUserVoice = tts.synthesizeUserVoice;streamer = h.addUser('streamer', { role: 'streamer' });mod = h.addUser('moddy');viewer = h.addUser('viewer');@@@ -236,83 +235,167 @@ t('channel !sounds go through the same queue (announce at once, audio in turn)',await until(async () => (await aq.getRequest(audio.request_id)).state === 'played', 3000);});--// ── Restart: Chat as its own process ────────────────────────────────────────────────────────────--let chat, port;--// This process's handle on the database the spawned Chat serves from (test/helpers/pg-preload.mjs).--const raw = async (fn) => await fn(h.db.getDb());++// ── A restart, step by step: injected clock, timers, instance id and performer gates, no sleeps ─────────────++// The same instance id (a restarted process) lives three lives on one database; every wait is a hand-fired timer++// or a gate the test opens, so each ordering the old code left to chance is forced: a pump asked for while++// recover() runs, and stop() arriving while a clip is being made.t('restart: what was playing is finished, what waited plays in order once, stale requests expire', async () => {-- tts.synthesize = realSynth.synthesize;-- tts.synthesizeUserVoice = realSynth.synthesizeUserVoice;-- for (const w of sockets) { try { w.close(); } catch { /* */ } }-- await h.detach();-- // Long enough clips that the queue is still full when Chat stops.-- const file = path.join(process.env.SOUNDS_PATH, 'long-test.wav');-- fs.writeFileSync(file, wav(2));-- await raw(async (d) => {-- await d.prepare("INSERT INTO channel_sounds (channel_owner_id, command, url, mime, duration_seconds) VALUES (?, 'long', ?, 'audio/wav', 2)").run(streamer.id, file);-- // Left over from an earlier life: one queued an hour ago, one that crashed Chat three times.-- const ins = d.prepare("INSERT INTO audio_requests (room, stream_id, kind, state, label, payload, created_at, attempts) VALUES (?, ?, 'channel-sound', 'queued', ?, ?, ?, ?)");-- await ins.run(`stream:${streamId}`, streamId, 'stale', JSON.stringify({ file, title: '!stale' }), Date.now() - 3600e3, 0);-- await ins.run(`stream:${streamId}`, streamId, 'crashy', JSON.stringify({ file, title: '!crashy' }), Date.now(), 3);-- });-- port = await h.freePort();-- chat = await h.spawnChat({ port, env: { CHAT_AUDIO_INSTANCE_ID: 'chat-under-test' } });-- assert.strictEqual((await raw(async (d) => await d.prepare("SELECT state, error FROM audio_requests WHERE label = 'stale'").get())).error, 'expired');-- // The room's queue is held until it has listeners again (a second after the first rejoins).-- await h.sleep(300);-- assert.strictEqual(await raw(async (d) => (await d.prepare("SELECT state FROM audio_requests WHERE label = 'crashy'").get()).state), 'queued', 'held for listeners');-- const ears = await joined(listenerUser, '198.51.100.120');-- await until(async () => await raw(async (d) => (await d.prepare("SELECT state FROM audio_requests WHERE label = 'crashy'").get()).state) === 'failed');-- assert.strictEqual((await raw(async (d) => await d.prepare("SELECT error FROM audio_requests WHERE label = 'crashy'").get())).error, 'gave up after repeated attempts');++ const T = Date.now() + 2 * 24 * 3600e3; // ahead of the wall clock (and of the boundary test's day): nothing else's rows look stale or live++ const clock = { now: T };++ const plays = [];++ const room = 'stream:9701';++ const NAME = 'chat-under-test';++ const put = async (label, { attempts = 0, created = clock.now } = {}) => Number((await h.db.run(++ `INSERT INTO audio_requests (room, stream_id, kind, state, label, payload, created_at, attempts)++ VALUES (?, 9701, 'tts', 'queued', ?, '{}', ?, ?)`, [room, label, created, attempts])).lastInsertRowid);++ const row = async (id) => await h.db.get('SELECT state, attempts, claimed_by, started_at, error FROM audio_requests WHERE id = ?', [id]);++ const states = async (...ids) => await Promise.all(ids.map(async (id) => (await row(id)).state));-- for (const [i, c] of chatters.slice(0, 3).entries()) {-- const ws = await joined(c, `198.51.100.${121 + i}`);-- ws.sendJson({ type: 'chat', message: '!long' });-- await ws.next((m) => m.type === 'chat' && m.message_type === 'channel-sound');-- ws.close();-- }-- const firstClip = await ears.next((m) => m.type === 'soundboard-audio' && m.title === '!long');-- // The announce comes before the durable insert: wait for all three rows, not just the first clip.-- await until(async () => await raw(async (d) => (await d.prepare("SELECT COUNT(*) AS n FROM audio_requests WHERE label = '!long'").get()).n) === 3, 3000);-- const queued = await raw(async (d) => await d.prepare("SELECT id, state FROM audio_requests WHERE label = '!long' ORDER BY id").all());-- assert.deepStrictEqual(queued.map((x) => x.state), ['playing', 'queued', 'queued']);-- assert.strictEqual(queued[0].id, firstClip.request_id);++ // Timers by hand: advance(life, ms) fires what falls due in order, letting each one's work finish first.++ const arrivals = [];++ let blocked = 0;++ const settled = (q) => (blocked ? Promise.resolve() : Promise.race([q.idle(), new Promise((r) => arrivals.push(r))]));++ const gates = new Map();++ const gateOn = (id) => {++ const g = {};++ g.released = new Promise((r) => { g.release = r; });++ gates.set(id, g);++ return g;++ };++ const performer = async (r) => {++ const g = gates.get(r.id);++ if (g) { blocked++; arrivals.splice(0).forEach((f) => f()); await g.released; blocked--; }++ return { frame: { type: 'tts-audio', audio: '' }, durationMs: 2000 };++ };++ const life = () => {++ const pending = new Map();++ let seq = 0;++ const timers = {++ setTimeout: (fn, ms) => { const handle = { seq: ++seq }; pending.set(handle, { fn, due: clock.now + ms }); return handle; },++ clearTimeout: (handle) => { pending.delete(handle); },++ };++ return { pending, q: proc(NAME, clock, plays, { performer, timers }) };++ };++ const advance = async (l, ms) => {++ const target = clock.now + ms;++ for (;;) {++ await settled(l.q);++ const next = [...l.pending.entries()].filter(([, x]) => x.due <= target).sort((x, y) => x[1].due - y[1].due)[0];++ if (!next) break;++ l.pending.delete(next[0]);++ clock.now = Math.max(clock.now, next[1].due);++ next[1].fn();++ }++ clock.now = target;++ await settled(l.q);++ };-- const exit = await h.stopChat(chat);-- assert.deepStrictEqual(exit, { code: 0, signal: null }, chat.log);-- chat = await h.spawnChat({ port, env: { CHAT_AUDIO_INSTANCE_ID: 'chat-under-test' } });-- await h.sleep(300);-- assert.strictEqual(await raw(async (d) => (await d.prepare('SELECT state FROM audio_requests WHERE id = ?').get(queued[1].id)).state), 'queued', 'nothing plays into an empty room');-- const ears2 = await joined(listenerUser, '198.51.100.130');-- const got = [];-- got.push(await ears2.next((m) => m.type === 'soundboard-audio' && m.title === '!long', 4000));-- got.push(await ears2.next((m) => m.type === 'soundboard-audio' && m.title === '!long', 5000));-- assert.deepStrictEqual(got.map((m) => m.request_id), [queued[1].id, queued[2].id], 'the waiting clips, in order');-- assert.ok(await ears2.none((m) => m.type === 'soundboard-audio' && m.request_id === queued[0].id, 100), 'the clip delivered before the restart is not played again');-- const after = await raw(async (d) => await d.prepare("SELECT id, state, error FROM audio_requests WHERE label = '!long' ORDER BY id").all());-- assert.strictEqual(after[0].state, 'played');-- assert.strictEqual(after[0].error, 'delivered before a restart');-- assert.strictEqual(after[1].state, 'played');-- assert.strictEqual(after[2].state, 'playing');++ // Life 1. Left over from an earlier life: one queued an hour ago, one that crashed Chat three times.++ const stale = await put('stale', { created: T - 3600e3 });++ const crashy = await put('crashy', { attempts: 3 });++ const a = life();++ const boot1 = await a.q.recover();++ assert.ok(boot1.expired >= 1, JSON.stringify(boot1));++ assert.deepStrictEqual([(await row(stale)).state, (await row(stale)).error], ['failed', 'expired']);++ await advance(a, 300);++ assert.strictEqual((await row(crashy)).state, 'queued', 'held until the room has listeners');++ a.q.roomJoined(room);++ await advance(a, 1000);++ assert.deepStrictEqual([(await row(crashy)).state, (await row(crashy)).error], ['failed', 'gave up after repeated attempts']);-- // Moderation still works on the new process.-- const s = await joined(mod, '198.51.100.131');-- s.sendJson({ type: 'chat', message: '/cleartts' });-- await s.next((m) => m.type === 'system' && /^Cleared 1 TTS\/sound request\./.test(m.message));-- assert.strictEqual(await raw(async (d) => (await d.prepare('SELECT state FROM audio_requests WHERE id = ?').get(queued[2].id)).state), 'skipped');++ const c1 = await put('c1');++ const c2 = await put('c2');++ const c3 = await put('c3');++ await a.q.pump(room);++ assert.deepStrictEqual(await states(c1, c2, c3), ['playing', 'queued', 'queued']);++ assert.deepStrictEqual(plays, [{ by: NAME, id: c1 }]);++ // SIGTERM while c1 plays: stop() cancels its window; nothing of this life finishes it.++ await a.q.stop();++ assert.strictEqual(a.pending.size, 0, 'no timer survives stop()');++ clock.now += 5000;++ assert.deepStrictEqual(await states(c1, c2, c3), ['playing', 'queued', 'queued']);++++ // Life 2, same instance id. A pump asked for while recover() runs (a socket's request) waits for it: it must++ // not claim a row (its own claim would be requeued by recover, and c3 would play before c2).++ const b = life();++ const booting = b.q.recover();++ await b.q.pump(room);++ const boot2 = await booting;++ assert.strictEqual(boot2.played, 1, JSON.stringify(boot2));++ assert.deepStrictEqual(await states(c1, c2, c3), ['played', 'queued', 'queued']);++ assert.strictEqual((await row(c1)).error, 'delivered before a restart', 'delivered before the restart: finished, not replayed');++ assert.strictEqual((await row(c2)).claimed_by, null);++ assert.strictEqual((await row(c3)).attempts, 0);++ await advance(b, 300);++ assert.deepStrictEqual(await states(c2, c3), ['queued', 'queued'], 'nothing plays into an empty room');++ const gate3 = gateOn(c3);++ b.q.roomJoined(room);++ await advance(b, 999);++ assert.deepStrictEqual(await states(c2, c3), ['queued', 'queued'], 'a second after the socket rejoins');++ await advance(b, 1);++ assert.deepStrictEqual(await states(c2, c3), ['playing', 'queued']);++ assert.deepStrictEqual(plays.map((x) => x.id), [c1, c2]);++ // c2's window ends, c3 is claimed and being made: SIGTERM arrives now. stop() waits for the clip in++ // hand, which then goes nowhere: no frame, no timer, and the row is left for the next life.++ await advance(b, 2250);++ assert.deepStrictEqual(await states(c2, c3), ['played', 'playing']);++ assert.strictEqual((await row(c3)).started_at, null, 'c3 is still being made');++ const stopping = b.q.stop();++ assert.strictEqual(await Promise.race([stopping.then(() => 'stopped'), new Promise((r) => setImmediate(() => r('waiting')))]), 'waiting', 'stop() waits for the clip being made');++ gate3.release();++ await stopping;++ assert.strictEqual(b.pending.size, 0, 'no timer survives stop()');++ assert.deepStrictEqual(plays.map((x) => x.id), [c1, c2], 'the clip made after the stop is not delivered');++ assert.deepStrictEqual({ ...await row(c3) }, { state: 'playing', attempts: 1, claimed_by: NAME, started_at: null, error: null });++++ // Life 3, same instance id: c3 never went out, so it is queued again in its place and plays, once.++ const c = life();++ const boot3 = await c.q.recover();++ assert.strictEqual(boot3.requeued, 1, JSON.stringify(boot3));++ assert.deepStrictEqual(await states(c3), ['queued']);++ c.q.roomJoined(room);++ await advance(c, 1000);++ assert.deepStrictEqual(plays, [c1, c2, c3].map((id) => ({ by: NAME, id })), 'every waiting clip, in order, once');++ await advance(c, 2250);++ await advance(c, 10 * 60e3);++ assert.deepStrictEqual(await states(c1, c2, c3), ['played', 'played', 'played']);++ assert.strictEqual(plays.length, 3, 'nothing plays twice');++ assert.deepStrictEqual((await c.q.list(room)).queued, []);++ await c.q.stop();++});++++t('restart: a stop that arrives as a clip is delivered leaves no pacing timer, and the clip counts as delivered', async () => {++ const clock = { now: Date.now() + 3 * 24 * 3600e3 };++ const plays = [];++ const set = [];++ const timers = { setTimeout: (fn, ms) => { const handle = { fn, ms }; set.push(handle); return handle; }, clearTimeout: (handle) => { handle.cleared = true; } };++ let q = null;++ let stopping = null;++ q = proc('chat-stop-in-delivery', clock, plays, { timers, onDeliver: () => { stopping = q.stop(); } });++ const id = Number((await h.db.run(++ `INSERT INTO audio_requests (room, stream_id, kind, state, label, payload, created_at, attempts) VALUES ('stream:9702', 9702, 'tts', 'queued', 'solo', '{}', ?, 0)`, [clock.now])).lastInsertRowid);++ await q.pump('stream:9702');++ await stopping;++ assert.deepStrictEqual(plays, [{ by: 'chat-stop-in-delivery', id }]);++ assert.deepStrictEqual(set.filter((x) => !x.cleared), [], 'no live timer after stop()');++ const next = proc('chat-stop-in-delivery', clock, plays, { timers });++ const out = await next.recover();++ assert.strictEqual(out.played, 1, JSON.stringify(out));++ assert.strictEqual((await h.db.get('SELECT state, error FROM audio_requests WHERE id = ?', [id])).error, 'delivered before a restart');++ await next.stop();});// ── The restart boundary between processes: injected clock and instance ids, no sleeps ─────────────────────────const QUEUE = require.resolve('../server/chat/audio-queue');/** A fresh copy of the queue module standing for one Chat process (its own instance id, clock and rooms). */--function proc(name, clockRef, plays) {++function proc(name, clockRef, plays, { performer = null, timers = null, onDeliver = null } = {}) {delete require.cache[QUEUE];const q = require(QUEUE);delete require.cache[QUEUE];q.init({-- performers: { tts: async () => ({ frame: { type: 'tts-audio', audio: '' }, durationMs: 1 }) },-- deliver: (row, frame) => { if (frame.type === 'tts-audio') plays.push({ by: name, id: row.id }); },-- instanceId: name, clock: () => clockRef.now, heartbeat: false,++ performers: { tts: performer || (async () => ({ frame: { type: 'tts-audio', audio: '' }, durationMs: 1 })) },++ deliver: (row, frame) => { if (frame.type === 'tts-audio') plays.push({ by: name, id: row.id }); if (onDeliver) onDeliver(); },++ instanceId: name, clock: () => clockRef.now, heartbeat: false, ...(timers ? { timers } : {}),});return q;}@@@ -400,7 +400,7 @@ async function boot({ env = {} } = {}) {* a restart is a real SIGTERM and a new process. h.port/h.base follow the child.*/h.detach = async () => {-- try { h.chatServer.close(); } catch { /* */ }++ try { await h.chatServer.close(); } catch { /* */ }try { h.ctx.stop(); h.mirrorRelay.stop(); h.eventsRelay.stop(); } catch { /* */ }await new Promise((r) => h.server.close(() => r()));try { await h.db.close(); } catch { /* */ }@@@ -443,7 +443,7 @@ async function boot({ env = {} } = {}) {h.close = async () => {for (const c of h.children || []) { if (c.exitCode == null && c.signalCode == null) { c.kill('SIGKILL'); await c.exited; } }-- try { h.chatServer.close(); } catch { /* */ }++ try { await h.chatServer.close(); } catch { /* */ }try { h.ctx.stop(); h.mirrorRelay.stop(); h.eventsRelay.stop(); h.eventsConsumer.stop(); h.subscriptions.stop(); } catch { /* */ }await new Promise((r) => h.server.close(() => r()));network.close(); liveServer.close(); if (liveServer.closeAllConnections) liveServer.closeAllConnections();@@@ -208,6 +208,6 @@ t('global chat (no creator room) never asks VIP', async () => {});t.run(async () => {-- try { h.chatServer.close(); } catch { /* */ }++ try { await h.chatServer.close(); } catch { /* */ }vip.server.close();});
@@ -37,6 +37,16 @@
@@ -37,6 +37,16 @@
* as above, fails rows queued too long ago to still matter, and plays the rest once the room has
* as above, fails rows queued too long ago to still matter, and plays the rest once the room has
* listeners again: a second after the first socket rejoins it (clients come back 1–4 s after a
* listeners again: a second after the first socket rejoins it (clients come back 1–4 s after a
* restart notice), or after RESUME_GRACE_MS whatever happens.
* restart notice), or after RESUME_GRACE_MS whatever happens.
+ *
+ * Start and stop are ordered, so a restart is the same every time:
+ * - `recover()` closes the pump gate before its first await and opens it when the queue is settled and held:
+ * a pump asked for meanwhile (a socket's request, the heartbeat) is parked and runs after, so a fresh claim of
+ * this process is never read by recover() as a row left over from before the restart, and nothing plays out of order;
+ * - `stop()` is async: it cancels the pacing timers, then waits for the work in flight (a pump making a clip, a
+ * finish, a tick, a recover) before it resolves, and what that work does after the stop is limited to what a
+ * restart settles anyway: a clip made but not yet delivered is left claimed and undelivered (replayed once), a
+ * clip delivered is left with its started_at (played, 'delivered before a restart') and gets no pacing timer.
+ * The chat server awaits it before it closes its sockets and the database.
* A keyed request (a TTS of chat message m<id>) is queued once per room, so a retried bridge call
* A keyed request (a TTS of chat message m<id>) is queued once per room, so a retried bridge call
* or a double delivery never reads it twice, even across a restart.
* or a double delivery never reads it twice, even across a restart.
*
*
@@ -134,7 +144,7 @@ async function getRequest(id) { return await db.get('SELECT * FROM audio_request
@@ -134,7 +144,7 @@ async function getRequest(id) { return await db.get('SELECT * FROM audio_request
* Move one request along the state machine. Returns true when it moved (the row was in a state
* Move one request along the state machine. Returns true when it moved (the row was in a state
* that allows `to`); false when something else moved it first.
* that allows `to`); false when something else moved it first.
*/
*/
-async function transition(id, to, { error = null, actor = null, durationMs = null, now = Date.now() } = {}) {
+async function transition(id, to, { error = null, actor = null, durationMs = null, now = clock() } = {}) {
const from = Object.keys(TRANSITIONS).filter((s) => TRANSITIONS[s].includes(to));
const from = Object.keys(TRANSITIONS).filter((s) => TRANSITIONS[s].includes(to));
if (!from.length) return false;
if (!from.length) return false;
const final = to !== 'playing';
const final = to !== 'playing';
@@ -154,7 +164,11 @@ const performers = {}; // kind → async (row, payload) → { frame, durationM
@@ -154,7 +164,11 @@ const performers = {}; // kind → async (row, payload) → { frame, durationM
let deliver = null; // (row, frame) → void
let deliver = null; // (row, frame) → void
let instanceId = process.env.CHAT_AUDIO_INSTANCE_ID || `${os.hostname()}:${process.env.PORT || 4400}`;
let instanceId = process.env.CHAT_AUDIO_INSTANCE_ID || `${os.hostname()}:${process.env.PORT || 4400}`;
let clock = () => Date.now();
let clock = () => Date.now();
+let timers = { setTimeout, clearTimeout }; // the pacing, hold and deferral timers (tests drive them by hand)
let heartbeat = null;
let heartbeat = null;
+const inflight = new Set(); // pumps, finishes, ticks and recovers still running: stop() waits for them
+const deferred = new Set(); // rooms a pump was asked for while recover() ran: pumped when it is done
+let recovering = false;
const waiting = new Set(); // rooms whose claim lost to another process's playing row: pumped again by the heartbeat
const waiting = new Set(); // rooms whose claim lost to another process's playing row: pumped again by the heartbeat
const rooms = new Map(); // room → { busy, timer, playingId }
const rooms = new Map(); // room → { busy, timer, playingId }
let stopped = false;
let stopped = false;
@@ -166,32 +180,37 @@ function roomState(room) {
@@ -166,32 +180,37 @@ function roomState(room) {
/** Hold a room's queue until `until` (restart recovery), then play. A shorter hold replaces a longer one. */
/** Hold a room's queue until `until` (restart recovery), then play. A shorter hold replaces a longer one. */
function holdRoom(room, until) {
function holdRoom(room, until) {
+ if (stopped) return;
const st = roomState(room);
const st = roomState(room);
if (st.holdUntil && st.holdUntil <= until) return;
if (st.holdUntil && st.holdUntil <= until) return;
st.holdUntil = until;
st.holdUntil = until;
- if (st.holdTimer) clearTimeout(st.holdTimer);
- st.holdTimer = setTimeout(() => { st.holdUntil = 0; st.holdTimer = null; pump(room).catch((err) => console.warn('[AudioQueue] pump:', err.message)); }, Math.max(0, until - Date.now()));
+ if (st.holdTimer) timers.clearTimeout(st.holdTimer);
+ st.holdTimer = timers.setTimeout(() => { st.holdUntil = 0; st.holdTimer = null; pump(room).catch((err) => console.warn('[AudioQueue] pump:', err.message)); }, Math.max(0, until - clock()));
if (st.holdTimer.unref) st.holdTimer.unref();
if (st.holdTimer.unref) st.holdTimer.unref();
}
}
/** A socket joined a room: a queue held since a restart plays shortly. */
/** A socket joined a room: a queue held since a restart plays shortly. */
function roomJoined(room) {
function roomJoined(room) {
const st = room && rooms.get(room);
const st = room && rooms.get(room);
- if (st && st.holdUntil > Date.now()) holdRoom(room, Date.now() + RESUME_SETTLE_MS);
+ if (st && st.holdUntil > clock()) holdRoom(room, clock() + RESUME_SETTLE_MS);
}
}
/**
/**
* Wire the queue to the chat server: how each kind is made, and how a frame reaches a room.
* Wire the queue to the chat server: how each kind is made, and how a frame reaches a room.
* `performers[kind](row, payload)` resolves { frame, durationMs } or null (nothing to play).
* `performers[kind](row, payload)` resolves { frame, durationMs } or null (nothing to play).
* `instanceId` names this process's claims (default: host and port); `clock` (ms) is for tests.
* `instanceId` names this process's claims (default: host and port); `clock` (ms) is for tests.
- * `heartbeat: false` leaves lease renewal and the sweep to the caller (tests call tick()).
+ * `heartbeat: false` leaves lease renewal and the sweep to the caller (tests call tick()); `timers`
+ * ({ setTimeout, clearTimeout }) replaces the pacing, hold and deferral timers (tests fire them by hand).
*/
*/
-function init({ performers: p, deliver: d, instanceId: id = null, clock: c = null, heartbeat: beat = true }) {
+function init({ performers: p, deliver: d, instanceId: id = null, clock: c = null, timers: t = null, heartbeat: beat = true }) {
Object.assign(performers, p || {});
Object.assign(performers, p || {});
deliver = d;
deliver = d;
if (id) instanceId = String(id);
if (id) instanceId = String(id);
if (c) clock = c;
if (c) clock = c;
+ if (t) timers = { ...timers, ...t };
stopped = false;
stopped = false;
+ recovering = false;
+ deferred.clear();
if (heartbeat) clearInterval(heartbeat);
if (heartbeat) clearInterval(heartbeat);
heartbeat = null;
heartbeat = null;
if (beat) {
if (beat) {
@@ -200,17 +219,37 @@ function init({ performers: p, deliver: d, instanceId: id = null, clock: c = nul
@@ -200,17 +219,37 @@ function init({ performers: p, deliver: d, instanceId: id = null, clock: c = nul
}
}
}
}
-/** Stop pacing timers (shutdown). States stay as they are; recover() picks them up at boot. */
-function stop() {
+/** Run `work` as something stop() and idle() wait for. */
+function track(work) {
+ inflight.add(work);
+ const done = () => inflight.delete(work);
+ work.then(done, done);
+ return work;
+}
+/** Resolves once nothing is in flight (what a timer or a request started has finished). */
+async function idle() {
+ while (inflight.size) await Promise.allSettled([...inflight]);
+}
+
+/**
+ * Stop (shutdown): cancel the pacing timers, then wait (at most `graceMs`) for the work in flight, so nothing
+ * of this process's queue is still touching the database or the sockets once it resolves. States stay as they are;
+ * recover() picks them up at boot. Never rejects.
+ */
+async function stop({ graceMs = 3000 } = {}) {
stopped = true;
stopped = true;
if (heartbeat) clearInterval(heartbeat);
if (heartbeat) clearInterval(heartbeat);
heartbeat = null;
heartbeat = null;
waiting.clear();
waiting.clear();
+ deferred.clear();
for (const st of rooms.values()) {
for (const st of rooms.values()) {
- if (st.timer) clearTimeout(st.timer);
- if (st.holdTimer) clearTimeout(st.holdTimer);
+ if (st.timer) timers.clearTimeout(st.timer);
+ if (st.holdTimer) timers.clearTimeout(st.holdTimer);
st.timer = null; st.holdTimer = null;
st.timer = null; st.holdTimer = null;
}
}
+ let bound;
+ await Promise.race([idle(), new Promise((resolve) => { bound = setTimeout(resolve, graceMs); })]);
+ clearTimeout(bound);
rooms.clear();
rooms.clear();
}
}
@@ -243,7 +282,7 @@ async function enqueue({ kind, streamId = null, channelUserId = null, requestedB
@@ -243,7 +282,7 @@ async function enqueue({ kind, streamId = null, channelUserId = null, requestedB
if (!r.changes) return { queued: false, reason: 'duplicate' };
if (!r.changes) return { queued: false, reason: 'duplicate' };
return { queued: true, id: Number(r.lastInsertRowid), room };
return { queued: true, id: Number(r.lastInsertRowid), room };
});
});
- if (result.queued) setImmediate(() => { pump(result.room).catch((err) => console.warn('[AudioQueue] pump:', err.message)); });
+ if (result.queued) timers.setTimeout(() => { pump(result.room).catch((err) => console.warn('[AudioQueue] pump:', err.message)); }, 0);
return result;
return result;
}
}
@@ -278,17 +317,20 @@ async function claimOrBusy(id, now) {
@@ -278,17 +317,20 @@ async function claimOrBusy(id, now) {
}
}
}
}
-/** Play the next request of a room if none is playing. */
-async function pump(room) {
+/** Play the next request of a room if none is playing (not while recover() runs: the room is pumped when it is done). */
+function pump(room) { return track(pumpRoom(room)); }
+async function pumpRoom(room) {
if (stopped || !room) return;
if (stopped || !room) return;
+ if (recovering) { deferred.add(room); return; }
const st = roomState(room);
const st = roomState(room);
- if (st.busy || st.playingId || st.holdUntil > Date.now()) return;
+ if (st.busy || st.playingId || st.holdUntil > clock()) return;
st.busy = true;
st.busy = true;
try {
try {
for (;;) {
for (;;) {
if (stopped) return;
if (stopped) return;
let row;
let row;
try { row = await db.get("SELECT * FROM audio_requests WHERE room = ? AND state = 'queued' ORDER BY id LIMIT 1", [room]); } catch { return; }
try { row = await db.get("SELECT * FROM audio_requests WHERE room = ? AND state = 'queued' ORDER BY id LIMIT 1", [room]); } catch { return; }
+ if (stopped) return;
if (!row) { waiting.delete(room); return; }
if (!row) { waiting.delete(room); return; }
if (row.attempts >= MAX_ATTEMPTS) { await transition(row.id, 'failed', { error: 'gave up after repeated attempts', actor: 'system' }); continue; }
if (row.attempts >= MAX_ATTEMPTS) { await transition(row.id, 'failed', { error: 'gave up after repeated attempts', actor: 'system' }); continue; }
const { row: claimed, busy } = await claimOrBusy(row.id, clock());
const { row: claimed, busy } = await claimOrBusy(row.id, clock());
@@ -310,12 +352,12 @@ async function pump(room) {
@@ -310,12 +352,12 @@ async function pump(room) {
if (error) { await transition(row.id, 'failed', { error: String(error).slice(0, 300), actor: 'system' }); continue; }
if (error) { await transition(row.id, 'failed', { error: String(error).slice(0, 300), actor: 'system' }); continue; }
const durationMs = Math.round(Math.min(MAX_PLAY_MS, Math.max(MIN_PLAY_MS, Number(made.durationMs) || estimatePlayMs(made.frame))));
const durationMs = Math.round(Math.min(MAX_PLAY_MS, Math.max(MIN_PLAY_MS, Number(made.durationMs) || estimatePlayMs(made.frame))));
// Skipped or cleared while it was being made: nothing goes out.
// Skipped or cleared while it was being made: nothing goes out.
- const now = Date.now();
- const started = await db.run("UPDATE audio_requests SET started_at = ?, duration_ms = ? WHERE id = ? AND state = 'playing'", [now, durationMs, row.id]);
+ const started = await db.run("UPDATE audio_requests SET started_at = ?, duration_ms = ? WHERE id = ? AND state = 'playing'", [clock(), durationMs, row.id]);
if (!started.changes) continue;
if (!started.changes) continue;
try { if (deliver) deliver(row, { ...made.frame, request_id: row.id }); } catch (err) { console.warn('[AudioQueue] deliver:', err.message); }
try { if (deliver) deliver(row, { ...made.frame, request_id: row.id }); } catch (err) { console.warn('[AudioQueue] deliver:', err.message); }
+ if (stopped) return; // delivered (started_at is set: a restart counts it played); stop() cleared the timers, so no window to pace
st.playingId = row.id;
st.playingId = row.id;
- st.timer = setTimeout(() => { finish(room, row.id, 'played').catch((err) => console.warn('[AudioQueue] finish:', err.message)); }, durationMs + GAP_MS);
+ st.timer = timers.setTimeout(() => { finish(room, row.id, 'played').catch((err) => console.warn('[AudioQueue] finish:', err.message)); }, durationMs + GAP_MS);
if (st.timer.unref) st.timer.unref();
if (st.timer.unref) st.timer.unref();
return;
return;
}
}
@@ -323,15 +365,16 @@ async function pump(room) {
@@ -323,15 +365,16 @@ async function pump(room) {
}
}
/** End the playing request of a room (its window ended, it was skipped, or the client reported). */
/** End the playing request of a room (its window ended, it was skipped, or the client reported). */
-async function finish(room, id, to, opts = {}) {
+function finish(room, id, to, opts = {}) { return track(finishRoom(room, id, to, opts)); }
+async function finishRoom(room, id, to, opts) {
const st = roomState(room);
const st = roomState(room);
let moved = false;
let moved = false;
try { moved = await transition(id, to, opts); } catch { moved = false; }
try { moved = await transition(id, to, opts); } catch { moved = false; }
if (st.playingId === id) {
if (st.playingId === id) {
- if (st.timer) clearTimeout(st.timer);
+ if (st.timer) timers.clearTimeout(st.timer);
st.timer = null;
st.timer = null;
st.playingId = null;
st.playingId = null;
- setImmediate(() => { pump(room).catch((err) => console.warn('[AudioQueue] pump:', err.message)); });
+ if (!stopped) timers.setTimeout(() => { pump(room).catch((err) => console.warn('[AudioQueue] pump:', err.message)); }, 0);
}
}
return moved;
return moved;
}
}
@@ -411,7 +454,8 @@ async function reclaimDead({ now = clock(), mine = false } = {}) {
@@ -411,7 +454,8 @@ async function reclaimDead({ now = clock(), mine = false } = {}) {
* The heartbeat (every HEARTBEAT_MS, or by hand in tests): renew this process's leases, settle dead owners' rows
* The heartbeat (every HEARTBEAT_MS, or by hand in tests): renew this process's leases, settle dead owners' rows
* and pump the rooms that freed, and retry rooms whose claim lost to another process.
* and pump the rooms that freed, and retry rooms whose claim lost to another process.
*/
*/
-async function tick({ now = clock() } = {}) {
+function tick(opts) { return track(tickOnce(opts)); }
+async function tickOnce({ now = clock() } = {}) {
if (stopped) return { renewed: 0, rooms: [] };
if (stopped) return { renewed: 0, rooms: [] };
const renewed = (await db.run("UPDATE audio_requests SET lease_until = ? WHERE state = 'playing' AND claimed_by = ?", [now + LEASE_MS, instanceId])).changes;
const renewed = (await db.run("UPDATE audio_requests SET lease_until = ? WHERE state = 'playing' AND claimed_by = ?", [now + LEASE_MS, instanceId])).changes;
const swept = await reclaimDead({ now });
const swept = await reclaimDead({ now });
@@ -426,7 +470,19 @@ async function tick({ now = clock() } = {}) {
@@ -426,7 +470,19 @@ async function tick({ now = clock() } = {}) {
* queued too long ago fail as expired; every room with queued rows starts playing again. Another live process's
* queued too long ago fail as expired; every room with queued rows starts playing again. Another live process's
* playing rows stay as they are.
* playing rows stay as they are.
*/
*/
-async function recover({ now = clock(), graceMs = RESUME_GRACE_MS } = {}) {
+function recover(opts) { return track(recoverOnce(opts)); }
+async function recoverOnce({ now = clock(), graceMs = RESUME_GRACE_MS } = {}) {
+ recovering = true; // before the first await: no pump of this process runs until the queue is settled and held
+ try {
+ return await settleAtBoot(now, graceMs);
+ } finally {
+ recovering = false;
+ const asked = [...deferred];
+ deferred.clear();
+ for (const room of asked) pump(room).catch((err) => console.warn('[AudioQueue] pump:', err.message));
+ }
+}
+async function settleAtBoot(now, graceMs) {
const settled = await reclaimDead({ now, mine: true });
const settled = await reclaimDead({ now, mine: true });
const played = settled.played;
const played = settled.played;
const expired = (await db.run("UPDATE audio_requests SET state = 'failed', finished_at = ?, error = 'expired', actor = 'system' WHERE state = 'queued' AND created_at < ?", [now, now - STALE_MS])).changes;
const expired = (await db.run("UPDATE audio_requests SET state = 'failed', finished_at = ?, error = 'expired', actor = 'system' WHERE state = 'queued' AND created_at < ?", [now, now - STALE_MS])).changes;
@@ -442,6 +498,6 @@ async function recover({ now = clock(), graceMs = RESUME_GRACE_MS } = {}) {
@@ -442,6 +498,6 @@ async function recover({ now = clock(), graceMs = RESUME_GRACE_MS } = {}) {
module.exports = {
module.exports = {
STATES, TRANSITIONS, KINDS, STALE_MS, GAP_MS, RESUME_GRACE_MS, RESUME_SETTLE_MS, MAX_ATTEMPTS, LEASE_MS, HEARTBEAT_MS,
STATES, TRANSITIONS, KINDS, STALE_MS, GAP_MS, RESUME_GRACE_MS, RESUME_SETTLE_MS, MAX_ATTEMPTS, LEASE_MS, HEARTBEAT_MS,
canTransition, roomKey, estimatePlayMs, mp3Bitrate,
canTransition, roomKey, estimatePlayMs, mp3Bitrate,
- init, stop, enqueue, pump, claim, skip, clear, report, list, recover, tick, reclaimDead, roomJoined, getRequest, transition,
+ init, stop, idle, enqueue, pump, claim, skip, clear, report, list, recover, tick, reclaimDead, roomJoined, getRequest, transition,
get instanceId() { return instanceId; },
get instanceId() { return instanceId; },
};
};
@@ -2767,7 +2767,7 @@ class ChatServer {
@@ -2767,7 +2767,7 @@ class ChatServer {
}
}
}
}
- close() {
+ async close() {
if (this.wss) {
if (this.wss) {
if (this.heartbeatInterval) {
if (this.heartbeatInterval) {
clearInterval(this.heartbeatInterval);
clearInterval(this.heartbeatInterval);
@@ -2781,7 +2781,8 @@ class ChatServer {
@@ -2781,7 +2781,8 @@ class ChatServer {
clearInterval(this._autoDeleteSweepInterval);
clearInterval(this._autoDeleteSweepInterval);
this._autoDeleteSweepInterval = null;
this._autoDeleteSweepInterval = null;
}
}
- audioQueue.stop();
+ // The queue's work in flight (a clip being made or delivered) ends before the sockets and the database close.
+ await audioQueue.stop();
this.wss.clients.forEach(ws => ws.close());
this.wss.clients.forEach(ws => ws.close());
this.wss.close();
this.wss.close();
}
}
@@ -3,10 +3,12 @@
@@ -3,10 +3,12 @@
* The persisted TTS and sound queue (server/chat/audio-queue.js, roadmap Wave 6 D4): requests are
* The persisted TTS and sound queue (server/chat/audio-queue.js, roadmap Wave 6 D4): requests are
* rows that go queued → playing → played, or skipped / failed; one plays at a time per room; the
* rows that go queued → playing → played, or skipped / failed; one plays at a time per room; the
* broadcaster and moderators skip and clear (/skiptts, /cleartts, /api/tts/queue); failures are
* broadcaster and moderators skip and clear (/skiptts, /cleartts, /api/tts/queue); failures are
- * recorded and the queue moves on; a keyed request is queued once; and a Chat restart (a real
- * process, SIGTERM) keeps the queue: what was playing is finished, what waited plays, in order,
- * once. The frames clients already know (tts-audio, soundboard-audio) are unchanged apart from a
- * new request_id. The restart boundary across processes runs on an injected clock and instance ids:
+ * recorded and the queue moves on; a keyed request is queued once; and a Chat restart keeps the
+ * queue: what was playing is finished, what waited plays, in order, once. The restart tests run on an
+ * injected clock, timers, instance ids and performer gates (no sleeps): stop() waits for the work in
+ * flight, and a pump asked for while recover() runs waits for it. The frames clients already know
+ * (tts-audio, soundboard-audio) are unchanged apart from a new request_id. The restart boundary across
+ * processes runs the same way:
* a live owner's playing row is never touched by another process's recover(); a dead owner's row
* a live owner's playing row is never touched by another process's recover(); a dead owner's row
* is replayed once if it was never delivered, finished if it was.
* is replayed once if it was never delivered, finished if it was.
*/
*/
@@ -18,7 +20,6 @@ const { boot, suite } = require('./helpers');
@@ -18,7 +20,6 @@ const { boot, suite } = require('./helpers');
const t = suite('audio-queue');
const t = suite('audio-queue');
let h, aq, tts, streamer, mod, viewer, listenerUser, chatters, channelId, streamId, listener;
let h, aq, tts, streamer, mod, viewer, listenerUser, chatters, channelId, streamId, listener;
const sockets = [];
const sockets = [];
-const realSynth = {};
/** 8 kHz 8-bit mono WAV of `seconds`, base64 (1 s = 8000 bytes of audio). */
/** 8 kHz 8-bit mono WAV of `seconds`, base64 (1 s = 8000 bytes of audio). */
function wav(seconds) {
function wav(seconds) {
@@ -61,8 +62,6 @@ t('boot', async () => {
@@ -61,8 +62,6 @@ t('boot', async () => {
h = await boot();
h = await boot();
aq = require('../server/chat/audio-queue');
aq = require('../server/chat/audio-queue');
tts = require('../server/chat/tts-engine');
tts = require('../server/chat/tts-engine');
- realSynth.synthesize = tts.synthesize;
- realSynth.synthesizeUserVoice = tts.synthesizeUserVoice;
streamer = h.addUser('streamer', { role: 'streamer' });
streamer = h.addUser('streamer', { role: 'streamer' });
mod = h.addUser('moddy');
mod = h.addUser('moddy');
viewer = h.addUser('viewer');
viewer = h.addUser('viewer');
@@ -236,83 +235,167 @@ t('channel !sounds go through the same queue (announce at once, audio in turn)',
@@ -236,83 +235,167 @@ t('channel !sounds go through the same queue (announce at once, audio in turn)',
await until(async () => (await aq.getRequest(audio.request_id)).state === 'played', 3000);
await until(async () => (await aq.getRequest(audio.request_id)).state === 'played', 3000);
});
});
-// ── Restart: Chat as its own process ────────────────────────────────────────────────────────────
-let chat, port;
-// This process's handle on the database the spawned Chat serves from (test/helpers/pg-preload.mjs).
-const raw = async (fn) => await fn(h.db.getDb());
+// ── A restart, step by step: injected clock, timers, instance id and performer gates, no sleeps ─────────────
+// The same instance id (a restarted process) lives three lives on one database; every wait is a hand-fired timer
+// or a gate the test opens, so each ordering the old code left to chance is forced: a pump asked for while
+// recover() runs, and stop() arriving while a clip is being made.
t('restart: what was playing is finished, what waited plays in order once, stale requests expire', async () => {
t('restart: what was playing is finished, what waited plays in order once, stale requests expire', async () => {
- tts.synthesize = realSynth.synthesize;
- tts.synthesizeUserVoice = realSynth.synthesizeUserVoice;
- for (const w of sockets) { try { w.close(); } catch { /* */ } }
- await h.detach();
- // Long enough clips that the queue is still full when Chat stops.
- const file = path.join(process.env.SOUNDS_PATH, 'long-test.wav');
- fs.writeFileSync(file, wav(2));
- await raw(async (d) => {
- await d.prepare("INSERT INTO channel_sounds (channel_owner_id, command, url, mime, duration_seconds) VALUES (?, 'long', ?, 'audio/wav', 2)").run(streamer.id, file);
- // Left over from an earlier life: one queued an hour ago, one that crashed Chat three times.
- const ins = d.prepare("INSERT INTO audio_requests (room, stream_id, kind, state, label, payload, created_at, attempts) VALUES (?, ?, 'channel-sound', 'queued', ?, ?, ?, ?)");
- await ins.run(`stream:${streamId}`, streamId, 'stale', JSON.stringify({ file, title: '!stale' }), Date.now() - 3600e3, 0);
- await ins.run(`stream:${streamId}`, streamId, 'crashy', JSON.stringify({ file, title: '!crashy' }), Date.now(), 3);
- });
- port = await h.freePort();
- chat = await h.spawnChat({ port, env: { CHAT_AUDIO_INSTANCE_ID: 'chat-under-test' } });
- assert.strictEqual((await raw(async (d) => await d.prepare("SELECT state, error FROM audio_requests WHERE label = 'stale'").get())).error, 'expired');
- // The room's queue is held until it has listeners again (a second after the first rejoins).
- await h.sleep(300);
- assert.strictEqual(await raw(async (d) => (await d.prepare("SELECT state FROM audio_requests WHERE label = 'crashy'").get()).state), 'queued', 'held for listeners');
- const ears = await joined(listenerUser, '198.51.100.120');
- await until(async () => await raw(async (d) => (await d.prepare("SELECT state FROM audio_requests WHERE label = 'crashy'").get()).state) === 'failed');
- assert.strictEqual((await raw(async (d) => await d.prepare("SELECT error FROM audio_requests WHERE label = 'crashy'").get())).error, 'gave up after repeated attempts');
+ const T = Date.now() + 2 * 24 * 3600e3; // ahead of the wall clock (and of the boundary test's day): nothing else's rows look stale or live
+ const clock = { now: T };
+ const plays = [];
+ const room = 'stream:9701';
+ const NAME = 'chat-under-test';
+ const put = async (label, { attempts = 0, created = clock.now } = {}) => Number((await h.db.run(
+ `INSERT INTO audio_requests (room, stream_id, kind, state, label, payload, created_at, attempts)
+ VALUES (?, 9701, 'tts', 'queued', ?, '{}', ?, ?)`, [room, label, created, attempts])).lastInsertRowid);
+ const row = async (id) => await h.db.get('SELECT state, attempts, claimed_by, started_at, error FROM audio_requests WHERE id = ?', [id]);
+ const states = async (...ids) => await Promise.all(ids.map(async (id) => (await row(id)).state));
- for (const [i, c] of chatters.slice(0, 3).entries()) {
- const ws = await joined(c, `198.51.100.${121 + i}`);
- ws.sendJson({ type: 'chat', message: '!long' });
- await ws.next((m) => m.type === 'chat' && m.message_type === 'channel-sound');
- ws.close();
- }
- const firstClip = await ears.next((m) => m.type === 'soundboard-audio' && m.title === '!long');
- // The announce comes before the durable insert: wait for all three rows, not just the first clip.
- await until(async () => await raw(async (d) => (await d.prepare("SELECT COUNT(*) AS n FROM audio_requests WHERE label = '!long'").get()).n) === 3, 3000);
- const queued = await raw(async (d) => await d.prepare("SELECT id, state FROM audio_requests WHERE label = '!long' ORDER BY id").all());
- assert.deepStrictEqual(queued.map((x) => x.state), ['playing', 'queued', 'queued']);
- assert.strictEqual(queued[0].id, firstClip.request_id);
+ // Timers by hand: advance(life, ms) fires what falls due in order, letting each one's work finish first.
+ const arrivals = [];
+ let blocked = 0;
+ const settled = (q) => (blocked ? Promise.resolve() : Promise.race([q.idle(), new Promise((r) => arrivals.push(r))]));
+ const gates = new Map();
+ const gateOn = (id) => {
+ const g = {};
+ g.released = new Promise((r) => { g.release = r; });
+ gates.set(id, g);
+ return g;
+ };
+ const performer = async (r) => {
+ const g = gates.get(r.id);
+ if (g) { blocked++; arrivals.splice(0).forEach((f) => f()); await g.released; blocked--; }
+ return { frame: { type: 'tts-audio', audio: '' }, durationMs: 2000 };
+ };
+ const life = () => {
+ const pending = new Map();
+ let seq = 0;
+ const timers = {
+ setTimeout: (fn, ms) => { const handle = { seq: ++seq }; pending.set(handle, { fn, due: clock.now + ms }); return handle; },
+ clearTimeout: (handle) => { pending.delete(handle); },
+ };
+ return { pending, q: proc(NAME, clock, plays, { performer, timers }) };
+ };
+ const advance = async (l, ms) => {
+ const target = clock.now + ms;
+ for (;;) {
+ await settled(l.q);
+ const next = [...l.pending.entries()].filter(([, x]) => x.due <= target).sort((x, y) => x[1].due - y[1].due)[0];
+ if (!next) break;
+ l.pending.delete(next[0]);
+ clock.now = Math.max(clock.now, next[1].due);
+ next[1].fn();
+ }
+ clock.now = target;
+ await settled(l.q);
+ };
- const exit = await h.stopChat(chat);
- assert.deepStrictEqual(exit, { code: 0, signal: null }, chat.log);
- chat = await h.spawnChat({ port, env: { CHAT_AUDIO_INSTANCE_ID: 'chat-under-test' } });
- await h.sleep(300);
- assert.strictEqual(await raw(async (d) => (await d.prepare('SELECT state FROM audio_requests WHERE id = ?').get(queued[1].id)).state), 'queued', 'nothing plays into an empty room');
- const ears2 = await joined(listenerUser, '198.51.100.130');
- const got = [];
- got.push(await ears2.next((m) => m.type === 'soundboard-audio' && m.title === '!long', 4000));
- got.push(await ears2.next((m) => m.type === 'soundboard-audio' && m.title === '!long', 5000));
- assert.deepStrictEqual(got.map((m) => m.request_id), [queued[1].id, queued[2].id], 'the waiting clips, in order');
- assert.ok(await ears2.none((m) => m.type === 'soundboard-audio' && m.request_id === queued[0].id, 100), 'the clip delivered before the restart is not played again');
- const after = await raw(async (d) => await d.prepare("SELECT id, state, error FROM audio_requests WHERE label = '!long' ORDER BY id").all());
- assert.strictEqual(after[0].state, 'played');
- assert.strictEqual(after[0].error, 'delivered before a restart');
- assert.strictEqual(after[1].state, 'played');
- assert.strictEqual(after[2].state, 'playing');
+ // Life 1. Left over from an earlier life: one queued an hour ago, one that crashed Chat three times.
+ const stale = await put('stale', { created: T - 3600e3 });
+ const crashy = await put('crashy', { attempts: 3 });
+ const a = life();
+ const boot1 = await a.q.recover();
+ assert.ok(boot1.expired >= 1, JSON.stringify(boot1));
+ assert.deepStrictEqual([(await row(stale)).state, (await row(stale)).error], ['failed', 'expired']);
+ await advance(a, 300);
+ assert.strictEqual((await row(crashy)).state, 'queued', 'held until the room has listeners');
+ a.q.roomJoined(room);
+ await advance(a, 1000);
+ assert.deepStrictEqual([(await row(crashy)).state, (await row(crashy)).error], ['failed', 'gave up after repeated attempts']);
- // Moderation still works on the new process.
- const s = await joined(mod, '198.51.100.131');
- s.sendJson({ type: 'chat', message: '/cleartts' });
- await s.next((m) => m.type === 'system' && /^Cleared 1 TTS\/sound request\./.test(m.message));
- assert.strictEqual(await raw(async (d) => (await d.prepare('SELECT state FROM audio_requests WHERE id = ?').get(queued[2].id)).state), 'skipped');
+ const c1 = await put('c1');
+ const c2 = await put('c2');
+ const c3 = await put('c3');
+ await a.q.pump(room);
+ assert.deepStrictEqual(await states(c1, c2, c3), ['playing', 'queued', 'queued']);
+ assert.deepStrictEqual(plays, [{ by: NAME, id: c1 }]);
+ // SIGTERM while c1 plays: stop() cancels its window; nothing of this life finishes it.
+ await a.q.stop();
+ assert.strictEqual(a.pending.size, 0, 'no timer survives stop()');
+ clock.now += 5000;
+ assert.deepStrictEqual(await states(c1, c2, c3), ['playing', 'queued', 'queued']);
+
+ // Life 2, same instance id. A pump asked for while recover() runs (a socket's request) waits for it: it must
+ // not claim a row (its own claim would be requeued by recover, and c3 would play before c2).
+ const b = life();
+ const booting = b.q.recover();
+ await b.q.pump(room);
+ const boot2 = await booting;
+ assert.strictEqual(boot2.played, 1, JSON.stringify(boot2));
+ assert.deepStrictEqual(await states(c1, c2, c3), ['played', 'queued', 'queued']);
+ assert.strictEqual((await row(c1)).error, 'delivered before a restart', 'delivered before the restart: finished, not replayed');
+ assert.strictEqual((await row(c2)).claimed_by, null);
+ assert.strictEqual((await row(c3)).attempts, 0);
+ await advance(b, 300);
+ assert.deepStrictEqual(await states(c2, c3), ['queued', 'queued'], 'nothing plays into an empty room');
+ const gate3 = gateOn(c3);
+ b.q.roomJoined(room);
+ await advance(b, 999);
+ assert.deepStrictEqual(await states(c2, c3), ['queued', 'queued'], 'a second after the socket rejoins');
+ await advance(b, 1);
+ assert.deepStrictEqual(await states(c2, c3), ['playing', 'queued']);
+ assert.deepStrictEqual(plays.map((x) => x.id), [c1, c2]);
+ // c2's window ends, c3 is claimed and being made: SIGTERM arrives now. stop() waits for the clip in
+ // hand, which then goes nowhere: no frame, no timer, and the row is left for the next life.
+ await advance(b, 2250);
+ assert.deepStrictEqual(await states(c2, c3), ['played', 'playing']);
+ assert.strictEqual((await row(c3)).started_at, null, 'c3 is still being made');
+ const stopping = b.q.stop();
+ assert.strictEqual(await Promise.race([stopping.then(() => 'stopped'), new Promise((r) => setImmediate(() => r('waiting')))]), 'waiting', 'stop() waits for the clip being made');
+ gate3.release();
+ await stopping;
+ assert.strictEqual(b.pending.size, 0, 'no timer survives stop()');
+ assert.deepStrictEqual(plays.map((x) => x.id), [c1, c2], 'the clip made after the stop is not delivered');
+ assert.deepStrictEqual({ ...await row(c3) }, { state: 'playing', attempts: 1, claimed_by: NAME, started_at: null, error: null });
+
+ // Life 3, same instance id: c3 never went out, so it is queued again in its place and plays, once.
+ const c = life();
+ const boot3 = await c.q.recover();
+ assert.strictEqual(boot3.requeued, 1, JSON.stringify(boot3));
+ assert.deepStrictEqual(await states(c3), ['queued']);
+ c.q.roomJoined(room);
+ await advance(c, 1000);
+ assert.deepStrictEqual(plays, [c1, c2, c3].map((id) => ({ by: NAME, id })), 'every waiting clip, in order, once');
+ await advance(c, 2250);
+ await advance(c, 10 * 60e3);
+ assert.deepStrictEqual(await states(c1, c2, c3), ['played', 'played', 'played']);
+ assert.strictEqual(plays.length, 3, 'nothing plays twice');
+ assert.deepStrictEqual((await c.q.list(room)).queued, []);
+ await c.q.stop();
+});
+
+t('restart: a stop that arrives as a clip is delivered leaves no pacing timer, and the clip counts as delivered', async () => {
+ const clock = { now: Date.now() + 3 * 24 * 3600e3 };
+ const plays = [];
+ const set = [];
+ const timers = { setTimeout: (fn, ms) => { const handle = { fn, ms }; set.push(handle); return handle; }, clearTimeout: (handle) => { handle.cleared = true; } };
+ let q = null;
+ let stopping = null;
+ q = proc('chat-stop-in-delivery', clock, plays, { timers, onDeliver: () => { stopping = q.stop(); } });
+ const id = Number((await h.db.run(
+ `INSERT INTO audio_requests (room, stream_id, kind, state, label, payload, created_at, attempts) VALUES ('stream:9702', 9702, 'tts', 'queued', 'solo', '{}', ?, 0)`, [clock.now])).lastInsertRowid);
+ await q.pump('stream:9702');
+ await stopping;
+ assert.deepStrictEqual(plays, [{ by: 'chat-stop-in-delivery', id }]);
+ assert.deepStrictEqual(set.filter((x) => !x.cleared), [], 'no live timer after stop()');
+ const next = proc('chat-stop-in-delivery', clock, plays, { timers });
+ const out = await next.recover();
+ assert.strictEqual(out.played, 1, JSON.stringify(out));
+ assert.strictEqual((await h.db.get('SELECT state, error FROM audio_requests WHERE id = ?', [id])).error, 'delivered before a restart');
+ await next.stop();
});
});
// ── The restart boundary between processes: injected clock and instance ids, no sleeps ─────────────────────────
// ── The restart boundary between processes: injected clock and instance ids, no sleeps ─────────────────────────
const QUEUE = require.resolve('../server/chat/audio-queue');
const QUEUE = require.resolve('../server/chat/audio-queue');
/** A fresh copy of the queue module standing for one Chat process (its own instance id, clock and rooms). */
/** A fresh copy of the queue module standing for one Chat process (its own instance id, clock and rooms). */
-function proc(name, clockRef, plays) {
+function proc(name, clockRef, plays, { performer = null, timers = null, onDeliver = null } = {}) {
delete require.cache[QUEUE];
delete require.cache[QUEUE];
const q = require(QUEUE);
const q = require(QUEUE);
delete require.cache[QUEUE];
delete require.cache[QUEUE];
q.init({
q.init({
- performers: { tts: async () => ({ frame: { type: 'tts-audio', audio: '' }, durationMs: 1 }) },
- deliver: (row, frame) => { if (frame.type === 'tts-audio') plays.push({ by: name, id: row.id }); },
- instanceId: name, clock: () => clockRef.now, heartbeat: false,
+ performers: { tts: performer || (async () => ({ frame: { type: 'tts-audio', audio: '' }, durationMs: 1 })) },
+ deliver: (row, frame) => { if (frame.type === 'tts-audio') plays.push({ by: name, id: row.id }); if (onDeliver) onDeliver(); },
+ instanceId: name, clock: () => clockRef.now, heartbeat: false, ...(timers ? { timers } : {}),
});
});
return q;
return q;
}
}
@@ -400,7 +400,7 @@ async function boot({ env = {} } = {}) {
@@ -400,7 +400,7 @@ async function boot({ env = {} } = {}) {
* a restart is a real SIGTERM and a new process. h.port/h.base follow the child.
* a restart is a real SIGTERM and a new process. h.port/h.base follow the child.
*/
*/
h.detach = async () => {
h.detach = async () => {
- try { h.chatServer.close(); } catch { /* */ }
+ try { await h.chatServer.close(); } catch { /* */ }
try { h.ctx.stop(); h.mirrorRelay.stop(); h.eventsRelay.stop(); } catch { /* */ }
try { h.ctx.stop(); h.mirrorRelay.stop(); h.eventsRelay.stop(); } catch { /* */ }
await new Promise((r) => h.server.close(() => r()));
await new Promise((r) => h.server.close(() => r()));
try { await h.db.close(); } catch { /* */ }
try { await h.db.close(); } catch { /* */ }
@@ -443,7 +443,7 @@ async function boot({ env = {} } = {}) {
@@ -443,7 +443,7 @@ async function boot({ env = {} } = {}) {
h.close = async () => {
h.close = async () => {
for (const c of h.children || []) { if (c.exitCode == null && c.signalCode == null) { c.kill('SIGKILL'); await c.exited; } }
for (const c of h.children || []) { if (c.exitCode == null && c.signalCode == null) { c.kill('SIGKILL'); await c.exited; } }
- try { h.chatServer.close(); } catch { /* */ }
+ try { await h.chatServer.close(); } catch { /* */ }
try { h.ctx.stop(); h.mirrorRelay.stop(); h.eventsRelay.stop(); h.eventsConsumer.stop(); h.subscriptions.stop(); } catch { /* */ }
try { h.ctx.stop(); h.mirrorRelay.stop(); h.eventsRelay.stop(); h.eventsConsumer.stop(); h.subscriptions.stop(); } catch { /* */ }
await new Promise((r) => h.server.close(() => r()));
await new Promise((r) => h.server.close(() => r()));
network.close(); liveServer.close(); if (liveServer.closeAllConnections) liveServer.closeAllConnections();
network.close(); liveServer.close(); if (liveServer.closeAllConnections) liveServer.closeAllConnections();
@@ -208,6 +208,6 @@ t('global chat (no creator room) never asks VIP', async () => {
@@ -208,6 +208,6 @@ t('global chat (no creator room) never asks VIP', async () => {
});
});
t.run(async () => {
t.run(async () => {
- try { h.chatServer.close(); } catch { /* */ }
+ try { await h.chatServer.close(); } catch { /* */ }
vip.server.close();
vip.server.close();
});
});