fix(node-agent): await stopDecklinkBridge and clean up stale occurrences

This commit is contained in:
Wild Dragon Dev 2026-06-04 00:54:29 +00:00
parent d1b40f5303
commit 315b31a68b

View file

@ -77,7 +77,7 @@ const DC_BOARD = process.env.DELTACAST_BOARD || '0';
const FC_URL = process.env.FC_URL || 'http://framecache:7435'; const FC_URL = process.env.FC_URL || 'http://framecache:7435';
// Node identity for framecache slot IDs (e.g. "decklink-zampp3-0"). // Node identity for framecache slot IDs (e.g. "decklink-zampp3-0").
// Set NODE_NAME in .env.worker so slot IDs are stable across restarts. // Set NODE_NAME in .env.worker so slot IDs are stable across restarts.
const FC_NODE_ID = process.env.NODE_NAME || os.hostname(); const FC_NODE_ID = process.env.NODE_NAME || process.env.HOSTNAME || 'local';
let _dcBridge = null; // ChildProcess | null let _dcBridge = null; // ChildProcess | null
let _dcSidecarCount = 0; // active deltacast sidecars on this node let _dcSidecarCount = 0; // active deltacast sidecars on this node
@ -205,6 +205,9 @@ async function startDecklinkBridge(deviceIndices) {
if (await _dlBridgeRunning()) return; if (await _dlBridgeRunning()) return;
const devCsv = Array.isArray(deviceIndices) ? deviceIndices.join(',') : String(deviceIndices || '0'); const devCsv = Array.isArray(deviceIndices) ? deviceIndices.join(',') : String(deviceIndices || '0');
const DL_IMAGE = 'wild-dragon-capture:latest';
const DL_BIN = '/usr/local/bin/decklink-bridge';
// Pass correct IP to containerized bridge. Default falls back to framecache:7435. // Pass correct IP to containerized bridge. Default falls back to framecache:7435.
const _fcUrl = process.env.FRAMECACHE_IP ? `http://${process.env.FRAMECACHE_IP}:7435` : FC_URL; const _fcUrl = process.env.FRAMECACHE_IP ? `http://${process.env.FRAMECACHE_IP}:7435` : FC_URL;
@ -213,12 +216,102 @@ async function startDecklinkBridge(deviceIndices) {
'--fc-url', _fcUrl, '--fc-url', _fcUrl,
'--audio-pipe-dir', DL_AUDIO_DIR, '--audio-pipe-dir', DL_AUDIO_DIR,
]; ];
console.log(`[dl-bridge] spawning containerized bridge for devices: ${devCsv}`);
const spec = { const spec = {
Image: DL_IMAGE, Image: DL_IMAGE,
Entrypoint: [DL_BIN], Entrypoint: [DL_BIN],
Cmd: bridgeArgs, Cmd: bridgeArgs,
Env: [`NODE_ID=${FC_NODE_ID}`, `FC_URL=${_fcUrl}`], Env: [`NODE_ID=${FC_NODE_ID}`, `FC_URL=${_fcUrl}`],
HostConfig: {
NetworkMode: 'host',
Privileged: true,
Binds: ['/dev:/dev', '/dev/shm:/dev/shm'],
RestartPolicy: { Name: 'unless-stopped' },
},
};
try {
const createRes = await dockerApi('POST', '/containers/create?name=decklink-bridge', spec);
if (createRes.status !== 201 && createRes.status !== 409) {
console.error('[dl-bridge] create failed:', createRes.data);
return;
}
const containerId = createRes.status === 409 ? 'decklink-bridge' : createRes.data.Id;
const startRes = await dockerApi('POST', `/containers/${containerId}/start`);
if (startRes.status !== 204 && startRes.status !== 304) {
console.error('[dl-bridge] start failed:', startRes.data);
return;
}
_dlBridgeId = containerId;
_attachDlBridgeLogs(containerId);
console.log(`[dl-bridge] running in container ${containerId}`);
} catch (err) {
console.error(`[dl-bridge] spawn error: ${err.message}`);
}
}
async function stopDecklinkBridge() {
if (!_dlBridgeId) return;
console.log('[dl-bridge] stopping container');
try {
await dockerApi('POST', `/containers/${_dlBridgeId}/stop?t=5`);
await dockerApi('DELETE', `/containers/${_dlBridgeId}?force=true`);
} catch (err) {
console.error(`[dl-bridge] stop error: ${err.message}`);
}
_dlBridgeId = null;
}
function _dcBridgeRunning() {
return _dcBridge !== null && _dcBridge.exitCode === null && _dcBridge.signalCode === null;
}
// Check /proc on Linux to see if a deltacast-bridge process is alive.
// Used by startDeltacastBridge() to detect a bridge started outside node-agent
// (e.g. manually with sudo, or from a prior node-agent process).
function _dcBridgeProcessAlive() {
try {
for (const pid of fs.readdirSync('/proc')) {
if (!/^\d+$/.test(pid)) continue;
try {
// cmdline is NUL-delimited; read as binary-friendly string.
const cmdline = fs.readFileSync(`/proc/${pid}/cmdline`, 'latin1');
if (cmdline.includes('deltacast-bridge')) return true;
} catch (_) { /* process may have exited mid-scan */ }
}
} catch (_) {}
return false;
}
function startDeltacastBridge() {
if (_dcBridgeRunning()) return; // already up (we spawned it)
try { fs.mkdirSync(DC_PIPE_DIR, { recursive: true }); } catch (_) {}
// FIFOs may exist from a previous run. Only skip the spawn if a
// deltacast-bridge process is actually alive on the host — stale FIFOs with
// no live writer cause ffmpeg to block on open() indefinitely (no audio/video).
const _v0 = DC_PIPE_DIR + '/video-0.fifo';
if (fs.existsSync(_v0)) {
if (_dcBridgeProcessAlive()) {
console.log('[dc-bridge] FIFOs exist and bridge process alive — skipping spawn');
return;
}
console.log('[dc-bridge] FIFOs exist but bridge is NOT running — spawning fresh bridge');
// Stale FIFOs are harmless: the bridge recreates them (mkfifo ignores EEXIST).
}
const args = [
'--device', DC_BOARD,
'--ports', DC_PORTS_CSV,
'--video-pipe-dir', DC_PIPE_DIR,
'--audio-pipe-dir', DC_PIPE_DIR,
'--fc-url', FC_URL,
];
console.log(`[dc-bridge] launching: ${DC_BRIDGE_BIN} ${args.join(' ')}`); console.log(`[dc-bridge] launching: ${DC_BRIDGE_BIN} ${args.join(' ')}`);
const proc = spawn(DC_BRIDGE_BIN, args, { const proc = spawn(DC_BRIDGE_BIN, args, {
@ -529,7 +622,7 @@ async function handleSidecarStart(body, res) {
if (_dcSidecarCount <= 0) { _dcSidecarCount = 0; stopDeltacastBridge(); } if (_dcSidecarCount <= 0) { _dcSidecarCount = 0; stopDeltacastBridge(); }
} else if (sourceType === 'sdi' || sourceType === 'blackmagic') { } else if (sourceType === 'sdi' || sourceType === 'blackmagic') {
_dlSidecarCount--; _dlSidecarCount--;
if (_dlSidecarCount <= 0) { _dlSidecarCount = 0; stopDecklinkBridge(); } if (_dlSidecarCount <= 0) { _dlSidecarCount = 0; await stopDecklinkBridge(); }
} else if (sourceType === 'srt' || sourceType === 'rtmp') { } else if (sourceType === 'srt' || sourceType === 'rtmp') {
// net_ingest may be keyed by the temp id (create not yet succeeded) or // net_ingest may be keyed by the temp id (create not yet succeeded) or
// the real containerId (remapped). Stop whichever exists. // the real containerId (remapped). Stop whichever exists.
@ -687,7 +780,7 @@ async function handleSidecarStandby(body, res) {
if (_dcSidecarCount <= 0) { _dcSidecarCount = 0; stopDeltacastBridge(); } if (_dcSidecarCount <= 0) { _dcSidecarCount = 0; stopDeltacastBridge(); }
} else if (sourceType === 'sdi' || sourceType === 'blackmagic') { } else if (sourceType === 'sdi' || sourceType === 'blackmagic') {
_dlSidecarCount--; _dlSidecarCount--;
if (_dlSidecarCount <= 0) { _dlSidecarCount = 0; stopDecklinkBridge(); } if (_dlSidecarCount <= 0) { _dlSidecarCount = 0; await stopDecklinkBridge(); }
} }
}; };
@ -746,7 +839,7 @@ async function handleSidecarStop(containerId, res) {
_dlSidecarCount--; _dlSidecarCount--;
if (_dlSidecarCount <= 0) { if (_dlSidecarCount <= 0) {
_dlSidecarCount = 0; _dlSidecarCount = 0;
stopDecklinkBridge(); await stopDecklinkBridge();
} }
} else if (_srcType === 'srt' || _srcType === 'rtmp') { } else if (_srcType === 'srt' || _srcType === 'rtmp') {
stopNetIngest(containerId); stopNetIngest(containerId);