diff --git a/CloudServer/server.js b/CloudServer/server.js index 6c656ec..54e99df 100644 --- a/CloudServer/server.js +++ b/CloudServer/server.js @@ -500,10 +500,29 @@ function getCableGroupByStationName(stationName) { return null; } +function sendWsToAllConnectionsOfStation(stationName, msgObj) { + const clean = String(stationName || '').trim(); + if (!clean) return 0; + const jsonStr = (typeof msgObj === 'string') ? msgObj : JSON.stringify(msgObj); + let sent = 0; + for (const ws of getAllStationConnections(clean)) { + try { ws.send(jsonStr); sent++; } catch (e) {} + } + return sent; +} + +function sendBinaryToAllConnectionsOfStation(stationName, buf) { + let sent = 0; + for (const ws of getAllStationConnections(stationName)) { + try { ws.send(buf); sent++; } catch (e) {} + } + return sent; +} + function sendWsToStationMaster(stationName, msgObj) { const { ws } = getStationMaster(stationName); - if (!ws || ws.readyState !== WebSocket.OPEN) return false; - try { ws.send(JSON.stringify(msgObj)); return true; } catch { return false; } + if (!ws || ws.readyState !== WebSocket.OPEN) return 0; + try { ws.send(JSON.stringify(msgObj)); return 1; } catch { return 0; } } function sendStationGroupState(stationName) { @@ -527,11 +546,12 @@ function sendStationGroupState(stationName) { } } -async function acquireCableToken(cableGroupId, stationName, audioId, priority, cancelToken) { +async function acquireCableToken(cableGroupId, stationName, audioId, priority, cancelToken, taskDurationHintSec) { const cgId = String(cableGroupId || '').trim(); const stName = String(stationName || '').trim(); const audId = String(audioId || ''); const prio = Number(priority || 0); + const durHint = Number(taskDurationHintSec || 0) || 0; if (!cgId) return { ok: true, reason: 'no_cable_group' }; if (!stName) return { ok: false, reason: 'no_station' }; @@ -543,9 +563,10 @@ async function acquireCableToken(cableGroupId, stationName, audioId, priority, c holderAudioId: audId, holderPriority: prio, acquiredAt: Date.now(), + durationHintSec: durHint, cancelToken: cancelToken || null }); - sendWsToStationMaster(stName, { + sendWsToAllConnectionsOfStation(stName, { type: 'cable_token_granted', cableGroupId: cgId, stationName: stName, @@ -566,6 +587,7 @@ async function acquireCableToken(cableGroupId, stationName, audioId, priority, c holderAudioId: audId, holderPriority: prio, acquiredAt: Date.now(), + durationHintSec: durHint, cancelToken: cancelToken || null }); return { ok: true, reason: 'same_station_updated' }; @@ -573,15 +595,14 @@ async function acquireCableToken(cableGroupId, stationName, audioId, priority, c // Higher priority can preempt lower if (prio > existing.holderPriority) { - sendWsToStationMaster(existing.holderStationName, { + sendWsToAllConnectionsOfStation(existing.holderStationName, { type: 'audio_preempt', reason: 'higher_priority', byStationName: stName, priority: prio, audioId: audId }); - // Send denied to previous holder - sendWsToStationMaster(existing.holderStationName, { + sendWsToAllConnectionsOfStation(existing.holderStationName, { type: 'cable_token_denied', cableGroupId: cgId, stationName: existing.holderStationName, @@ -596,9 +617,10 @@ async function acquireCableToken(cableGroupId, stationName, audioId, priority, c holderAudioId: audId, holderPriority: prio, acquiredAt: Date.now(), + durationHintSec: durHint, cancelToken: cancelToken || null }); - sendWsToStationMaster(stName, { + sendWsToAllConnectionsOfStation(stName, { type: 'cable_token_granted', cableGroupId: cgId, stationName: stName, @@ -607,8 +629,52 @@ async function acquireCableToken(cableGroupId, stationName, audioId, priority, c return { ok: true, reason: 'preempted_lower' }; } - // Lower or equal priority: deny - sendWsToStationMaster(stName, { + // Equal priority: preempt if existing token has run beyond (durHint + EQUAL_PRIO_GRACE_SEC). + // This solves "holder=CE-01" starve forever because both approaching are p=2 equal priority. + const EQUAL_PRIO_GRACE_SEC = 30; + if (prio === existing.holderPriority) { + const ageMs = Date.now() - (existing.acquiredAt || 0); + const existingDurHintSec = Number(existing.durationHintSec || 0); + const deadlineMs = Math.max(EQUAL_PRIO_GRACE_SEC, (existingDurHintSec + EQUAL_PRIO_GRACE_SEC)) * 1000; + if (ageMs >= deadlineMs) { + log('WARN', `Cable token equal-prio switch: group=${cgId} old=${existing.holderStationName} age=${Math.round(ageMs/1000)}s new=${stName}`); + sendWsToAllConnectionsOfStation(existing.holderStationName, { + type: 'audio_preempt', + reason: 'equal_prio_timeout', + byStationName: stName, + priority: prio, + audioId: audId + }); + sendWsToAllConnectionsOfStation(existing.holderStationName, { + type: 'cable_token_denied', + cableGroupId: cgId, + stationName: existing.holderStationName, + reason: 'equal_prio_timeout', + holderStationName: stName, + holderPriority: prio + }); + if (existing.cancelToken) existing.cancelToken.cancelled = true; + await delay(200); + audioCableTokens.set(cgId, { + holderStationName: stName, + holderAudioId: audId, + holderPriority: prio, + acquiredAt: Date.now(), + durationHintSec: durHint, + cancelToken: cancelToken || null + }); + sendWsToAllConnectionsOfStation(stName, { + type: 'cable_token_granted', + cableGroupId: cgId, + stationName: stName, + audioId: audId + }); + return { ok: true, reason: 'equal_prio_timeout' }; + } + } + + // Lower or equal-but-not-expired priority: deny + sendWsToAllConnectionsOfStation(stName, { type: 'cable_token_denied', cableGroupId: cgId, stationName: stName, @@ -622,17 +688,38 @@ async function acquireCableToken(cableGroupId, stationName, audioId, priority, c function releaseCableToken(cableGroupId, stationName, audioId) { const cgId = String(cableGroupId || '').trim(); const stName = String(stationName || '').trim(); - const audId = String(audioId || ''); if (!cgId) return false; const existing = audioCableTokens.get(cgId); if (!existing) return false; - if (existing.holderStationName !== stName) return false; - if (audId && existing.holderAudioId && existing.holderAudioId !== audId) return false; + if (stName && existing.holderStationName !== stName) return false; if (existing.cancelToken) existing.cancelToken.cancelled = true; audioCableTokens.delete(cgId); return true; } +// Cable token watchdog: force-release tokens whose holder went offline, or that are stale +// beyond a generous safety bound (music loops can legitimately be long but they should +// release via stop API, watchdog is only the last line of defense against leaks). +const CABLE_TOKEN_MAX_HOLD_MS = 10 * 60 * 1000; // 10 minutes hard cap +function runCableTokenWatchdog() { + const now = Date.now(); + for (const [cgId, tok] of Array.from(audioCableTokens.entries())) { + if (!tok) { audioCableTokens.delete(cgId); continue; } + const st = tok.holderStationName ? stations[tok.holderStationName] : null; + const connected = !!(st && Array.isArray(st.nodes) && st.nodes.some(n => n.ws && n.ws.readyState === WebSocket.OPEN)); + const age = now - (tok.acquiredAt || 0); + let release = false; + let reason = ''; + if (!connected) { release = true; reason = 'holder_offline'; } + else if (age >= CABLE_TOKEN_MAX_HOLD_MS) { release = true; reason = 'max_hold_timeout'; } + if (release) { + log('WARN', `Cable token watchdog release: group=${cgId} holder=${tok.holderStationName || ''} age=${Math.round(age/1000)}s reason=${reason}`); + if (tok.cancelToken) try { tok.cancelToken.cancelled = true; } catch {} + audioCableTokens.delete(cgId); + } + } +} + function getLocalStationScriptVersion() { try { if (!fs.existsSync(stationScriptPath)) return ''; @@ -994,25 +1081,26 @@ function sendAudioControl(stationName, msg, fallbackWs) { async function sendAudioBufferToStation(stationName, fallbackWs, task) { const { ws: targetWs } = getStationMaster(stationName); const masterWs = targetWs || (fallbackWs && fallbackWs.readyState === WebSocket.OPEN ? fallbackWs : null); - if (!masterWs) throw new Error(`Station "${stationName}" not connected`); + const allTargets = getAllStationConnections(stationName); + if (!masterWs && allTargets.length === 0) throw new Error(`Station "${stationName}" not connected`); const cableGrp = getCableGroupByStationName(stationName); const audioId = String(task.audioId || `${Date.now()}-${Math.random().toString(16).slice(2)}`); const prio = Number(task.priority || 0); + const durationSec = Number(task.durationSec || 0); if (cableGrp) { - const res = await acquireCableToken(cableGrp.id, stationName, audioId, prio, task.cancelToken || null); + const res = await acquireCableToken(cableGrp.id, stationName, audioId, prio, task.cancelToken || null, durationSec); if (!res.ok) { throw new Error(`Cable token denied: ${res.reason || 'unknown'} (holder=${res.holderStationName || 'none'} p=${res.holderPriority ?? 0})`); } } const buf = task.buffer || null; - const durationSec = Number(task.durationSec || 0); const audioFormat = String(task.audioFormat || 'dfpwm').toLowerCase(); const streamUrl = task.streamUrl ? String(task.streamUrl) : null; - const startMsg = { + const startMsg = JSON.stringify({ type: 'audio_start', stationName, audioId, @@ -1026,15 +1114,18 @@ async function sendAudioBufferToStation(stationName, fallbackWs, task) { interruptMode: task.interruptMode || getInterruptModeForPriority(prio), source: task.source || 'unknown', cableGroupId: cableGrp?.id || '' - }; + }); + sendWsToAllConnectionsOfStation(stationName, startMsg); - masterWs.send(JSON.stringify(startMsg)); + const releaseToken = () => { + if (cableGrp) try { releaseCableToken(cableGrp.id, stationName, audioId); } catch {} + }; if (streamUrl) { if (cableGrp && durationSec > 0) { - setTimeout(() => { - try { releaseCableToken(cableGrp.id, stationName, audioId); } catch {} - }, Math.max(100, Math.floor(durationSec * 1000) + 500)); + setTimeout(() => releaseToken(), Math.max(100, Math.floor(durationSec * 1000) + 500)); + } else if (cableGrp) { + releaseToken(); } return { audioId, cancelled: false, durationSec, cableGroupId: cableGrp?.id || '' }; } @@ -1045,49 +1136,36 @@ async function sendAudioBufferToStation(stationName, fallbackWs, task) { try { if (buf.length <= SINGLE_SEND_LIMIT) { - if (token && token.cancelled) { - if (cableGrp) releaseCableToken(cableGrp.id, stationName, audioId); - return { audioId, cancelled: true, durationSec, cableGroupId: cableGrp?.id || '' }; - } - masterWs.send(buf); + if (token && token.cancelled) { releaseToken(); return { audioId, cancelled: true, durationSec, cableGroupId: cableGrp?.id || '' }; } + sendBinaryToAllConnectionsOfStation(stationName, buf); } else { for (let offset = 0; offset < buf.length; offset += CHUNK_SIZE) { - if (token && token.cancelled) { - if (cableGrp) releaseCableToken(cableGrp.id, stationName, audioId); - return { audioId, cancelled: true, durationSec, cableGroupId: cableGrp?.id || '' }; - } + if (token && token.cancelled) { releaseToken(); return { audioId, cancelled: true, durationSec, cableGroupId: cableGrp?.id || '' }; } const end = Math.min(offset + CHUNK_SIZE, buf.length); - masterWs.send(buf.slice(offset, end)); + sendBinaryToAllConnectionsOfStation(stationName, buf.slice(offset, end)); const isInitial = offset < CHUNK_SIZE * 4; await delay(isInitial ? 20 : 60); } } - if (token && token.cancelled) { - if (cableGrp) releaseCableToken(cableGrp.id, stationName, audioId); - return { audioId, cancelled: true, durationSec, cableGroupId: cableGrp?.id || '' }; - } + if (token && token.cancelled) { releaseToken(); return { audioId, cancelled: true, durationSec, cableGroupId: cableGrp?.id || '' }; } - masterWs.send(JSON.stringify({ + sendWsToAllConnectionsOfStation(stationName, { type: 'audio_end', stationName, audioId, durationSec - })); + }); if (cableGrp && durationSec > 0) { - setTimeout(() => { - try { releaseCableToken(cableGrp.id, stationName, audioId); } catch {} - }, Math.max(100, Math.floor(durationSec * 1000) + 500)); - } else if (cableGrp) { - releaseCableToken(cableGrp.id, stationName, audioId); + setTimeout(() => releaseToken(), Math.max(100, Math.floor(durationSec * 1000) + 500)); + } else { + releaseToken(); } return { audioId, cancelled: false, durationSec, cableGroupId: cableGrp?.id || '' }; } catch (e) { - if (cableGrp) { - try { releaseCableToken(cableGrp.id, stationName, audioId); } catch {} - } + releaseToken(); throw e; } } @@ -1118,7 +1196,8 @@ function buildWavChunkFromPcm(buffer, sampleRate = 48000, channels = 1, bitsPerS async function sendLiveWavChunksToStation(stationName, fallbackWs, task) { const { ws: targetWs } = getStationMaster(stationName); const masterWs = targetWs || (fallbackWs && fallbackWs.readyState === WebSocket.OPEN ? fallbackWs : null); - if (!masterWs) throw new Error(`Station "${stationName}" not connected`); + const allTargets = getAllStationConnections(stationName); + if (!masterWs && allTargets.length === 0) throw new Error(`Station "${stationName}" not connected`); const cableGrp = getCableGroupByStationName(stationName); const audioId = String(task.audioId || `${Date.now()}-${Math.random().toString(16).slice(2)}`); @@ -1126,19 +1205,22 @@ async function sendLiveWavChunksToStation(stationName, fallbackWs, task) { const durationSec = Number(task.durationSec || 0); if (cableGrp) { - const res = await acquireCableToken(cableGrp.id, stationName, audioId, prio, task.cancelToken || null); + const res = await acquireCableToken(cableGrp.id, stationName, audioId, prio, task.cancelToken || null, durationSec || (task.loop ? 0 : 120)); if (!res.ok) { throw new Error(`Cable token denied: ${res.reason || 'unknown'} (holder=${res.holderStationName || 'none'} p=${res.holderPriority ?? 0})`); } } const sourcePath = String(task.sourcePath || '').trim(); + const releaseToken = () => { + if (cableGrp) try { releaseCableToken(cableGrp.id, stationName, audioId); } catch {} + }; if (!sourcePath) { - if (cableGrp) releaseCableToken(cableGrp.id, stationName, audioId); + releaseToken(); throw new Error('Live WAV sourcePath required'); } if (!fs.existsSync(sourcePath)) { - if (cableGrp) releaseCableToken(cableGrp.id, stationName, audioId); + releaseToken(); throw new Error(`Live WAV source missing: ${sourcePath}`); } @@ -1150,7 +1232,7 @@ async function sendLiveWavChunksToStation(stationName, fallbackWs, task) { const chunkDurationSec = 0.25; const chunkBytes = Math.max(4096, Math.floor(sampleRate * channels * (bitsPerSample / 8) * chunkDurationSec)); - masterWs.send(JSON.stringify({ + sendWsToAllConnectionsOfStation(stationName, { type: 'audio_start', stationName, audioId, @@ -1163,12 +1245,12 @@ async function sendLiveWavChunksToStation(stationName, fallbackWs, task) { interruptMode: task.interruptMode || getInterruptModeForPriority(prio), source: task.source || 'unknown', cableGroupId: cableGrp?.id || '' - })); + }); const ffmpegArgs = ['-hide_banner', '-loglevel', 'error']; if (loop) ffmpegArgs.push('-stream_loop', '-1'); ffmpegArgs.push('-re', '-i', sourcePath, '-f', 's16le', '-acodec', 'pcm_s16le', '-ac', String(channels), '-ar', String(sampleRate), 'pipe:1'); - log('INFO', `Live WAV chunk stream start for ${stationName}: ${path.basename(sourcePath)}${loop ? ' (loop)' : ''} chunk=${chunkDurationSec}s`); + log('INFO', `Live WAV chunk stream start for ${stationName}: ${path.basename(sourcePath)}${loop ? ' (loop)' : ''} chunk=${chunkDurationSec}s liveTargets=${allTargets.length}`); return await new Promise((resolve, reject) => { let stderr = ''; @@ -1176,12 +1258,6 @@ async function sendLiveWavChunksToStation(stationName, fallbackWs, task) { let proc = null; let pcmPending = Buffer.alloc(0); - const releaseToken = () => { - if (cableGrp) { - try { releaseCableToken(cableGrp.id, stationName, audioId); } catch {} - } - }; - const finish = (result, error) => { if (settled) return; settled = true; @@ -1192,15 +1268,12 @@ async function sendLiveWavChunksToStation(stationName, fallbackWs, task) { return; } try { - const { ws: endWs } = getStationMaster(stationName); - if (endWs && endWs.readyState === WebSocket.OPEN) { - endWs.send(JSON.stringify({ - type: 'audio_end', - stationName, - audioId, - durationSec: result.durationSec - })); - } + sendWsToAllConnectionsOfStation(stationName, { + type: 'audio_end', + stationName, + audioId, + durationSec: result.durationSec + }); } catch {} if (cableGrp && !loop && durationSec > 0) { setTimeout(() => releaseToken(), Math.max(100, Math.floor(durationSec * 1000) + 500)); @@ -1228,15 +1301,14 @@ async function sendLiveWavChunksToStation(stationName, fallbackWs, task) { const flushChunk = (pcmChunk) => { if (!pcmChunk || !pcmChunk.length) return; - const { ws: fws } = getStationMaster(stationName); - const sendWs = fws || masterWs; - if (!sendWs || sendWs.readyState !== WebSocket.OPEN) { + const alive = getAllStationConnections(stationName); + if (alive.length === 0) { cancelStream(); return; } - try { - sendWs.send(buildWavChunkFromPcm(pcmChunk, sampleRate, channels, bitsPerSample)); + const framed = buildWavChunkFromPcm(pcmChunk, sampleRate, channels, bitsPerSample); + sendBinaryToAllConnectionsOfStation(stationName, framed); } catch (err) { finish(null, err); } @@ -4421,6 +4493,12 @@ app.use((err, req, res, next) => { server.listen(PORT, () => { loadData(); try { ensureSilenceWav(); } catch (e) { log('WARN', `ensureSilenceWav failed at startup: ${e.message}`); } + try { + const hCableWatchdog = setInterval(() => { + try { runCableTokenWatchdog(); } catch (e) { log('WARN', `cable token watchdog error: ${e.message}`); } + }, 20 * 1000); + if (typeof hCableWatchdog.unref === 'function') hCableWatchdog.unref(); + } catch {} if (!audioVersion) audioVersion = String(Date.now()); if (!Array.isArray(versions) || versions.length === 0) { const snap = snapshotFiles(); diff --git a/station_broadcast.lua b/station_broadcast.lua index ee8aceb..41cd60d 100644 --- a/station_broadcast.lua +++ b/station_broadcast.lua @@ -1,6 +1,6 @@ -- 1. Configuration & Constants -local APP_VERSION_TEXT = "v1.0.113" +local APP_VERSION_TEXT = "v1.0.114" local CONFIG = { CLOUD_URL = "ws://stabadmin.fse-media.group",