From 9c5d6e96597c36ed8459227bf6017ee539333864 Mon Sep 17 00:00:00 2001 From: HenryDu8133 <813367384@qq.com> Date: Mon, 17 Aug 2026 19:48:36 +0800 Subject: [PATCH] =?UTF-8?q?fix(audio):=20=E4=BF=AE=E5=A4=8D=E7=BA=BF?= =?UTF-8?q?=E7=BC=86=20token=20=E6=AD=BB=E9=94=81(holder=3DCE-01)=20+=20?= =?UTF-8?q?=E5=90=8C=E7=BB=84=20approaching=20=E7=AD=89=E4=BC=98=E5=85=88?= =?UTF-8?q?=E7=BA=A7=E9=A5=BF=E6=AD=BB=20+=20=E7=BA=BF=E7=BC=86=E5=A4=AA?= =?UTF-8?q?=E9=95=BF=E5=89=8D=E7=AB=AF=E4=B8=BB=E6=9C=BA=E5=90=AC=E4=B8=8D?= =?UTF-8?q?=E5=88=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. releaseCableToken(L688): 去掉 audioId 严格匹配, 只要 stationName 对就释放 修复: CE-01 因 audioId 在 setTimeout/release 分支里 miss-match, token 永远不释放, 导致同组其它站 approaching 100% 被 deny reason=lower_priority holder=CE-01 2. acquireCableToken(L549): 新增 equal-prio-timeout 抢占, 同一线缆分组多站连续 approaching(p=2) 时不会再出现 holder 永远持住的死锁: 旧 token 持有超过 max(30s, durHint+30s) 后, 等优先级新请求可强制切换 (日志 WARN: equal-prio switch) 3. Cable token watchdog(每20s扫描): holder 站点所有连接断开立刻强制释放 (holder_offline); 最长持有 10min 兜底强制释放 (max_hold_timeout). 彻底避免 音乐循环/循环临时广播 崩溃后 token 泄漏永远释放不了的情况 4. 线缆太长前端主机听不到 (架构修复): 旧方案: sendAudioBufferToStation / sendLiveWavChunksToStation 里全部 WS 消息 (audio_start + audio_chunk + audio_end) 只发给站点唯一 masterWs 连接, 若 master 物理位置在线缆尾部, 中间节点断链, 线缆最前面主机就完全收不到音频 新方案: 新增 sendWsToAllConnectionsOfStation + sendBinaryToAllConnectionsOfStation (L503-L520), 所有 (audio_start/end, cable_token_granted, audio_preempt) JSON 消息 及所有 binary WAV 分块, 广播到该站点 getAllStationConnections 返回的 *所有* 已连接 WS node, 每台主机独立从云端服务器拿一份流, 不再依赖线缆上 master 转发 对 10 站长线缆场景, 只要 node 还连着云端, 就肯定能收到音频, 彻底避免前端 主机因线缆中段节点断网/掉电听不到音频的故障 端到端单元测试 5/5 PASS: A releaseCableToken audioId 不严格匹配 PASS B p=2 approaching dur=8s age=70s 时 equal-prio 切换 PASS B2 p=2 age=5s 时不抢占 PASS (防抖动) C watchdog holder_offline 检测后立刻释放 PASS D 同站 4 个 node 同时都收到 audio_start 消息 PASS (前端主机不漏消息) v1.0.114 --- CloudServer/server.js | 224 ++++++++++++++++++++++++++++-------------- station_broadcast.lua | 2 +- 2 files changed, 152 insertions(+), 74 deletions(-) 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",