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/.
249 lines
12 KiB
TypeScript
249 lines
12 KiB
TypeScript
import type { KeyObject } from "node:crypto";
|
|
import { sandboxRelay, supervisorSandboxEnabled } from "./sandbox";
|
|
import { verifyCommand } from "../bus/envelope";
|
|
import { DurableFence } from "../bus/fence";
|
|
import type { ObservedStore } from "../bus/redis";
|
|
import type { ChannelState, DesiredChannel, ObservedChannel } from "../types";
|
|
|
|
export interface TransitionLogger {
|
|
transition(event: { workspaceId: string; channelId: string; from: ChannelState; to: ChannelState; reason: string }): void;
|
|
}
|
|
interface ChannelRuntime {
|
|
desired: DesiredChannel;
|
|
state: ChannelState;
|
|
reason: string;
|
|
realizedVersion: number | null;
|
|
relay: RelayJob | null;
|
|
retries: number;
|
|
retryTimer: { cancel(): void } | null;
|
|
/** Consecutive failed swaps for the current desired version (the previous version stays on air). */
|
|
swapFailures: number;
|
|
}
|
|
export interface RelayJob {
|
|
start(channel: DesiredChannel): Promise<void>;
|
|
stop(): Promise<void>;
|
|
/**
|
|
* Optional in-place change (production jobs: an A/B compositor swap behind the
|
|
* frame gate). Resolves true once the new version is on air; false means the
|
|
* change needs a restart (sources, output or profile changed).
|
|
*/
|
|
reconfigure?(channel: DesiredChannel): Promise<boolean>;
|
|
}
|
|
export interface RetryScheduler { schedule(action: () => Promise<void>, delayMs: number): { cancel(): void } }
|
|
|
|
/**
|
|
* A swap that fails (B never went on air; A is still on air) is retried this many times, 1 s then
|
|
* 2 s later, before the supervisor falls back to a full restart. The full restart converges too,
|
|
* but it takes the encoder and the push session down (WP9 "output that does not stop").
|
|
*/
|
|
export const SWAP_RETRIES = 2;
|
|
const defaultRetryScheduler: RetryScheduler = {
|
|
schedule(action, delayMs) {
|
|
const timer = setTimeout(() => { void action(); }, delayMs);
|
|
return { cancel: () => clearTimeout(timer) };
|
|
},
|
|
};
|
|
|
|
/** Accepted versions are durable; realized versions describe only this process lifetime. */
|
|
export class MediaSupervisor {
|
|
private readonly channels = new Map<string, ChannelRuntime>();
|
|
private readonly operations = new Map<string, Promise<unknown>>();
|
|
private heartbeatTimer: ReturnType<typeof setInterval> | null = null;
|
|
/** Set by `stop()`: late retries and relaunches become no-ops. */
|
|
private stopped = false;
|
|
constructor(
|
|
private readonly workerId: string,
|
|
desired: DesiredChannel[],
|
|
private readonly store: ObservedStore,
|
|
private readonly fence: DurableFence,
|
|
private readonly controlKeys: ReadonlyMap<string, KeyObject>,
|
|
private readonly createRelay: (channel: DesiredChannel, onExit: (reason: string) => void) => RelayJob,
|
|
private readonly logger: TransitionLogger,
|
|
private readonly retryScheduler: RetryScheduler = defaultRetryScheduler,
|
|
) {
|
|
supervisorSandboxEnabled(); // Production cannot silently opt out of confinement.
|
|
for (const channel of desired) {
|
|
if (channel.workerId !== workerId) throw new Error("config worker mismatch");
|
|
this.channels.set(this.key(channel.workspaceId, channel.channelId), {
|
|
desired: channel, state: "STOPPED", reason: "worker boot", realizedVersion: null,
|
|
relay: null, retries: 0, retryTimer: null, swapFailures: 0,
|
|
});
|
|
}
|
|
}
|
|
|
|
private key(workspaceId: string, channelId: string): string { return `${workspaceId}:${channelId}`; }
|
|
status(): ObservedChannel[] {
|
|
return [...this.channels.values()].map((runtime) => ({
|
|
workspaceId: runtime.desired.workspaceId, channelId: runtime.desired.channelId,
|
|
workerId: this.workerId, assignmentEpoch: runtime.desired.assignmentEpoch,
|
|
state: runtime.state, realizedVersion: runtime.realizedVersion,
|
|
updatedAt: new Date().toISOString(), reason: runtime.reason,
|
|
}));
|
|
}
|
|
private async transition(runtime: ChannelRuntime, to: ChannelState, reason: string): Promise<void> {
|
|
const from = runtime.state;
|
|
runtime.state = to;
|
|
runtime.reason = reason;
|
|
this.logger.transition({ workspaceId: runtime.desired.workspaceId, channelId: runtime.desired.channelId, from, to, reason });
|
|
await this.store.put(this.status().find((item) => item.workspaceId === runtime.desired.workspaceId && item.channelId === runtime.desired.channelId)!);
|
|
}
|
|
|
|
async start(): Promise<void> {
|
|
for (const runtime of this.channels.values()) {
|
|
const old = await this.store.get(runtime.desired.workspaceId, runtime.desired.channelId);
|
|
await this.transition(runtime, "STOPPED", old
|
|
? `restart reconcile: prior ${old.state}; awaiting fresh signed snapshot`
|
|
: "start reconcile: awaiting signed snapshot");
|
|
}
|
|
await this.beat();
|
|
this.heartbeatTimer = setInterval(() => { void this.beat().catch((error: unknown) => this.logger.transition({ workspaceId: "", channelId: "", from: "DEGRADED", to: "DEGRADED", reason: `heartbeat failed: ${String(error)}` })); }, 5_000);
|
|
}
|
|
|
|
async beat(): Promise<void> { await this.store.heartbeat(this.workerId, this.status()); }
|
|
|
|
async handle(raw: unknown, now = Date.now()): Promise<"accepted" | "replay" | "stale"> {
|
|
const command = verifyCommand(raw, this.controlKeys, this.workerId, now);
|
|
const body = command.body;
|
|
return this.serialize(this.key(body.workspaceId, body.channelId), () => this.apply(body, now));
|
|
}
|
|
|
|
/** One operation at a time per channel (commands and swap retries). */
|
|
private serialize<T>(key: string, operation: () => Promise<T>): Promise<T> {
|
|
const previous = this.operations.get(key) ?? Promise.resolve();
|
|
const task = previous.catch(() => undefined).then(operation);
|
|
this.operations.set(key, task);
|
|
void task.then(() => { if (this.operations.get(key) === task) this.operations.delete(key); },
|
|
() => { if (this.operations.get(key) === task) this.operations.delete(key); });
|
|
return task;
|
|
}
|
|
|
|
private async apply(body: ReturnType<typeof verifyCommand>["body"], now: number): Promise<"accepted" | "replay" | "stale"> {
|
|
const runtime = this.channels.get(this.key(body.workspaceId, body.channelId));
|
|
if (!runtime || runtime.desired.assignmentEpoch !== body.assignmentEpoch) throw new Error("unassigned channel or epoch");
|
|
const result = this.fence.accept(body, now);
|
|
if (result !== "accepted") return result;
|
|
// RUNNING → RUNNING on a job that can swap: the encoder and push session stay up,
|
|
// and the version is realized only once the new compositor is on air.
|
|
const relay = runtime.relay;
|
|
if (runtime.state === "RUNNING" && body.snapshot.desiredStatus === "RUNNING" && relay?.reconfigure) {
|
|
runtime.desired = body.snapshot;
|
|
runtime.retryTimer?.cancel();
|
|
runtime.retryTimer = null;
|
|
runtime.swapFailures = 0;
|
|
if (await this.swap(runtime, relay) !== "restart") return result;
|
|
}
|
|
runtime.desired = body.snapshot;
|
|
runtime.swapFailures = 0;
|
|
runtime.realizedVersion = null;
|
|
runtime.retryTimer?.cancel();
|
|
runtime.retryTimer = null;
|
|
await runtime.relay?.stop();
|
|
runtime.relay = null;
|
|
if (body.snapshot.desiredStatus === "STOPPED") {
|
|
await this.transition(runtime, "STOPPED", "signed desired state stopped");
|
|
runtime.realizedVersion = body.desiredVersion;
|
|
await this.beat();
|
|
return result;
|
|
}
|
|
runtime.retries = 0;
|
|
await this.launch(runtime);
|
|
return result;
|
|
}
|
|
|
|
/**
|
|
* One swap attempt for `runtime.desired`. "done": on air (or the job is gone and its own exit
|
|
* path owns the channel); "retrying": B failed, A stays on air (`realizedVersion` keeps naming
|
|
* what viewers see) and a retry is scheduled; "restart": not swappable, or out of retries.
|
|
*/
|
|
private async swap(runtime: ChannelRuntime, relay: RelayJob): Promise<"done" | "retrying" | "restart"> {
|
|
const version = runtime.desired.desiredVersion;
|
|
try {
|
|
if (!(await relay.reconfigure?.(runtime.desired))) return "restart";
|
|
if (runtime.relay !== relay) return "done";
|
|
runtime.swapFailures = 0;
|
|
runtime.realizedVersion = version;
|
|
await this.transition(runtime, "RUNNING", "compositor swapped at a frame boundary");
|
|
await this.beat();
|
|
return "done";
|
|
} catch (error: unknown) {
|
|
if (runtime.relay !== relay) return "done";
|
|
const name = error instanceof Error ? error.name : "unknown error";
|
|
runtime.swapFailures += 1;
|
|
if (runtime.swapFailures > SWAP_RETRIES) {
|
|
runtime.reason = `swap failed: ${name}`;
|
|
return "restart";
|
|
}
|
|
const delay = 1_000 * 2 ** (runtime.swapFailures - 1);
|
|
await this.transition(runtime, "RUNNING", `swap to v${version} failed (${name}); v${runtime.realizedVersion ?? "?"} stays on air; retry ${runtime.swapFailures}/${SWAP_RETRIES} in ${delay}ms`);
|
|
const key = this.key(runtime.desired.workspaceId, runtime.desired.channelId);
|
|
runtime.retryTimer = this.retryScheduler.schedule(() => this.serialize(key, async () => {
|
|
runtime.retryTimer = null;
|
|
if (this.stopped) return;
|
|
// A newer command or an exit has taken over since.
|
|
if (runtime.desired.desiredVersion !== version || runtime.relay !== relay || runtime.state !== "RUNNING") return;
|
|
if (await this.swap(runtime, relay) === "restart") await this.restart(runtime);
|
|
}), delay);
|
|
return "retrying";
|
|
}
|
|
}
|
|
|
|
/** Full restart for the current desired snapshot (the swap path gave up). */
|
|
private async restart(runtime: ChannelRuntime): Promise<void> {
|
|
runtime.realizedVersion = null;
|
|
runtime.swapFailures = 0;
|
|
await runtime.relay?.stop();
|
|
runtime.relay = null;
|
|
runtime.retries = 0;
|
|
await this.launch(runtime);
|
|
}
|
|
|
|
private async launch(runtime: ChannelRuntime): Promise<void> {
|
|
await this.transition(runtime, runtime.retries ? "RESTARTING" : "STARTING", runtime.retries ? "relay retry" : "signed desired state running");
|
|
try {
|
|
const relay = sandboxRelay(this.createRelay(runtime.desired, (reason) => {
|
|
if (runtime.relay === relay) void this.relayExited(runtime, reason);
|
|
}), runtime.desired);
|
|
runtime.relay = relay;
|
|
await relay.start(runtime.desired);
|
|
if (runtime.relay !== relay) return;
|
|
runtime.realizedVersion = runtime.desired.desiredVersion;
|
|
await this.transition(runtime, "RUNNING", "ffmpeg spawned with local relay pipes");
|
|
await this.beat();
|
|
} catch (error: unknown) {
|
|
runtime.relay = null;
|
|
await this.relayExited(runtime, `start failed: ${error instanceof Error ? error.name : "unknown error"}`);
|
|
}
|
|
}
|
|
|
|
private async relayExited(runtime: ChannelRuntime, reason: string): Promise<void> {
|
|
if (this.stopped || runtime.desired.desiredStatus !== "RUNNING") return;
|
|
runtime.realizedVersion = null;
|
|
runtime.relay = null;
|
|
await this.transition(runtime, "DEGRADED", reason);
|
|
if (runtime.retries >= 5) { await this.transition(runtime, "FAILED", "relay retry budget exhausted"); return; }
|
|
const delay = Math.min(30_000, 1_000 * 2 ** runtime.retries);
|
|
runtime.retries += 1;
|
|
await this.transition(runtime, "RESTARTING", `retry in ${delay}ms`);
|
|
runtime.retryTimer = this.retryScheduler.schedule(() => { runtime.retryTimer = null; return this.stopped ? Promise.resolve() : this.launch(runtime); }, delay);
|
|
}
|
|
|
|
/**
|
|
* Worker shutdown. In-flight per-channel operations (a command, a swap retry, a relaunch) finish
|
|
* first, so none of them can start a job after the channels were stopped.
|
|
*/
|
|
async stop(): Promise<void> {
|
|
this.stopped = true;
|
|
if (this.heartbeatTimer) clearInterval(this.heartbeatTimer);
|
|
this.heartbeatTimer = null;
|
|
for (const runtime of this.channels.values()) runtime.retryTimer?.cancel();
|
|
await Promise.allSettled([...this.operations.values()]);
|
|
for (const runtime of this.channels.values()) {
|
|
runtime.retryTimer?.cancel();
|
|
runtime.retryTimer = null;
|
|
runtime.desired = { ...runtime.desired, desiredStatus: "STOPPED" };
|
|
await runtime.relay?.stop();
|
|
runtime.realizedVersion = null;
|
|
await this.transition(runtime, "STOPPED", "worker shutdown");
|
|
}
|
|
}
|
|
}
|