Henry_Du/CC-StationBroadcast
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(audio): 修复线缆 token 死锁(holder=CE-01) + 同组 approaching 等优先级饿死 + 线缆太长前端主机听不到

Browse Source 
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
This commit is contained in:
Henry_Du
2026-08-17 19:48:36 +08:00
parent dc8f94ce22
commit 9c5d6e9659
2 changed files with 152 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CloudServer/server.js
+151 -73
View File
@@ -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();
station_broadcast.lua
+1 -1
View File
@@ -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",