glass beadian
aa5e262f3a
Self-hosted multi-platform broadcast newsroom for activist collectives: publishing pipeline with group approval and DST-proof scheduling, TweetDeck-style column wall with hostile-content-safe rendering, live control room with Auto Director, bridged chat governance with polls and quorum, distributed sandboxed transcode workers, encrypted content-addressed archive, monthly analytics. MIT licensed. See README.md and docs/.
65 lines
4.2 KiB
JavaScript
65 lines
4.2 KiB
JavaScript
'use strict';
|
|
// Both WinSW and interactive validation use this supervisor. Only official IPC
|
|
// drains the runner; loss of admission never kills an in-flight video job.
|
|
const fs = require('node:fs');
|
|
const path = require('node:path');
|
|
const { spawn, execFile } = require('node:child_process');
|
|
const root = process.env.PT_ROOT || 'C:\\pt-worker';
|
|
const state = path.join(root, 'state');
|
|
const cfg = JSON.parse(fs.readFileSync(path.join(state, 'worker.json'), 'utf8').replace(/^\uFEFF/, ''));
|
|
const env = { ...process.env, PT_ROOT: root, PT_STATE_DIR: state,
|
|
PT_WORKER_CONFIG: path.join(state, 'worker.json'),
|
|
APPDATA: path.join(root, 'profile', 'Roaming'), LOCALAPPDATA: path.join(root, 'profile', 'Local'),
|
|
TMP: path.join(root, 'tmp'), TEMP: path.join(root, 'tmp'),
|
|
FFMPEG_PATH: path.join(root, 'bin', 'ffmpeg-nvenc.exe'),
|
|
FFPROBE_PATH: path.join(root, 'bin', 'ffprobe.exe'), PT_NVENC_DEVICE: String(cfg.nvenc_device || 0) };
|
|
let child = null, draining = false, asking = false, stopping = false, restartAfter = 0;
|
|
function read(name) { try { return JSON.parse(fs.readFileSync(path.join(state, name), 'utf8').replace(/^\uFEFF/, '')); } catch { return {}; } }
|
|
function emit(data) { fs.appendFileSync(path.join(state, 'events.jsonl'), JSON.stringify({ time: Date.now()/1000, ...data }) + '\n'); }
|
|
function atomic(name, data) { const file = path.join(state, name), tmp = file+'.'+process.pid+'.tmp'; fs.writeFileSync(tmp, JSON.stringify(data)); fs.renameSync(tmp, file); }
|
|
function drain() {
|
|
if (!child || asking || draining) return;
|
|
const target = child;
|
|
asking = true;
|
|
execFile(cfg.node_binary, [path.join(root, 'bin', 'ipc-nvenc.cjs'), 'graceful-shutdown'],
|
|
{ env, timeout: 17000, windowsHide: true }, err => { asking = false; if (!err && child === target) draining = true; });
|
|
}
|
|
function tick() {
|
|
if (fs.existsSync(path.join(state, 'supervisor-stop'))) stopping = true;
|
|
const now = Date.now()/1000, admission = read('status.json'), control = read('control.json');
|
|
const allow = !stopping && control.paused === false && admission.admission_allowed === true &&
|
|
now - admission.updated_at < 30 && ['AVAILABLE', 'BUSY'].includes(admission.status) &&
|
|
!fs.existsSync(path.join(state, 'gpu-unavailable'));
|
|
if (child && !allow) drain();
|
|
if (!child && allow && now >= restartAfter) {
|
|
draining = false;
|
|
child = spawn(cfg.node_binary, ['--require', path.join(root, 'bin', 'observe.cjs'),
|
|
path.join(root, 'runner', 'node_modules', '@peertube', 'peertube-runner', 'dist', 'peertube-runner.js'),
|
|
'server', '--enable-job', 'vod-web-video-transcoding', '--enable-job', 'vod-hls-transcoding',
|
|
'--enable-job', 'vod-audio-merge-transcoding'], { env, windowsHide: true, stdio: ['ignore', 'pipe', 'pipe'] });
|
|
// Upstream logs can contain tokens. Retain only known safe milestones.
|
|
let buffer = '';
|
|
for (const stream of [child.stdout, child.stderr]) stream.on('data', chunk => {
|
|
buffer += chunk.toString();
|
|
const lines = buffer.split(/\r?\n/); buffer = lines.pop().slice(-8192);
|
|
for (const line of lines) {
|
|
const match = line.match(/(Finished processing|Processing) job of type ([\w-]+): ([a-f\d-]{36})/);
|
|
if (match) emit({ kind: 'runner_milestone', phase: match[1].startsWith('Finished') ? 'completed' : 'processing', job_type: match[2], uuid: match[3] });
|
|
if (line.includes('Connected to ') && line.includes('/runners')) emit({kind: 'runner_connected'});
|
|
if (line.includes('Cannot process job')) emit({kind: 'runner_error', reason: 'process_job_failed'});
|
|
}
|
|
});
|
|
child.once('error', () => { emit({kind: 'runner_error', reason: 'spawn_failed'}); });
|
|
child.once('exit', code => {
|
|
emit({kind: 'runner_exit', code, graceful_requested: draining || asking});
|
|
child = null; draining = false; asking = false; restartAfter = Date.now()/1000 + 15;
|
|
});
|
|
emit({kind: 'runner_start', concurrency: cfg.concurrency});
|
|
}
|
|
atomic('runner.json', {running: !!child, draining, updated_at: now, pid: child?.pid || null});
|
|
if (stopping && !child) process.exit(0);
|
|
}
|
|
fs.rmSync(path.join(state, 'supervisor-stop'), {force: true});
|
|
process.on('SIGINT', () => { stopping = true; drain(); });
|
|
process.on('SIGTERM', () => { stopping = true; drain(); });
|
|
setInterval(tick, 1000); tick();
|