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/.
274 lines
12 KiB
TypeScript
274 lines
12 KiB
TypeScript
import { closeSync, mkdirSync, rmSync } from "node:fs";
|
||
import { join } from "node:path";
|
||
import { resolveHelperBinary } from "../gate/binary";
|
||
import { createFifoPair, makeFifoPair, openBlockingReader, type FifoPair } from "../gate/fifo";
|
||
import { assertGeometry, type FrameGeometry } from "../gate/protocol";
|
||
import { spawnManaged, type ManagedProcess, type StderrRedactor } from "../relay/ffmpeg";
|
||
|
||
/**
|
||
* Per-source fan-out between one ingest and every consumer of the source's raw
|
||
* frames: the compositors that use it (two during an A/B swap) and its preview
|
||
* taps.
|
||
*
|
||
* The bytes never pass through Node. A `source-bus` helper process
|
||
* (src/gate/rust/src/bin/source-bus.rs) reads the ingest's FIFOs and writes
|
||
* each subscriber's FIFOs; this class is its process manager and speaks its
|
||
* control lines. (S2a moved the frames on the worker's event loop: at
|
||
* 4 × 1080p30 that left no headroom and the buses dropped ~7.5 % of source
|
||
* pairs; runs/wp4/s2b-report.md item 1.)
|
||
*
|
||
* Semantics are unchanged from S2a: frames are delivered as A/V *pairs*, a
|
||
* subscriber more than `maxBacklogFrames + historyFrames` behind loses whole
|
||
* pairs, and the last `historyFrames` pairs are replayed to a new subscriber so
|
||
* the gate can splice a swap-in compositor at the on-air one's next source frame.
|
||
*/
|
||
export interface SourceBusOptions {
|
||
geometry: FrameGeometry;
|
||
scratchDir: string;
|
||
redactor: StderrRedactor;
|
||
onStderr(message: string): void;
|
||
/** The helper died (not on `stop()`): the source is gone. */
|
||
onExit?(reason: string): void;
|
||
maxSkewFrames?: number;
|
||
maxBacklogFrames?: number;
|
||
historyFrames?: number;
|
||
}
|
||
|
||
export interface SubscribeOptions {
|
||
video: boolean;
|
||
audio: boolean;
|
||
/** Take every `stride`-th pair only (thumbnails). Default 1. */
|
||
stride?: number;
|
||
}
|
||
|
||
/** A subscription: the pair index it starts at, and the pairs it has lost to backlog since. */
|
||
export interface BusSubscription {
|
||
readonly bus: SourceBus;
|
||
readonly id: number;
|
||
readonly startPair: number;
|
||
readonly startedAt: bigint;
|
||
readonly dropped: number;
|
||
/** Blocking read ends for the consumer's stdio (null when not subscribed to that stream). */
|
||
readonly videoFd: number | null;
|
||
readonly audioFd: number | null;
|
||
/** Pairs `other` had lost when this subscription started (exact, unlike its live `dropped`). */
|
||
droppedBefore(other: BusSubscription): number | undefined;
|
||
/** Close the worker's copies of the read ends (after spawning the consumer). */
|
||
releaseReaders(): void;
|
||
/** The caller takes ownership of the read ends (an in-process consumer that closes them itself). */
|
||
adoptReaders(): void;
|
||
unsubscribe(): void;
|
||
}
|
||
|
||
export interface SourceBusStats {
|
||
pairs: number; silenceFilled: number; audioTrimmed: number; subscriberDrops: number; allocs: number;
|
||
input: { videoWaiting: number; audioWaiting: number; videoEof: boolean; audioEof: boolean };
|
||
/** Per subscriber: pairs dropped to backlog, and frames queued in the helper right now. */
|
||
subscribers: Array<{ id: number; dropped: number; queued: number }>;
|
||
}
|
||
|
||
interface Waiter { match(line: string[]): boolean; resolve(line: string[]): void; reject(error: Error): void; timer: ReturnType<typeof setTimeout> }
|
||
|
||
let nextBus = 1;
|
||
|
||
export class SourceBus {
|
||
private proc: ManagedProcess | null = null;
|
||
private readonly busId = nextBus++;
|
||
private nextSub = 1;
|
||
private lineBuf = "";
|
||
private waiters: Waiter[] = [];
|
||
private readonly drops = new Map<number, number>();
|
||
private stopping = false;
|
||
private exited = false;
|
||
private ingest: FifoPair | null = null;
|
||
readonly maxSkewFrames: number;
|
||
readonly maxBacklogFrames: number;
|
||
readonly historyFrames: number;
|
||
/** Set when the helper reports the ingest closed both streams. */
|
||
ended = false;
|
||
|
||
constructor(private readonly options: SourceBusOptions) {
|
||
assertGeometry(options.geometry);
|
||
this.maxSkewFrames = options.maxSkewFrames ?? 15;
|
||
this.maxBacklogFrames = options.maxBacklogFrames ?? 4;
|
||
this.historyFrames = options.historyFrames ?? 12;
|
||
mkdirSync(options.scratchDir, { recursive: true, mode: 0o700 });
|
||
}
|
||
|
||
get geometry(): FrameGeometry { return this.options.geometry; }
|
||
get pid(): number | undefined { return this.proc?.child.pid; }
|
||
|
||
/**
|
||
* Starts the helper on fresh ingest FIFOs. The returned fds are blocking write ends for the
|
||
* producer's stdio (ingest: video on fd 1, PCM on fd 3); call `releaseIngestWriters()` once
|
||
* the producer has them. Until a producer writes, the bus simply has no pairs.
|
||
*/
|
||
async start(): Promise<{ videoFd: number; audioFd: number }> {
|
||
const g = this.options.geometry;
|
||
const fifos = createFifoPair(this.options.scratchDir, this.busId, "bus");
|
||
this.ingest = fifos;
|
||
const argv = [String(g.width), String(g.height), String(g.fps), String(this.maxSkewFrames), String(this.maxBacklogFrames), String(this.historyFrames),
|
||
fifos.videoPath, fifos.audioPath];
|
||
const proc = spawnManaged(resolveHelperBinary("source-bus"), argv, ["pipe", "pipe", "pipe"], this.options.redactor, this.options.onStderr);
|
||
this.proc = proc;
|
||
proc.child.stdout?.setEncoding("utf8");
|
||
proc.child.stdout?.on("data", (chunk: string) => this.onData(chunk));
|
||
// EPIPE on a control line to a helper that just died: its exit is reported below, never as a crash.
|
||
proc.child.stdin?.on("error", () => undefined);
|
||
void proc.exited.then(({ code, signal }) => {
|
||
this.exited = true;
|
||
const reason = `source bus exited code=${code} signal=${signal}`;
|
||
for (const w of this.waiters.splice(0)) { clearTimeout(w.timer); w.reject(new Error(reason)); }
|
||
if (!this.stopping) this.options.onExit?.(reason);
|
||
});
|
||
const ready = this.waitFor((l) => l[0] === "READY", 10_000);
|
||
try {
|
||
await proc.started;
|
||
await ready;
|
||
} catch (error) {
|
||
ready.catch(() => undefined);
|
||
fifos.releaseHolders();
|
||
fifos.releaseWriters();
|
||
fifos.unlink();
|
||
throw error;
|
||
} finally {
|
||
// The helper holds the read ends now; from here only the producer keeps the FIFOs open.
|
||
fifos.releaseHolders();
|
||
fifos.unlink();
|
||
}
|
||
return { videoFd: fifos.videoFd, audioFd: fifos.audioFd };
|
||
}
|
||
|
||
/** Close the worker's copies of the ingest write ends (after spawning the producer). */
|
||
releaseIngestWriters(): void { this.ingest?.releaseWriters(); }
|
||
|
||
/** The caller takes ownership of the ingest write ends (an in-process producer that closes them itself). */
|
||
adoptIngestWriters(): void { this.ingest = null; }
|
||
|
||
private onData(chunk: string): void {
|
||
this.lineBuf += chunk;
|
||
let nl = this.lineBuf.indexOf("\n");
|
||
while (nl >= 0) {
|
||
const line = this.lineBuf.slice(0, nl).trim().split(/\s+/);
|
||
this.lineBuf = this.lineBuf.slice(nl + 1);
|
||
nl = this.lineBuf.indexOf("\n");
|
||
if (line[0] === "DROP") this.drops.set(Number(line[1]), Number(line[2]));
|
||
if (line[0] === "EOF") this.ended = true;
|
||
const i = this.waiters.findIndex((w) => w.match(line));
|
||
if (i >= 0) {
|
||
const [w] = this.waiters.splice(i, 1);
|
||
if (w) { clearTimeout(w.timer); w.resolve(line); }
|
||
}
|
||
}
|
||
}
|
||
|
||
private waitFor(match: (line: string[]) => boolean, timeoutMs: number): Promise<string[]> {
|
||
return new Promise<string[]>((resolve, reject) => {
|
||
const waiter: Waiter = {
|
||
match, resolve, reject,
|
||
timer: setTimeout(() => { this.waiters = this.waiters.filter((w) => w !== waiter); reject(new Error("source bus reply timeout")); }, timeoutMs),
|
||
};
|
||
this.waiters.push(waiter);
|
||
});
|
||
}
|
||
|
||
private send(line: string): void {
|
||
const stdin = this.proc?.child.stdin;
|
||
if (this.exited || !stdin || stdin.destroyed) throw new Error("source bus is not running");
|
||
stdin.write(line);
|
||
}
|
||
|
||
/** Sends a control line and waits for its reply; a failed send never leaves a waiter behind. */
|
||
private request(line: string, match: (l: string[]) => boolean, timeoutMs: number): Promise<string[]> {
|
||
const reply = this.waitFor(match, timeoutMs);
|
||
try {
|
||
this.send(line);
|
||
} catch (error) {
|
||
reply.catch(() => undefined);
|
||
const waiter = this.waiters.at(-1);
|
||
if (waiter) { clearTimeout(waiter.timer); this.waiters.pop(); }
|
||
throw error;
|
||
}
|
||
return reply;
|
||
}
|
||
|
||
/**
|
||
* New consumer. Delivery starts with the replayed history (a strided subscriber gets only the
|
||
* newest pair), then live pairs; `startPair` is the first pair's index. Spawn the consumer with
|
||
* `videoFd`/`audioFd`, then `releaseReaders()`.
|
||
*/
|
||
async subscribe(opts: SubscribeOptions): Promise<BusSubscription> {
|
||
if (!opts.video && !opts.audio) throw new Error("subscription needs video or audio");
|
||
const stride = opts.stride ?? 1;
|
||
if (!Number.isSafeInteger(stride) || stride < 1 || stride > 3600) throw new Error("stride out of range");
|
||
const id = this.nextSub++;
|
||
const paths = makeFifoPair(this.options.scratchDir, id, `sub-${this.busId}`);
|
||
// The consumer (ffmpeg) needs *blocking* read ends; a temporary writer keeps each open() from
|
||
// blocking and is released once the helper has opened its non-blocking write end.
|
||
const video = opts.video ? openBlockingReader(paths.videoPath) : null;
|
||
const audio = opts.audio ? openBlockingReader(paths.audioPath) : null;
|
||
const unlink = (): void => { rmSync(paths.videoPath, { force: true }); rmSync(paths.audioPath, { force: true }); };
|
||
let line: string[];
|
||
try {
|
||
line = await this.request(`SUB ${id} ${video ? paths.videoPath : "-"} ${audio ? paths.audioPath : "-"} ${stride}\n`,
|
||
(l) => (l[0] === "SUBSCRIBED" || l[0] === "ERROR") && Number(l[1]) === id, 5_000);
|
||
if (line[0] !== "SUBSCRIBED") throw new Error("source bus could not open subscriber fifos");
|
||
} catch (error) {
|
||
for (const r of [video, audio]) if (r) { r.releaseHolder(); closeSync(r.fd); }
|
||
throw error;
|
||
} finally {
|
||
video?.releaseHolder();
|
||
audio?.releaseHolder();
|
||
unlink();
|
||
}
|
||
const peers = new Map<number, number>();
|
||
if (line[3] && line[3] !== "-") {
|
||
for (const entry of line[3].split(",")) {
|
||
const [peer, dropped] = entry.split(":").map(Number);
|
||
if (peer !== undefined && dropped !== undefined) peers.set(peer, dropped);
|
||
}
|
||
}
|
||
const drops = this.drops;
|
||
let readersOpen = true;
|
||
const subscription: BusSubscription = {
|
||
bus: this, id, startPair: Number(line[2]), startedAt: process.hrtime.bigint(),
|
||
get dropped() { return drops.get(id) ?? 0; },
|
||
videoFd: video?.fd ?? null, audioFd: audio?.fd ?? null,
|
||
droppedBefore: (other) => (other.bus === this ? peers.get(other.id) : undefined),
|
||
releaseReaders() { if (readersOpen) { readersOpen = false; for (const r of [video, audio]) if (r) closeSync(r.fd); } },
|
||
adoptReaders() { readersOpen = false; },
|
||
unsubscribe: () => {
|
||
subscription.releaseReaders();
|
||
drops.delete(id);
|
||
try { this.send(`UNSUB ${id}\n`); } catch { /* helper gone */ }
|
||
},
|
||
};
|
||
return subscription;
|
||
}
|
||
|
||
async stats(): Promise<SourceBusStats> {
|
||
const line = await this.request("STATS\n", (l) => l[0] === "STATS", 5_000);
|
||
const kv = new Map(line.slice(1).map((p) => p.split("=") as [string, string]));
|
||
const n = (k: string): number => Number(kv.get(k) ?? 0);
|
||
return { pairs: n("pairs"), silenceFilled: n("silence_filled"), audioTrimmed: n("audio_trimmed"), subscriberDrops: n("subscriber_drops"), allocs: n("allocs"),
|
||
input: { videoWaiting: n("video_waiting"), audioWaiting: n("audio_waiting"), videoEof: n("video_eof") !== 0, audioEof: n("audio_eof") !== 0 },
|
||
subscribers: (kv.get("subs") ?? "-").split(",").filter((e) => e !== "-").map((e) => {
|
||
const [id, dropped, queued] = e.split(":").map(Number);
|
||
return { id: id ?? 0, dropped: dropped ?? 0, queued: queued ?? 0 };
|
||
}) };
|
||
}
|
||
|
||
async stop(): Promise<void> {
|
||
this.stopping = true;
|
||
const proc = this.proc;
|
||
this.ingest?.releaseWriters();
|
||
if (!proc) return;
|
||
try { this.send("QUIT\n"); } catch { /* already gone */ }
|
||
const timer = setTimeout(() => proc.kill("SIGKILL"), 2_000);
|
||
await proc.exited;
|
||
clearTimeout(timer);
|
||
this.proc = null;
|
||
rmSync(join(this.options.scratchDir, `bus-${this.busId}.v.fifo`), { force: true });
|
||
rmSync(join(this.options.scratchDir, `bus-${this.busId}.a.fifo`), { force: true });
|
||
}
|
||
}
|