openvibers/openvibe.chat Commits
Serialize audio claims per room and lease them to their process (#2)
Changes
@@@ -31,10 +31,17 @@ before changing behaviour.** Browser JavaScript does not change; nginx routes th`/api/tts/*`), 101soundboards and channel sounds (`/api/sounds*`), moderation utils and the wordfilter, the deploy-notice card. Changes are confined to where they touched Live's data.- **The TTS and sound queue** (`audio-queue.js`, table `audio_requests`). TTS, channel `!sounds` and-- 101soundboards clips are persisted requests played one at a time per room — `queued → playing →-- played`, or `skipped` / `failed` — so the queue survives a Chat restart (what was playing is-- finished, what waited plays once the room has listeners again, requests older than 5 minutes-- expire) and a keyed request (the TTS of message `m<id>`) is read once. The broadcaster and++ 101soundboards clips are persisted requests played one at a time per room, across every Chat++ process (a claim holds the room's advisory lock; a partial unique index allows one `playing` row++ per room) — `queued → playing → played`, or `skipped` / `failed` — so the queue survives a Chat++ restart (what waited plays once the room has listeners again, requests older than 5 minutes++ expire) and a keyed request (the TTS of message `m<id>`) is read once. A playing row is owned:++ `claimed_by` (the process's instance id, host and port or `CHAT_AUDIO_INSTANCE_ID`) and a++ 30-second `lease_until` its owner renews every 10 s. Only a dead owner's row is settled — its++ lease passed, or it is the restarted process's own — at boot and by the heartbeat's sweep; a live++ process's clip is never touched. Such a row is `played` if its delivery had started, otherwise++ queued again in its old place, once: claimed a second time and dead again undelivered, it++ `failed` (`attempts` counts claims), so a clip that crashes Chat cannot loop. The broadcaster andmoderators control it: `/skiptts [id]`, `/cleartts`, and `GET /api/tts/queue`,`POST /api/tts/queue/skip|clear`, `POST /api/tts/queue/:id/report` (the playing client's`played`/`failed`). Browsers get the same `tts-audio` / `soundboard-audio` frames as before, now@@@ -88,6 +88,14 @@ The gaps the script found on 2026-09-25, with the product decisions taken for ththe channel room and the global feed too. Cursor reads name them in `deleted_ids`.- **`/clear` clears screens** (Twitch semantics) and now says so to the moderator: "Chat cleared onscreen; messages stay in history — use purge to remove them."++- **One TTS/sound clip per room across Chat processes** (2026-10-01). Two processes could each claim a++ different queued request of one room and play both at once, and a booting process finished every++ `playing` row, other live processes' included, dropping a clip whose owner died before delivering++ it. A claim now holds the room's advisory lock and takes nothing while the room plays (partial++ unique index `idx_audio_requests_one_playing`, `migrations/0002_audio_claim_owner.sql`); a playing++ row carries its owner (`claimed_by`) and a lease (`lease_until`, renewed by the owner), and only a++ dead owner's row is settled: `played` if its delivery had started, else replayed once in its old++ place, then `failed` (see README, the TTS and sound queue).## Fixed while building itnnew file mode 100644@@@ -0,0 +1,22 @@++-- phase: expand++-- The audio queue's claims are owned and one per room (server/chat/audio-queue.js):++-- claimed_by the Chat process (its instance id) that claimed the row and plays it;++-- lease_until ms; that process renews it while the row is playing. A playing row whose lease has passed++-- belongs to a process that died: recover() and the sweeper settle it, never a live owner's.++-- At most one playing row per room, across processes: the partial unique index is the store's guarantee++-- (claim() also takes a per-room advisory lock, so a losing claimer just gets nothing).++-- Safe on the live table and idempotent: nullable columns, no rewrite. The index is built in this transaction++-- under a short SHARE ROW EXCLUSIVE lock (the table holds about a week of requests), so the duplicates it++-- would refuse can be settled first and no claim of a previous-release process slips in between; CREATE INDEX++-- CONCURRENTLY could not run next to that clean-up, and a failed concurrent build leaves an invalid index that++-- IF NOT EXISTS would then keep.++ALTER TABLE audio_requests ADD COLUMN IF NOT EXISTS claimed_by text COLLATE "C";++ALTER TABLE audio_requests ADD COLUMN IF NOT EXISTS lease_until bigint;++++LOCK TABLE audio_requests IN SHARE ROW EXCLUSIVE MODE;++-- Two clips playing at once in one room (the race this closes): keep the oldest, the others reached the room too.++UPDATE audio_requests a SET state = 'played', error = 'delivered before a restart',++ finished_at = COALESCE(a.finished_at, a.started_at, a.created_at)++ WHERE a.state = 'playing'++ AND EXISTS (SELECT 1 FROM audio_requests b WHERE b.room = a.room AND b.state = 'playing' AND b.id < a.id);++CREATE UNIQUE INDEX IF NOT EXISTS idx_audio_requests_one_playing ON audio_requests(room) WHERE state = 'playing';@@@ -16,10 +16,27 @@* window ends it is played and the next one goes out.* played, skipped, failed final.*-- * A Chat restart keeps the queue: `recover()` at boot finishes rows that were playing (they reached-- * the room before the restart), 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.++ * One clip per room across every Chat process: `claim()` takes the room's advisory lock in its++ * transaction and takes nothing while the room has a playing row (the partial unique index++ * idx_audio_requests_one_playing, migrations/0002, is the store's guarantee). A process that loses++ * gets nothing; the room's next pump (the owner's, when its clip ends, or the heartbeat) plays it.++ *++ * Claims are owned: a playing row records its process (`claimed_by`, the instance id: host and port,++ * or CHAT_AUDIO_INSTANCE_ID) and a lease (`lease_until`) the owner renews every HEARTBEAT_MS while it++ * lives. Only a row whose owner is dead is ever settled by another hand — its lease has passed, or it++ * is this process's own row from before a restart (same instance id) — by `recover()` at boot and by++ * the heartbeat's sweep. Such a row:++ * - was never delivered (no started_at: its owner died while making it): it goes back to queued,++ * in its old place (the queue plays by id), once; `attempts` counts claims, so a row claimed++ * MAX_ATTEMPTS (2) times that dies again undelivered fails ('not delivered: its owner stopped++ * twice') and a clip that crashes Chat cannot loop;++ * - was delivered (started_at set): it is played ('delivered before a restart'), never replayed.++ * A row of a live owner, on any process, is never touched.++ *++ * A Chat restart keeps the queue: `recover()` at boot settles its own and dead owners' playing rows++ * 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.* 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.*@@@ -31,6 +48,7 @@*/'use strict';++const os = require('os');const db = require('../db/database');const STATES = ['queued', 'playing', 'played', 'skipped', 'failed'];@@@ -47,7 +65,10 @@ const GAP_MS = 250; // between two clipsconst MIN_PLAY_MS = 500;const MAX_PLAY_MS = 60 * 1000;const STALE_MS = 5 * 60 * 1000; // a request still queued this long after it was made is dropped--const MAX_ATTEMPTS = 3; // a request whose making crashed Chat this often fails++const MAX_ATTEMPTS = 2; // claims of one request: its owner may die undelivered once (one replay)++const LEASE_MS = 30 * 1000; // a playing row's owner is dead once its lease is this old…++const HEARTBEAT_MS = 10 * 1000; // …and a live owner renews it this often (and sweeps dead owners' rows)++const LOCK_SPACE = 71_440_216; // pg_advisory_xact_lock(LOCK_SPACE, hashtext(room)): one claim per room at a timeconst KEEP_MS = 7 * 24 * 3600 * 1000;const RESUME_GRACE_MS = 8000; // after a restart, a room with waiting requests resumes by then…const RESUME_SETTLE_MS = 1000; // …or this long after a socket rejoins it@@@ -131,6 +152,10 @@ async function transition(id, to, { error = null, actor = null, durationMs = nul// ── Runtime ─────────────────────────────────────────────────────────────────────────────────────const performers = {}; // kind → async (row, payload) → { frame, durationMs } | nulllet deliver = null; // (row, frame) → void++let instanceId = process.env.CHAT_AUDIO_INSTANCE_ID || `${os.hostname()}:${process.env.PORT || 4400}`;++let clock = () => Date.now();++let heartbeat = null;++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;@@@ -158,16 +183,29 @@ function roomJoined(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).++ * `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()).*/--function init({ performers: p, deliver: d }) {++function init({ performers: p, deliver: d, instanceId: id = null, clock: c = null, heartbeat: beat = true }) {Object.assign(performers, p || {});deliver = d;++ if (id) instanceId = String(id);++ if (c) clock = c;stopped = false;++ if (heartbeat) clearInterval(heartbeat);++ heartbeat = null;++ if (beat) {++ heartbeat = setInterval(() => { tick().catch((err) => console.warn('[AudioQueue] heartbeat:', err.message)); }, HEARTBEAT_MS);++ if (heartbeat.unref) heartbeat.unref();++ }}/** Stop pacing timers (shutdown). States stay as they are; recover() picks them up at boot. */function stop() {stopped = true;++ if (heartbeat) clearInterval(heartbeat);++ heartbeat = null;++ waiting.clear();for (const st of rooms.values()) {if (st.timer) clearTimeout(st.timer);if (st.holdTimer) clearTimeout(st.holdTimer);@@@ -210,11 +248,34 @@ async function enqueue({ kind, streamId = null, channelUserId = null, requestedB}/**-- * Take a queued request to play (plan T3 decision 3): one atomic UPDATE, so of two pumps (two processes) at most-- * one gets the row back; null = someone else took it, or it was skipped. Counts the attempt.++ * Take a queued request to play (plan T3 decision 3), owned by this process under a lease. One transaction holding++ * the room's advisory lock: nothing while the room has a playing row (any process's), else one atomic UPDATE, so of++ * two pumps (two processes) at most one gets a row of a room back. Counts the attempt.++ * → the row, or null with `busy` set when the room is playing (null without it: someone took or skipped this row).*/--async function claim(id) {-- return await db.get("UPDATE audio_requests SET state = 'playing', attempts = attempts + 1 WHERE id = ? AND state = 'queued' RETURNING *", [id]) || null;++async function claim(id, { now = clock() } = {}) {++ const out = await claimOrBusy(id, now);++ return out.row;++}++async function claimOrBusy(id, now) {++ try {++ return await db.tx(async () => {++ const q = await db.get("SELECT room FROM audio_requests WHERE id = ? AND state = 'queued'", [id]);++ if (!q) return { row: null, busy: false };++ await db.get('SELECT pg_advisory_xact_lock(?, hashtext(?)) AS locked', [LOCK_SPACE, q.room]);++ if (await db.get("SELECT 1 AS one FROM audio_requests WHERE room = ? AND state = 'playing'", [q.room])) return { row: null, busy: true };++ const row = await db.get(++ `UPDATE audio_requests SET state = 'playing', attempts = attempts + 1, claimed_by = ?, lease_until = ?++ WHERE id = ? AND state = 'queued' RETURNING *`,++ [instanceId, now + LEASE_MS, id],++ );++ return { row: row || null, busy: false };++ });++ } catch (err) {++ // The one-playing-row index refused it: a claimer that does not take the lock (a previous release) won.++ if (err && err.code === '23505') return { row: null, busy: true };++ throw err;++ }}/** Play the next request of a room if none is playing. */@@@ -228,10 +289,12 @@ async function pump(room) {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 (!row) 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 claimed = await claim(row.id);++ const { row: claimed, busy } = await claimOrBusy(row.id, clock());++ if (busy) { waiting.add(room); return; } // another process plays this room: its finish (or the heartbeat) pumps againif (!claimed) continue; // another pump took it (or it was skipped): the next one++ waiting.delete(room);row = claimed;// The audio is made outside any transaction (a synth takes seconds); the row stays 'playing' meanwhile.st.makingId = row.id;@@@ -326,21 +389,59 @@ async function list(room, { recent = 20 } = {}) {}/**-- * Boot: rows playing when Chat stopped were delivered, so they are played; rows queued too long-- * ago fail as expired; every room with queued rows starts playing again.++ * Settle the playing rows of dead owners: a lease that has passed (or none: a previous release's claim), and with++ * `mine` this instance's own rows (at boot they are from before the restart). Delivered → played; undelivered →++ * queued again in its old place, or failed once it has been claimed MAX_ATTEMPTS times. A live owner's row is++ * never touched. → { played, requeued, failed, rooms }*/--async function recover({ now = Date.now(), graceMs = RESUME_GRACE_MS } = {}) {-- const played = (await db.run("UPDATE audio_requests SET state = 'played', finished_at = ?, error = 'delivered before a restart' WHERE state = 'playing'", [now])).changes;++async function reclaimDead({ now = clock(), mine = false } = {}) {++ const dead = `state = 'playing' AND (lease_until IS NULL OR lease_until < ?${mine ? ' OR claimed_by = ?' : ''})`;++ const args = mine ? [now, instanceId] : [now];++ const played = await db.all(`UPDATE audio_requests SET state = 'played', finished_at = ?, error = 'delivered before a restart'++ WHERE ${dead} AND started_at IS NOT NULL RETURNING room`, [now, ...args]);++ const failed = await db.all(`UPDATE audio_requests SET state = 'failed', finished_at = ?, error = 'not delivered: its owner stopped twice', actor = 'system'++ WHERE ${dead} AND started_at IS NULL AND attempts >= ? RETURNING room`, [now, ...args, MAX_ATTEMPTS]);++ const requeued = await db.all(`UPDATE audio_requests SET state = 'queued', claimed_by = NULL, lease_until = NULL, duration_ms = NULL++ WHERE ${dead} AND started_at IS NULL AND attempts < ? RETURNING room`, [...args, MAX_ATTEMPTS]);++ const rooms = [...new Set([...played, ...failed, ...requeued].map((r) => r.room))];++ return { played: played.length, requeued: requeued.length, failed: failed.length, rooms };++}++++/**++ * 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() } = {}) {++ 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 });++ if (swept.rooms.length) console.log(`[AudioQueue] swept dead owners' rows: ${swept.played} finished, ${swept.requeued} requeued, ${swept.failed} failed`);++ const rooms = [...new Set([...swept.rooms, ...waiting])];++ for (const room of rooms) await pump(room).catch((err) => console.warn('[AudioQueue] pump:', err.message));++ return { renewed, ...swept, rooms };++}++++/**++ * Boot: this instance's rows from before the restart and dead owners' rows are settled (reclaimDead); rows++ * 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 } = {}) {++ 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;await db.run("DELETE FROM audio_requests WHERE state IN ('played', 'skipped', 'failed') AND COALESCE(finished_at, created_at) < ?", [now - KEEP_MS]);const pending = (await db.all("SELECT DISTINCT room FROM audio_requests WHERE state = 'queued'")).map((r) => r.room);for (const room of pending) holdRoom(room, now + graceMs);-- if (played || expired || pending.length) console.log(`[AudioQueue] recovered: ${played} finished, ${expired} expired, ${pending.length} room(s) resuming`);-- return { played, expired, rooms: pending };++ if (played || settled.requeued || settled.failed || expired || pending.length) {++ console.log(`[AudioQueue] recovered: ${played} finished, ${settled.requeued} requeued, ${settled.failed} failed, ${expired} expired, ${pending.length} room(s) resuming`);++ }++ return { played, requeued: settled.requeued, failed: settled.failed, expired, rooms: pending };}module.exports = {-- STATES, TRANSITIONS, KINDS, STALE_MS, GAP_MS, RESUME_GRACE_MS, RESUME_SETTLE_MS,++ 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, roomJoined, getRequest, transition,++ init, stop, enqueue, pump, claim, skip, clear, report, list, recover, tick, reclaimDead, roomJoined, getRequest, transition,++ get instanceId() { return instanceId; },};@@@ -4,7 +4,9 @@* each with its own pool connections on one database — race over the same queued requests. The claim is one atomic* UPDATE … WHERE state = 'queued' RETURNING *, so every request plays exactly once; the synth runs outside any* transaction. And enqueue's dedupe is the unique (room, dedupe_key) index: concurrent enqueues of one key store-- * one row. On PostgreSQL (npm run test:pg) the two pumps really run on separate connections.++ * one row. And a room plays one request at a time across processes: claim() holds the room's advisory lock and++ * takes nothing while the room has a playing row (backed by the one-playing-row partial unique index).++ * On PostgreSQL (npm run test:pg) the two pumps really run on separate connections.*/const assert = require('assert');const { suite } = require('./helpers');@@@ -26,6 +28,7 @@ function instance(name, plays) {},},deliver: (row, frame) => { if (frame.type === 'tts-audio') plays.push({ by: name, id: row.id }); },++ instanceId: `proc-${name}`, heartbeat: false,});return aq;}@@@ -71,6 +74,51 @@ t('a request claimed by one pump is not claimed again', async () => {assert.strictEqual((await db.get('SELECT attempts FROM audio_requests WHERE id = ?', [r.id])).attempts, 1);});++t('two requests in one room, two claimers at once: one plays, the other gets nothing until it ends', async () => {++ const ids = [];++ for (const label of ['first', 'second']) {++ const r = await db.run(`INSERT INTO audio_requests (room, stream_id, kind, state, label, payload, created_at)++ VALUES ('stream:9300', 9300, 'tts', 'queued', ?, '{}', ?)`, [label, Date.now()]);++ ids.push(Number(r.lastInsertRowid));++ }++ for (let round = 0; round < 5; round++) {++ // Each claimer goes for a different request of the room at the same time.++ const [x, y] = await Promise.all([a.claim(ids[0]), b.claim(ids[1])]);++ assert.strictEqual([x, y].filter(Boolean).length, 1, `round ${round}: exactly one claim wins`);++ const won = x || y;++ const loser = x ? b : a;++ const other = ids.find((id) => id !== won.id);++ assert.strictEqual(won.claimed_by, x ? 'proc-a' : 'proc-b');++ assert.ok(won.lease_until > Date.now());++ assert.strictEqual((await db.get("SELECT COUNT(*) AS n FROM audio_requests WHERE room = 'stream:9300' AND state = 'playing'")).n, 1);++ // While it plays, the loser gets nothing, whichever request it tries.++ assert.strictEqual(await loser.claim(other), null);++ assert.strictEqual(await a.claim(other), null);++ assert.strictEqual((await db.get('SELECT state, attempts FROM audio_requests WHERE id = ?', [other])).state, 'queued');++ // It ends: the loser's next try takes the other one.++ assert.ok(await a.transition(won.id, 'played'));++ const next = await loser.claim(other);++ assert.ok(next && next.id === other && next.state === 'playing');++ assert.ok(await a.transition(other, 'played'));++ // Again with two fresh requests.++ ids.length = 0;++ for (const label of ['first', 'second']) {++ const r = await db.run(`INSERT INTO audio_requests (room, stream_id, kind, state, label, payload, created_at)++ VALUES ('stream:9300', 9300, 'tts', 'queued', ?, '{}', ?)`, [`${label}${round}`, Date.now()]);++ ids.push(Number(r.lastInsertRowid));++ }++ }++});++++t('the store refuses a second playing row in a room, whoever writes it', async () => {++ const r = await db.run(`INSERT INTO audio_requests (room, stream_id, kind, state, label, payload, created_at)++ VALUES ('stream:9300', 9300, 'tts', 'queued', 'raw', '{}', ?)`, [Date.now()]);++ const playing = await a.claim(Number(r.lastInsertRowid) - 2) || await a.claim(Number(r.lastInsertRowid) - 1);++ assert.ok(playing, 'one of the earlier requests is playing');++ // A claimer without the lock (the previous release's bare UPDATE) is refused by the partial unique index.++ await assert.rejects(db.run("UPDATE audio_requests SET state = 'playing' WHERE id = ?", [Number(r.lastInsertRowid)]), (err) => err.code === '23505');++});++t('concurrent enqueues of one dedupe key store one row', async () => {const outs = await Promise.all([a, b, a, b].map((q) => q.enqueue({ kind: 'tts', streamId: 9200, label: 'dup', payload: {}, dedupeKey: 'k1' })));assert.strictEqual(outs.filter((o) => o.queued).length, 1, JSON.stringify(outs));@@@ -6,7 +6,9 @@* 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.++ * new request_id. The restart boundary across processes runs on an injected clock and instance ids:++ * 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.*/const assert = require('assert');const fs = require('fs');@@@ -254,7 +256,7 @@ t('restart: what was playing is finished, what waited plays in order once, staleawait ins.run(`stream:${streamId}`, streamId, 'crashy', JSON.stringify({ file, title: '!crashy' }), Date.now(), 3);});port = await h.freePort();-- chat = await h.spawnChat({ port });++ 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);@@@ -278,7 +280,7 @@ t('restart: what was playing is finished, what waited plays in order once, staleconst exit = await h.stopChat(chat);assert.deepStrictEqual(exit, { code: 0, signal: null }, chat.log);-- chat = await h.spawnChat({ port });++ 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');@@@ -300,6 +302,88 @@ t('restart: what was playing is finished, what waited plays in order once, staleassert.strictEqual(await raw(async (d) => (await d.prepare('SELECT state FROM audio_requests WHERE id = ?').get(queued[2].id)).state), 'skipped');});++// ── 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) {++ 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,++ });++ return q;++}++t('restart boundary: live owners are left alone, a dead owner\'s clip is replayed once if undelivered, finished if delivered', async () => {++ // A day ahead of the wall clock: nothing here can look expired to the queues running on real time in this run.++ const T = Date.now() + 24 * 3600e3;++ const clock = { now: T };++ const plays = [];++ const LEASE = aq.LEASE_MS;++ const put = async (room, label, f) => Number((await h.db.run(++ `INSERT INTO audio_requests (room, stream_id, kind, state, label, payload, created_at, attempts, claimed_by, lease_until, started_at)++ VALUES (?, ?, 'tts', ?, ?, '{}', ?, ?, ?, ?, ?)`,++ [room, Number(room.split(':')[1]), f.state, label, T - 1000, f.attempts || 0, f.owner || null, f.lease || null, f.started || null],++ )).lastInsertRowid);++ const row = async (id) => await h.db.get('SELECT state, attempts, claimed_by, lease_until, error FROM audio_requests WHERE id = ?', [id]);++++ // stream:9601 — process P is alive and playing; stream:9602 — a dead owner's clip still being made;++ // stream:9603 — a dead owner's clip already delivered.++ const live = await put('stream:9601', 'live', { state: 'playing', attempts: 1, owner: 'proc-p', lease: T + LEASE, started: T - 500 });++ const liveNext = await put('stream:9601', 'live-next', { state: 'queued' });++ const undelivered = await put('stream:9602', 'undelivered', { state: 'playing', attempts: 1, owner: 'proc-dead', lease: T - 1 });++ const behind = await put('stream:9602', 'behind', { state: 'queued' });++ const delivered = await put('stream:9603', 'delivered', { state: 'playing', attempts: 1, owner: 'proc-dead', lease: T - 1, started: T - 3000 });++++ const p = proc('proc-p', clock, plays);++ const q = proc('proc-q', clock, plays);++ const out = await q.recover({ now: T }); // Q boots while P lives++ assert.ok(out.requeued >= 1 && out.played >= 1, JSON.stringify(out));++ assert.deepStrictEqual(await row(live), { state: 'playing', attempts: 1, claimed_by: 'proc-p', lease_until: T + LEASE, error: null }, 'a live owner\'s row is untouched');++ assert.strictEqual(await q.claim(liveNext), null, 'and its room plays nothing else meanwhile');++ assert.deepStrictEqual(await row(undelivered), { state: 'queued', attempts: 1, claimed_by: null, lease_until: null, error: null }, 'undelivered: queued again');++ assert.deepStrictEqual((await q.list('stream:9602')).queued.map((x) => x.id), [undelivered, behind], 'in its old place');++ assert.deepStrictEqual(await row(delivered), { state: 'played', attempts: 1, claimed_by: 'proc-dead', lease_until: T - 1, error: 'delivered before a restart' }, 'delivered: finished, not replayed');++++ // P's heartbeat renews its lease; at the old expiry, Q's sweep still leaves P's row alone.++ clock.now = T + LEASE - 1;++ assert.strictEqual((await p.tick()).renewed, 1);++ clock.now = T + LEASE + 1;++ await q.tick();++ assert.strictEqual((await row(live)).state, 'playing');++ assert.strictEqual((await row(live)).claimed_by, 'proc-p');++++ // The replay: Q claims the requeued clip (its second claim), then dies before delivering it.++ const again = await q.claim(undelivered);++ assert.strictEqual(again.attempts, 2);++ assert.strictEqual(again.claimed_by, 'proc-q');++ q.stop();++ // Within Q's lease nobody touches it; past it, R's sweep fails it (no second replay) and the room moves on.++ const r = proc('proc-r', clock, plays);++ clock.now = again.lease_until - 1;++ await p.tick(); // P is still alive++ clock.now = again.lease_until;++ await r.tick();++ assert.strictEqual((await row(undelivered)).state, 'playing', 'lease not yet passed');++ clock.now = again.lease_until + 1;++ const swept = await r.tick();++ assert.ok(swept.failed >= 1, JSON.stringify(swept));++ const dead = await row(undelivered);++ assert.strictEqual(dead.state, 'failed');++ assert.strictEqual(dead.error, 'not delivered: its owner stopped twice');++ assert.strictEqual((await h.db.get('SELECT state, claimed_by FROM audio_requests WHERE id = ?', [behind])).claimed_by, 'proc-r');++ assert.deepStrictEqual(plays, [{ by: 'proc-r', id: behind }], 'the next clip plays; the failed one never does');++ assert.strictEqual((await row(live)).claimed_by, 'proc-p', 'P, alive throughout, still owns its clip');++ assert.strictEqual((await row(live)).state, 'playing');++ // A later recover() by anyone does not bring the failed clip back.++ const s = proc('proc-s', clock, plays);++ await s.recover({ now: clock.now });++ assert.strictEqual((await row(undelivered)).state, 'failed');++ for (const x of [p, r, s]) x.stop();++});++t.run(async () => {for (const w of sockets) { try { w.close(); } catch { /* */ } }if (h) await h.close();
@@ -31,10 +31,17 @@ before changing behaviour.** Browser JavaScript does not change; nginx routes th
@@ -31,10 +31,17 @@ before changing behaviour.** Browser JavaScript does not change; nginx routes th
`/api/tts/*`), 101soundboards and channel sounds (`/api/sounds*`), moderation utils and the word
`/api/tts/*`), 101soundboards and channel sounds (`/api/sounds*`), moderation utils and the word
filter, the deploy-notice card. Changes are confined to where they touched Live's data.
filter, the deploy-notice card. Changes are confined to where they touched Live's data.
- **The TTS and sound queue** (`audio-queue.js`, table `audio_requests`). TTS, channel `!sounds` and
- **The TTS and sound queue** (`audio-queue.js`, table `audio_requests`). TTS, channel `!sounds` and
- 101soundboards clips are persisted requests played one at a time per room — `queued → playing →
- played`, or `skipped` / `failed` — so the queue survives a Chat restart (what was playing is
- finished, what waited plays once the room has listeners again, requests older than 5 minutes
- expire) and a keyed request (the TTS of message `m<id>`) is read once. The broadcaster and
+ 101soundboards clips are persisted requests played one at a time per room, across every Chat
+ process (a claim holds the room's advisory lock; a partial unique index allows one `playing` row
+ per room) — `queued → playing → played`, or `skipped` / `failed` — so the queue survives a Chat
+ restart (what waited plays once the room has listeners again, requests older than 5 minutes
+ expire) and a keyed request (the TTS of message `m<id>`) is read once. A playing row is owned:
+ `claimed_by` (the process's instance id, host and port or `CHAT_AUDIO_INSTANCE_ID`) and a
+ 30-second `lease_until` its owner renews every 10 s. Only a dead owner's row is settled — its
+ lease passed, or it is the restarted process's own — at boot and by the heartbeat's sweep; a live
+ process's clip is never touched. Such a row is `played` if its delivery had started, otherwise
+ queued again in its old place, once: claimed a second time and dead again undelivered, it
+ `failed` (`attempts` counts claims), so a clip that crashes Chat cannot loop. The broadcaster and
moderators control it: `/skiptts [id]`, `/cleartts`, and `GET /api/tts/queue`,
moderators control it: `/skiptts [id]`, `/cleartts`, and `GET /api/tts/queue`,
`POST /api/tts/queue/skip|clear`, `POST /api/tts/queue/:id/report` (the playing client's
`POST /api/tts/queue/skip|clear`, `POST /api/tts/queue/:id/report` (the playing client's
`played`/`failed`). Browsers get the same `tts-audio` / `soundboard-audio` frames as before, now
`played`/`failed`). Browsers get the same `tts-audio` / `soundboard-audio` frames as before, now
@@ -88,6 +88,14 @@ The gaps the script found on 2026-09-25, with the product decisions taken for th
@@ -88,6 +88,14 @@ The gaps the script found on 2026-09-25, with the product decisions taken for th
the channel room and the global feed too. Cursor reads name them in `deleted_ids`.
the channel room and the global feed too. Cursor reads name them in `deleted_ids`.
- **`/clear` clears screens** (Twitch semantics) and now says so to the moderator: "Chat cleared on
- **`/clear` clears screens** (Twitch semantics) and now says so to the moderator: "Chat cleared on
screen; messages stay in history — use purge to remove them."
screen; messages stay in history — use purge to remove them."
+- **One TTS/sound clip per room across Chat processes** (2026-10-01). Two processes could each claim a
+ different queued request of one room and play both at once, and a booting process finished every
+ `playing` row, other live processes' included, dropping a clip whose owner died before delivering
+ it. A claim now holds the room's advisory lock and takes nothing while the room plays (partial
+ unique index `idx_audio_requests_one_playing`, `migrations/0002_audio_claim_owner.sql`); a playing
+ row carries its owner (`claimed_by`) and a lease (`lease_until`, renewed by the owner), and only a
+ dead owner's row is settled: `played` if its delivery had started, else replayed once in its old
+ place, then `failed` (see README, the TTS and sound queue).
## Fixed while building it
## Fixed while building it
new file mode 100644
new file mode 100644
@@ -0,0 +1,22 @@
@@ -0,0 +1,22 @@
+-- phase: expand
+-- The audio queue's claims are owned and one per room (server/chat/audio-queue.js):
+-- claimed_by the Chat process (its instance id) that claimed the row and plays it;
+-- lease_until ms; that process renews it while the row is playing. A playing row whose lease has passed
+-- belongs to a process that died: recover() and the sweeper settle it, never a live owner's.
+-- At most one playing row per room, across processes: the partial unique index is the store's guarantee
+-- (claim() also takes a per-room advisory lock, so a losing claimer just gets nothing).
+-- Safe on the live table and idempotent: nullable columns, no rewrite. The index is built in this transaction
+-- under a short SHARE ROW EXCLUSIVE lock (the table holds about a week of requests), so the duplicates it
+-- would refuse can be settled first and no claim of a previous-release process slips in between; CREATE INDEX
+-- CONCURRENTLY could not run next to that clean-up, and a failed concurrent build leaves an invalid index that
+-- IF NOT EXISTS would then keep.
+ALTER TABLE audio_requests ADD COLUMN IF NOT EXISTS claimed_by text COLLATE "C";
+ALTER TABLE audio_requests ADD COLUMN IF NOT EXISTS lease_until bigint;
+
+LOCK TABLE audio_requests IN SHARE ROW EXCLUSIVE MODE;
+-- Two clips playing at once in one room (the race this closes): keep the oldest, the others reached the room too.
+UPDATE audio_requests a SET state = 'played', error = 'delivered before a restart',
+ finished_at = COALESCE(a.finished_at, a.started_at, a.created_at)
+ WHERE a.state = 'playing'
+ AND EXISTS (SELECT 1 FROM audio_requests b WHERE b.room = a.room AND b.state = 'playing' AND b.id < a.id);
+CREATE UNIQUE INDEX IF NOT EXISTS idx_audio_requests_one_playing ON audio_requests(room) WHERE state = 'playing';
@@ -16,10 +16,27 @@
@@ -16,10 +16,27 @@
* window ends it is played and the next one goes out.
* window ends it is played and the next one goes out.
* played, skipped, failed final.
* played, skipped, failed final.
*
*
- * A Chat restart keeps the queue: `recover()` at boot finishes rows that were playing (they reached
- * the room before the restart), 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.
+ * One clip per room across every Chat process: `claim()` takes the room's advisory lock in its
+ * transaction and takes nothing while the room has a playing row (the partial unique index
+ * idx_audio_requests_one_playing, migrations/0002, is the store's guarantee). A process that loses
+ * gets nothing; the room's next pump (the owner's, when its clip ends, or the heartbeat) plays it.
+ *
+ * Claims are owned: a playing row records its process (`claimed_by`, the instance id: host and port,
+ * or CHAT_AUDIO_INSTANCE_ID) and a lease (`lease_until`) the owner renews every HEARTBEAT_MS while it
+ * lives. Only a row whose owner is dead is ever settled by another hand — its lease has passed, or it
+ * is this process's own row from before a restart (same instance id) — by `recover()` at boot and by
+ * the heartbeat's sweep. Such a row:
+ * - was never delivered (no started_at: its owner died while making it): it goes back to queued,
+ * in its old place (the queue plays by id), once; `attempts` counts claims, so a row claimed
+ * MAX_ATTEMPTS (2) times that dies again undelivered fails ('not delivered: its owner stopped
+ * twice') and a clip that crashes Chat cannot loop;
+ * - was delivered (started_at set): it is played ('delivered before a restart'), never replayed.
+ * A row of a live owner, on any process, is never touched.
+ *
+ * A Chat restart keeps the queue: `recover()` at boot settles its own and dead owners' playing rows
+ * 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.
* 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.
*
*
@@ -31,6 +48,7 @@
@@ -31,6 +48,7 @@
*/
*/
'use strict';
'use strict';
+const os = require('os');
const db = require('../db/database');
const db = require('../db/database');
const STATES = ['queued', 'playing', 'played', 'skipped', 'failed'];
const STATES = ['queued', 'playing', 'played', 'skipped', 'failed'];
@@ -47,7 +65,10 @@ const GAP_MS = 250; // between two clips
@@ -47,7 +65,10 @@ const GAP_MS = 250; // between two clips
const MIN_PLAY_MS = 500;
const MIN_PLAY_MS = 500;
const MAX_PLAY_MS = 60 * 1000;
const MAX_PLAY_MS = 60 * 1000;
const STALE_MS = 5 * 60 * 1000; // a request still queued this long after it was made is dropped
const STALE_MS = 5 * 60 * 1000; // a request still queued this long after it was made is dropped
-const MAX_ATTEMPTS = 3; // a request whose making crashed Chat this often fails
+const MAX_ATTEMPTS = 2; // claims of one request: its owner may die undelivered once (one replay)
+const LEASE_MS = 30 * 1000; // a playing row's owner is dead once its lease is this old…
+const HEARTBEAT_MS = 10 * 1000; // …and a live owner renews it this often (and sweeps dead owners' rows)
+const LOCK_SPACE = 71_440_216; // pg_advisory_xact_lock(LOCK_SPACE, hashtext(room)): one claim per room at a time
const KEEP_MS = 7 * 24 * 3600 * 1000;
const KEEP_MS = 7 * 24 * 3600 * 1000;
const RESUME_GRACE_MS = 8000; // after a restart, a room with waiting requests resumes by then…
const RESUME_GRACE_MS = 8000; // after a restart, a room with waiting requests resumes by then…
const RESUME_SETTLE_MS = 1000; // …or this long after a socket rejoins it
const RESUME_SETTLE_MS = 1000; // …or this long after a socket rejoins it
@@ -131,6 +152,10 @@ async function transition(id, to, { error = null, actor = null, durationMs = nul
@@ -131,6 +152,10 @@ async function transition(id, to, { error = null, actor = null, durationMs = nul
// ── Runtime ─────────────────────────────────────────────────────────────────────────────────────
// ── Runtime ─────────────────────────────────────────────────────────────────────────────────────
const performers = {}; // kind → async (row, payload) → { frame, durationMs } | null
const performers = {}; // kind → async (row, payload) → { frame, durationMs } | null
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 clock = () => Date.now();
+let heartbeat = null;
+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;
@@ -158,16 +183,29 @@ function roomJoined(room) {
@@ -158,16 +183,29 @@ function roomJoined(room) {
/**
/**
* 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.
+ * `heartbeat: false` leaves lease renewal and the sweep to the caller (tests call tick()).
*/
*/
-function init({ performers: p, deliver: d }) {
+function init({ performers: p, deliver: d, instanceId: id = null, clock: c = null, heartbeat: beat = true }) {
Object.assign(performers, p || {});
Object.assign(performers, p || {});
deliver = d;
deliver = d;
+ if (id) instanceId = String(id);
+ if (c) clock = c;
stopped = false;
stopped = false;
+ if (heartbeat) clearInterval(heartbeat);
+ heartbeat = null;
+ if (beat) {
+ heartbeat = setInterval(() => { tick().catch((err) => console.warn('[AudioQueue] heartbeat:', err.message)); }, HEARTBEAT_MS);
+ if (heartbeat.unref) heartbeat.unref();
+ }
}
}
/** Stop pacing timers (shutdown). States stay as they are; recover() picks them up at boot. */
/** Stop pacing timers (shutdown). States stay as they are; recover() picks them up at boot. */
function stop() {
function stop() {
stopped = true;
stopped = true;
+ if (heartbeat) clearInterval(heartbeat);
+ heartbeat = null;
+ waiting.clear();
for (const st of rooms.values()) {
for (const st of rooms.values()) {
if (st.timer) clearTimeout(st.timer);
if (st.timer) clearTimeout(st.timer);
if (st.holdTimer) clearTimeout(st.holdTimer);
if (st.holdTimer) clearTimeout(st.holdTimer);
@@ -210,11 +248,34 @@ async function enqueue({ kind, streamId = null, channelUserId = null, requestedB
@@ -210,11 +248,34 @@ async function enqueue({ kind, streamId = null, channelUserId = null, requestedB
}
}
/**
/**
- * Take a queued request to play (plan T3 decision 3): one atomic UPDATE, so of two pumps (two processes) at most
- * one gets the row back; null = someone else took it, or it was skipped. Counts the attempt.
+ * Take a queued request to play (plan T3 decision 3), owned by this process under a lease. One transaction holding
+ * the room's advisory lock: nothing while the room has a playing row (any process's), else one atomic UPDATE, so of
+ * two pumps (two processes) at most one gets a row of a room back. Counts the attempt.
+ * → the row, or null with `busy` set when the room is playing (null without it: someone took or skipped this row).
*/
*/
-async function claim(id) {
- return await db.get("UPDATE audio_requests SET state = 'playing', attempts = attempts + 1 WHERE id = ? AND state = 'queued' RETURNING *", [id]) || null;
+async function claim(id, { now = clock() } = {}) {
+ const out = await claimOrBusy(id, now);
+ return out.row;
+}
+async function claimOrBusy(id, now) {
+ try {
+ return await db.tx(async () => {
+ const q = await db.get("SELECT room FROM audio_requests WHERE id = ? AND state = 'queued'", [id]);
+ if (!q) return { row: null, busy: false };
+ await db.get('SELECT pg_advisory_xact_lock(?, hashtext(?)) AS locked', [LOCK_SPACE, q.room]);
+ if (await db.get("SELECT 1 AS one FROM audio_requests WHERE room = ? AND state = 'playing'", [q.room])) return { row: null, busy: true };
+ const row = await db.get(
+ `UPDATE audio_requests SET state = 'playing', attempts = attempts + 1, claimed_by = ?, lease_until = ?
+ WHERE id = ? AND state = 'queued' RETURNING *`,
+ [instanceId, now + LEASE_MS, id],
+ );
+ return { row: row || null, busy: false };
+ });
+ } catch (err) {
+ // The one-playing-row index refused it: a claimer that does not take the lock (a previous release) won.
+ if (err && err.code === '23505') return { row: null, busy: true };
+ throw err;
+ }
}
}
/** Play the next request of a room if none is playing. */
/** Play the next request of a room if none is playing. */
@@ -228,10 +289,12 @@ async function pump(room) {
@@ -228,10 +289,12 @@ async function pump(room) {
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 (!row) 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 claimed = await claim(row.id);
+ const { row: claimed, busy } = await claimOrBusy(row.id, clock());
+ if (busy) { waiting.add(room); return; } // another process plays this room: its finish (or the heartbeat) pumps again
if (!claimed) continue; // another pump took it (or it was skipped): the next one
if (!claimed) continue; // another pump took it (or it was skipped): the next one
+ waiting.delete(room);
row = claimed;
row = claimed;
// The audio is made outside any transaction (a synth takes seconds); the row stays 'playing' meanwhile.
// The audio is made outside any transaction (a synth takes seconds); the row stays 'playing' meanwhile.
st.makingId = row.id;
st.makingId = row.id;
@@ -326,21 +389,59 @@ async function list(room, { recent = 20 } = {}) {
@@ -326,21 +389,59 @@ async function list(room, { recent = 20 } = {}) {
}
}
/**
/**
- * Boot: rows playing when Chat stopped were delivered, so they are played; rows queued too long
- * ago fail as expired; every room with queued rows starts playing again.
+ * Settle the playing rows of dead owners: a lease that has passed (or none: a previous release's claim), and with
+ * `mine` this instance's own rows (at boot they are from before the restart). Delivered → played; undelivered →
+ * queued again in its old place, or failed once it has been claimed MAX_ATTEMPTS times. A live owner's row is
+ * never touched. → { played, requeued, failed, rooms }
*/
*/
-async function recover({ now = Date.now(), graceMs = RESUME_GRACE_MS } = {}) {
- const played = (await db.run("UPDATE audio_requests SET state = 'played', finished_at = ?, error = 'delivered before a restart' WHERE state = 'playing'", [now])).changes;
+async function reclaimDead({ now = clock(), mine = false } = {}) {
+ const dead = `state = 'playing' AND (lease_until IS NULL OR lease_until < ?${mine ? ' OR claimed_by = ?' : ''})`;
+ const args = mine ? [now, instanceId] : [now];
+ const played = await db.all(`UPDATE audio_requests SET state = 'played', finished_at = ?, error = 'delivered before a restart'
+ WHERE ${dead} AND started_at IS NOT NULL RETURNING room`, [now, ...args]);
+ const failed = await db.all(`UPDATE audio_requests SET state = 'failed', finished_at = ?, error = 'not delivered: its owner stopped twice', actor = 'system'
+ WHERE ${dead} AND started_at IS NULL AND attempts >= ? RETURNING room`, [now, ...args, MAX_ATTEMPTS]);
+ const requeued = await db.all(`UPDATE audio_requests SET state = 'queued', claimed_by = NULL, lease_until = NULL, duration_ms = NULL
+ WHERE ${dead} AND started_at IS NULL AND attempts < ? RETURNING room`, [...args, MAX_ATTEMPTS]);
+ const rooms = [...new Set([...played, ...failed, ...requeued].map((r) => r.room))];
+ return { played: played.length, requeued: requeued.length, failed: failed.length, rooms };
+}
+
+/**
+ * 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() } = {}) {
+ 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 });
+ if (swept.rooms.length) console.log(`[AudioQueue] swept dead owners' rows: ${swept.played} finished, ${swept.requeued} requeued, ${swept.failed} failed`);
+ const rooms = [...new Set([...swept.rooms, ...waiting])];
+ for (const room of rooms) await pump(room).catch((err) => console.warn('[AudioQueue] pump:', err.message));
+ return { renewed, ...swept, rooms };
+}
+
+/**
+ * Boot: this instance's rows from before the restart and dead owners' rows are settled (reclaimDead); rows
+ * 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 } = {}) {
+ 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;
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;
await db.run("DELETE FROM audio_requests WHERE state IN ('played', 'skipped', 'failed') AND COALESCE(finished_at, created_at) < ?", [now - KEEP_MS]);
await db.run("DELETE FROM audio_requests WHERE state IN ('played', 'skipped', 'failed') AND COALESCE(finished_at, created_at) < ?", [now - KEEP_MS]);
const pending = (await db.all("SELECT DISTINCT room FROM audio_requests WHERE state = 'queued'")).map((r) => r.room);
const pending = (await db.all("SELECT DISTINCT room FROM audio_requests WHERE state = 'queued'")).map((r) => r.room);
for (const room of pending) holdRoom(room, now + graceMs);
for (const room of pending) holdRoom(room, now + graceMs);
- if (played || expired || pending.length) console.log(`[AudioQueue] recovered: ${played} finished, ${expired} expired, ${pending.length} room(s) resuming`);
- return { played, expired, rooms: pending };
+ if (played || settled.requeued || settled.failed || expired || pending.length) {
+ console.log(`[AudioQueue] recovered: ${played} finished, ${settled.requeued} requeued, ${settled.failed} failed, ${expired} expired, ${pending.length} room(s) resuming`);
+ }
+ return { played, requeued: settled.requeued, failed: settled.failed, expired, rooms: pending };
}
}
module.exports = {
module.exports = {
- STATES, TRANSITIONS, KINDS, STALE_MS, GAP_MS, RESUME_GRACE_MS, RESUME_SETTLE_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, roomJoined, getRequest, transition,
+ init, stop, enqueue, pump, claim, skip, clear, report, list, recover, tick, reclaimDead, roomJoined, getRequest, transition,
+ get instanceId() { return instanceId; },
};
};
@@ -4,7 +4,9 @@
@@ -4,7 +4,9 @@
* each with its own pool connections on one database — race over the same queued requests. The claim is one atomic
* each with its own pool connections on one database — race over the same queued requests. The claim is one atomic
* UPDATE … WHERE state = 'queued' RETURNING *, so every request plays exactly once; the synth runs outside any
* UPDATE … WHERE state = 'queued' RETURNING *, so every request plays exactly once; the synth runs outside any
* transaction. And enqueue's dedupe is the unique (room, dedupe_key) index: concurrent enqueues of one key store
* transaction. And enqueue's dedupe is the unique (room, dedupe_key) index: concurrent enqueues of one key store
- * one row. On PostgreSQL (npm run test:pg) the two pumps really run on separate connections.
+ * one row. And a room plays one request at a time across processes: claim() holds the room's advisory lock and
+ * takes nothing while the room has a playing row (backed by the one-playing-row partial unique index).
+ * On PostgreSQL (npm run test:pg) the two pumps really run on separate connections.
*/
*/
const assert = require('assert');
const assert = require('assert');
const { suite } = require('./helpers');
const { suite } = require('./helpers');
@@ -26,6 +28,7 @@ function instance(name, plays) {
@@ -26,6 +28,7 @@ function instance(name, plays) {
},
},
},
},
deliver: (row, frame) => { if (frame.type === 'tts-audio') plays.push({ by: name, id: row.id }); },
deliver: (row, frame) => { if (frame.type === 'tts-audio') plays.push({ by: name, id: row.id }); },
+ instanceId: `proc-${name}`, heartbeat: false,
});
});
return aq;
return aq;
}
}
@@ -71,6 +74,51 @@ t('a request claimed by one pump is not claimed again', async () => {
@@ -71,6 +74,51 @@ t('a request claimed by one pump is not claimed again', async () => {
assert.strictEqual((await db.get('SELECT attempts FROM audio_requests WHERE id = ?', [r.id])).attempts, 1);
assert.strictEqual((await db.get('SELECT attempts FROM audio_requests WHERE id = ?', [r.id])).attempts, 1);
});
});
+t('two requests in one room, two claimers at once: one plays, the other gets nothing until it ends', async () => {
+ const ids = [];
+ for (const label of ['first', 'second']) {
+ const r = await db.run(`INSERT INTO audio_requests (room, stream_id, kind, state, label, payload, created_at)
+ VALUES ('stream:9300', 9300, 'tts', 'queued', ?, '{}', ?)`, [label, Date.now()]);
+ ids.push(Number(r.lastInsertRowid));
+ }
+ for (let round = 0; round < 5; round++) {
+ // Each claimer goes for a different request of the room at the same time.
+ const [x, y] = await Promise.all([a.claim(ids[0]), b.claim(ids[1])]);
+ assert.strictEqual([x, y].filter(Boolean).length, 1, `round ${round}: exactly one claim wins`);
+ const won = x || y;
+ const loser = x ? b : a;
+ const other = ids.find((id) => id !== won.id);
+ assert.strictEqual(won.claimed_by, x ? 'proc-a' : 'proc-b');
+ assert.ok(won.lease_until > Date.now());
+ assert.strictEqual((await db.get("SELECT COUNT(*) AS n FROM audio_requests WHERE room = 'stream:9300' AND state = 'playing'")).n, 1);
+ // While it plays, the loser gets nothing, whichever request it tries.
+ assert.strictEqual(await loser.claim(other), null);
+ assert.strictEqual(await a.claim(other), null);
+ assert.strictEqual((await db.get('SELECT state, attempts FROM audio_requests WHERE id = ?', [other])).state, 'queued');
+ // It ends: the loser's next try takes the other one.
+ assert.ok(await a.transition(won.id, 'played'));
+ const next = await loser.claim(other);
+ assert.ok(next && next.id === other && next.state === 'playing');
+ assert.ok(await a.transition(other, 'played'));
+ // Again with two fresh requests.
+ ids.length = 0;
+ for (const label of ['first', 'second']) {
+ const r = await db.run(`INSERT INTO audio_requests (room, stream_id, kind, state, label, payload, created_at)
+ VALUES ('stream:9300', 9300, 'tts', 'queued', ?, '{}', ?)`, [`${label}${round}`, Date.now()]);
+ ids.push(Number(r.lastInsertRowid));
+ }
+ }
+});
+
+t('the store refuses a second playing row in a room, whoever writes it', async () => {
+ const r = await db.run(`INSERT INTO audio_requests (room, stream_id, kind, state, label, payload, created_at)
+ VALUES ('stream:9300', 9300, 'tts', 'queued', 'raw', '{}', ?)`, [Date.now()]);
+ const playing = await a.claim(Number(r.lastInsertRowid) - 2) || await a.claim(Number(r.lastInsertRowid) - 1);
+ assert.ok(playing, 'one of the earlier requests is playing');
+ // A claimer without the lock (the previous release's bare UPDATE) is refused by the partial unique index.
+ await assert.rejects(db.run("UPDATE audio_requests SET state = 'playing' WHERE id = ?", [Number(r.lastInsertRowid)]), (err) => err.code === '23505');
+});
+
t('concurrent enqueues of one dedupe key store one row', async () => {
t('concurrent enqueues of one dedupe key store one row', async () => {
const outs = await Promise.all([a, b, a, b].map((q) => q.enqueue({ kind: 'tts', streamId: 9200, label: 'dup', payload: {}, dedupeKey: 'k1' })));
const outs = await Promise.all([a, b, a, b].map((q) => q.enqueue({ kind: 'tts', streamId: 9200, label: 'dup', payload: {}, dedupeKey: 'k1' })));
assert.strictEqual(outs.filter((o) => o.queued).length, 1, JSON.stringify(outs));
assert.strictEqual(outs.filter((o) => o.queued).length, 1, JSON.stringify(outs));
@@ -6,7 +6,9 @@
@@ -6,7 +6,9 @@
* recorded and the queue moves on; a keyed request is queued once; and a Chat restart (a real
* 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,
* 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
* once. The frames clients already know (tts-audio, soundboard-audio) are unchanged apart from a
- * new request_id.
+ * new request_id. The restart boundary across processes runs on an injected clock and instance ids:
+ * 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.
*/
*/
const assert = require('assert');
const assert = require('assert');
const fs = require('fs');
const fs = require('fs');
@@ -254,7 +256,7 @@ t('restart: what was playing is finished, what waited plays in order once, stale
@@ -254,7 +256,7 @@ t('restart: what was playing is finished, what waited plays in order once, stale
await ins.run(`stream:${streamId}`, streamId, 'crashy', JSON.stringify({ file, title: '!crashy' }), Date.now(), 3);
await ins.run(`stream:${streamId}`, streamId, 'crashy', JSON.stringify({ file, title: '!crashy' }), Date.now(), 3);
});
});
port = await h.freePort();
port = await h.freePort();
- chat = await h.spawnChat({ port });
+ 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');
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).
// The room's queue is held until it has listeners again (a second after the first rejoins).
await h.sleep(300);
await h.sleep(300);
@@ -278,7 +280,7 @@ t('restart: what was playing is finished, what waited plays in order once, stale
@@ -278,7 +280,7 @@ t('restart: what was playing is finished, what waited plays in order once, stale
const exit = await h.stopChat(chat);
const exit = await h.stopChat(chat);
assert.deepStrictEqual(exit, { code: 0, signal: null }, chat.log);
assert.deepStrictEqual(exit, { code: 0, signal: null }, chat.log);
- chat = await h.spawnChat({ port });
+ chat = await h.spawnChat({ port, env: { CHAT_AUDIO_INSTANCE_ID: 'chat-under-test' } });
await h.sleep(300);
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');
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 ears2 = await joined(listenerUser, '198.51.100.130');
@@ -300,6 +302,88 @@ t('restart: what was playing is finished, what waited plays in order once, stale
@@ -300,6 +302,88 @@ t('restart: what was playing is finished, what waited plays in order once, stale
assert.strictEqual(await raw(async (d) => (await d.prepare('SELECT state FROM audio_requests WHERE id = ?').get(queued[2].id)).state), 'skipped');
assert.strictEqual(await raw(async (d) => (await d.prepare('SELECT state FROM audio_requests WHERE id = ?').get(queued[2].id)).state), 'skipped');
});
});
+// ── 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) {
+ 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,
+ });
+ return q;
+}
+t('restart boundary: live owners are left alone, a dead owner\'s clip is replayed once if undelivered, finished if delivered', async () => {
+ // A day ahead of the wall clock: nothing here can look expired to the queues running on real time in this run.
+ const T = Date.now() + 24 * 3600e3;
+ const clock = { now: T };
+ const plays = [];
+ const LEASE = aq.LEASE_MS;
+ const put = async (room, label, f) => Number((await h.db.run(
+ `INSERT INTO audio_requests (room, stream_id, kind, state, label, payload, created_at, attempts, claimed_by, lease_until, started_at)
+ VALUES (?, ?, 'tts', ?, ?, '{}', ?, ?, ?, ?, ?)`,
+ [room, Number(room.split(':')[1]), f.state, label, T - 1000, f.attempts || 0, f.owner || null, f.lease || null, f.started || null],
+ )).lastInsertRowid);
+ const row = async (id) => await h.db.get('SELECT state, attempts, claimed_by, lease_until, error FROM audio_requests WHERE id = ?', [id]);
+
+ // stream:9601 — process P is alive and playing; stream:9602 — a dead owner's clip still being made;
+ // stream:9603 — a dead owner's clip already delivered.
+ const live = await put('stream:9601', 'live', { state: 'playing', attempts: 1, owner: 'proc-p', lease: T + LEASE, started: T - 500 });
+ const liveNext = await put('stream:9601', 'live-next', { state: 'queued' });
+ const undelivered = await put('stream:9602', 'undelivered', { state: 'playing', attempts: 1, owner: 'proc-dead', lease: T - 1 });
+ const behind = await put('stream:9602', 'behind', { state: 'queued' });
+ const delivered = await put('stream:9603', 'delivered', { state: 'playing', attempts: 1, owner: 'proc-dead', lease: T - 1, started: T - 3000 });
+
+ const p = proc('proc-p', clock, plays);
+ const q = proc('proc-q', clock, plays);
+ const out = await q.recover({ now: T }); // Q boots while P lives
+ assert.ok(out.requeued >= 1 && out.played >= 1, JSON.stringify(out));
+ assert.deepStrictEqual(await row(live), { state: 'playing', attempts: 1, claimed_by: 'proc-p', lease_until: T + LEASE, error: null }, 'a live owner\'s row is untouched');
+ assert.strictEqual(await q.claim(liveNext), null, 'and its room plays nothing else meanwhile');
+ assert.deepStrictEqual(await row(undelivered), { state: 'queued', attempts: 1, claimed_by: null, lease_until: null, error: null }, 'undelivered: queued again');
+ assert.deepStrictEqual((await q.list('stream:9602')).queued.map((x) => x.id), [undelivered, behind], 'in its old place');
+ assert.deepStrictEqual(await row(delivered), { state: 'played', attempts: 1, claimed_by: 'proc-dead', lease_until: T - 1, error: 'delivered before a restart' }, 'delivered: finished, not replayed');
+
+ // P's heartbeat renews its lease; at the old expiry, Q's sweep still leaves P's row alone.
+ clock.now = T + LEASE - 1;
+ assert.strictEqual((await p.tick()).renewed, 1);
+ clock.now = T + LEASE + 1;
+ await q.tick();
+ assert.strictEqual((await row(live)).state, 'playing');
+ assert.strictEqual((await row(live)).claimed_by, 'proc-p');
+
+ // The replay: Q claims the requeued clip (its second claim), then dies before delivering it.
+ const again = await q.claim(undelivered);
+ assert.strictEqual(again.attempts, 2);
+ assert.strictEqual(again.claimed_by, 'proc-q');
+ q.stop();
+ // Within Q's lease nobody touches it; past it, R's sweep fails it (no second replay) and the room moves on.
+ const r = proc('proc-r', clock, plays);
+ clock.now = again.lease_until - 1;
+ await p.tick(); // P is still alive
+ clock.now = again.lease_until;
+ await r.tick();
+ assert.strictEqual((await row(undelivered)).state, 'playing', 'lease not yet passed');
+ clock.now = again.lease_until + 1;
+ const swept = await r.tick();
+ assert.ok(swept.failed >= 1, JSON.stringify(swept));
+ const dead = await row(undelivered);
+ assert.strictEqual(dead.state, 'failed');
+ assert.strictEqual(dead.error, 'not delivered: its owner stopped twice');
+ assert.strictEqual((await h.db.get('SELECT state, claimed_by FROM audio_requests WHERE id = ?', [behind])).claimed_by, 'proc-r');
+ assert.deepStrictEqual(plays, [{ by: 'proc-r', id: behind }], 'the next clip plays; the failed one never does');
+ assert.strictEqual((await row(live)).claimed_by, 'proc-p', 'P, alive throughout, still owns its clip');
+ assert.strictEqual((await row(live)).state, 'playing');
+ // A later recover() by anyone does not bring the failed clip back.
+ const s = proc('proc-s', clock, plays);
+ await s.recover({ now: clock.now });
+ assert.strictEqual((await row(undelivered)).state, 'failed');
+ for (const x of [p, r, s]) x.stop();
+});
+
t.run(async () => {
t.run(async () => {
for (const w of sockets) { try { w.close(); } catch { /* */ } }
for (const w of sockets) { try { w.close(); } catch { /* */ } }
if (h) await h.close();
if (h) await h.close();