diff --git a/docs/stream-compression.md b/docs/stream-compression.md index 60d89f4..5c355a6 100644 --- a/docs/stream-compression.md +++ b/docs/stream-compression.md @@ -30,9 +30,21 @@ Every envelope names where its bytes were captured. `/api/channels/` receives, and it is what every relay today carries. - `source`: the bytes as they arrived, before ffmpeg. A library file's representation (`/api/media//relay`) is at this boundary. A live channel - cannot be relayed at it yet: ffmpeg reads the source itself and the original - bytes never pass through the server. Asking for it answers - `SOURCE_BOUNDARY_UNAVAILABLE`, not a remux labelled as the original. + can also be relayed at it, but only when the source is one nixamp can read + itself and pipe to ffmpeg: a transport stream from a plain http(s) URL or a + local file, joined from its start, with no per-request headers and no + separate audio file. For such a channel nixamp reads the source, hands + ffmpeg the bytes down a pipe, and taps that pipe, so a relay carries the + exact original bytes, padding and all. For anything else, ffmpeg owns the + source and the original bytes never pass through the server, so asking for + the source boundary answers `SOURCE_BOUNDARY_UNAVAILABLE` with the reason, + never a remux labelled as the original. + + A source-boundary relay is a live join: a receiver gets the stream from the + moment it connects, not from the beginning, and its demuxer re-syncs on the + next program table and keyframe. It carries no preface. A channel brought in + from another nixamp's source-boundary relay is itself read through this + server, so it plays here and can be relayed on again at either boundary. ## Wire layout @@ -213,7 +225,12 @@ POST /api/channels//relay (control key) starts a receiver on this server that dials the address, decodes the envelope and feeds a channel here named ``, which listeners hear at -`/api/channels/` exactly as if it were decoded here. It dials again +`/api/channels/` exactly as if it were decoded here. It probes the +relay first: a channel-boundary relay's decoded bytes are the channel's +output directly, while a source-boundary relay's decoded bytes are the +original transport stream, so they are handed to a channel's own ffmpeg +(read through this server) and the channel can be relayed on again. It +dials again after a clean end (the upstream started over: listeners here are ended and rejoin, as they would for a redial) and after a broken one (reported in `status.incoming.error`), and gives up after five dials without a byte. @@ -384,8 +401,9 @@ demand. ## What is not here yet -- A live source relayed at the `source` boundary: needs a tee in front of - ffmpeg for direct HTTP(S) transport-stream sources. +- A source-boundary tee for sources ffmpeg alone can reach: sources behind + per-request headers (a yt-dlp-resolved link), a separate audio track, or a + container other than transport stream stay channel-boundary only. - The PWA and desktop controls, and MCP tools: the API is the contract they will call. - A lower-bitrate quality profile: a separate setting, explicitly labelled, diff --git a/src/channels.ts b/src/channels.ts index 9bd6991..3b4f6b7 100644 --- a/src/channels.ts +++ b/src/channels.ts @@ -80,6 +80,13 @@ export interface ChannelInfo { * a member may have on at once. */ startedBy?: string; + /** + * Set when nixamp reads the source itself and hands ffmpeg the bytes down + * a pipe, so the original bytes pass through this process and can be + * relayed exactly as they came. Only a transport stream from a file or a + * plain URL is read this way, and only when a policy asks for it. + */ + teed?: boolean; } /** Where a pulled source is picked up from, and whether it can be at all. */ @@ -90,6 +97,24 @@ export interface PullResume { position: number; } +/** + * A source read by us rather than by ffmpeg: what to tell ffmpeg it is, + * and how to open it. Opened once per dial; the signal is pulled when that + * dial is over. + */ +export interface PullThrough { + /** ffmpeg's name for the container, e.g. `mpegts`. */ + format: string; + open(signal: AbortSignal): Promise>; +} + +/** + * Asked at every dial whether this source should be read through us. Null + * means ffmpeg dials it as it always has. `from` is where a film is being + * picked up from; a source read through us cannot be joined mid-way. + */ +export type ThroughProvider = (info: ChannelInfo, from: number, input: string[], audio: string) => PullThrough | null; + /** * How far back of the saved place a film is picked up from, in seconds. The * place is written down every so often and a restart lands between two @@ -203,6 +228,8 @@ export interface ChannelOptions { maxListenerQueueBytes?: number; /** How long a rate is measured over before the backlog is sized off it. Tests shorten it. */ rateWindowMs?: number; + /** Whether a pulled source is read here and piped to ffmpeg. See `Channels.setThrough`. */ + through?: ThroughProvider; } /** @@ -249,6 +276,10 @@ export class Channel { */ ephemeral = false; private idle: ReturnType | null = null; + /** Whoever wants the source's own bytes, when the source is read through us. */ + private readonly sourceTaps = new Set(); + /** Pulls the plug on the current read-through, when there is one. */ + private throughAbort: AbortController | null = null; constructor( readonly info: ChannelInfo, @@ -337,6 +368,15 @@ export class Channel { // and a film that has barely started is started. const from = resume.live ? 0 : Math.max(0, Math.floor((this.info.position ?? 0) - REWIND)); const seek = from > 0 ? ["-ss", String(from)] : []; + // Read the source here rather than in ffmpeg, when a policy wants the + // original bytes and the source is the kind that can be. ffmpeg then + // reads a pipe, and every byte that goes down it is also handed to + // whoever has tapped the source. Asked again at every dial, so a + // policy set after the channel started applies at its next restart. + this.throughAbort?.abort(); + this.throughAbort = null; + const through = this.options.through?.(this.info, from, input, audio) ?? null; + this.info.teed = through !== null; const child = spawn( command, [ @@ -348,36 +388,54 @@ export class Channel { // to. Not stderr, which is for what went wrong. "-progress", "pipe:3", "-stats_period", "1", - // A dropped source is normal over hours, and a channel that dies - // the first time a CDN hiccups is not a channel anybody can rely - // on. ffmpeg redials on its own before we have to. - // A connection that stops answering is an error after this long, - // and an error is a thing the reconnect knows what to do with. - // Without it a silent socket is waited on for ever. In - // microseconds, as ffmpeg wants it. - ...remoteArgs, - // Real time, always. A file read as fast as the disk allows is an - // hour of film in ninety seconds and a room that cannot be in it - // together; a live source is already paced and loses nothing. - ...(paced ? ["-re"] : []), - // What the source's site expects on the request: a user agent, a - // referer, a cookie. A link resolved by yt-dlp comes with these, - // and a CDN that got them from yt-dlp and not from us answers 403. - ...input, - ...seek, - "-i", source, - // The sound, when the site keeps it apart from the picture: a - // second input, dialled the same way, that the encode maps in. - ...(audio ? [...remoteArgs, ...(paced ? ["-re"] : []), ...input, ...seek, "-i", audio] : []), + ...(through + ? [ + // Stated, as for a publisher: ffmpeg mis-probes a pipe. Paced + // the same way, since a pipe is read as fast as it is written. + // The source's own input tuning still applies -- a transport + // stream read from a pipe needs the same probe depth and + // generated timestamps it would off a socket -- but request + // headers, which only a source read through us leaves out, + // are no use to a pipe and are not here (a source that needs + // them is not read through us in the first place). + ...input, + "-f", through.format, + ...(paced ? ["-re"] : []), + "-i", "pipe:0", + ] + : [ + // A dropped source is normal over hours, and a channel that dies + // the first time a CDN hiccups is not a channel anybody can rely + // on. ffmpeg redials on its own before we have to. + // A connection that stops answering is an error after this long, + // and an error is a thing the reconnect knows what to do with. + // Without it a silent socket is waited on for ever. In + // microseconds, as ffmpeg wants it. + ...remoteArgs, + // Real time, always. A file read as fast as the disk allows is an + // hour of film in ninety seconds and a room that cannot be in it + // together; a live source is already paced and loses nothing. + ...(paced ? ["-re"] : []), + // What the source's site expects on the request: a user agent, a + // referer, a cookie. A link resolved by yt-dlp comes with these, + // and a CDN that got them from yt-dlp and not from us answers 403. + ...input, + ...seek, + "-i", source, + // The sound, when the site keeps it apart from the picture: a + // second input, dialled the same way, that the encode maps in. + ...(audio ? [...remoteArgs, ...(paced ? ["-re"] : []), ...input, ...seek, "-i", audio] : []), + ]), ...encode, "pipe:1", ], - { stdio: ["ignore", "pipe", "pipe", "pipe"] }, + { stdio: [through ? "pipe" : "ignore", "pipe", "pipe", "pipe"] }, ); let sent = false; this.child = child; this.rearm(child); + if (through) void this.feedThrough(child, through); // ffmpeg's progress: key=value lines, out_time_us being how much it // has written, from where it was told to start. Read whole lines, // since a chunk can end mid-number. Drained whatever it says, for @@ -463,6 +521,22 @@ export class Channel { this.rateBytes = 0; this.rate = 0; this.hangUp(); + // The source's own bytes start over too: a new dial is a new stream + // from its beginning, and whoever was tapping it must not be handed the + // new beginning after the old middle. + this.endTaps(); + } + + /** Everybody tapping the source is told it ended. */ + private endTaps(): void { + for (const tap of this.sourceTaps) { + try { + tap.end(); + } catch { + // Gone already. + } + } + this.sourceTaps.clear(); } /** Expect output within STALL, or treat the source as gone and dial again. */ @@ -671,6 +745,69 @@ export class Channel { this.startOver(); } + /** + * Hear the source's own bytes, as they go down the pipe to ffmpeg. Only + * a channel read through us has any; for the rest this attaches nothing + * and returns null. Ended, like a listener, when the source starts over. + */ + tapSource(tap: Listener): (() => void) | null { + if (!this.info.teed || this.closing) return null; + this.sourceTaps.add(tap); + return () => { + this.sourceTaps.delete(tap); + }; + } + + private tap(chunk: Buffer): void { + for (const tap of this.sourceTaps) { + try { + tap.write(chunk); + } catch { + this.sourceTaps.delete(tap); + } + } + } + + /** + * Read the source and write it to this ffmpeg's stdin, at the rate ffmpeg + * takes it. Every chunk is handed to the taps first, so a tap sees exactly + * the bytes ffmpeg does. When the source ends, stdin is ended, ffmpeg + * finishes, and its close handler dials again -- the same path a source + * that ffmpeg read itself takes when it drops. + */ + private async feedThrough(child: ChildProcess, through: PullThrough): Promise { + const controller = new AbortController(); + this.throughAbort = controller; + const stdin = child.stdin; + if (!stdin) return; + stdin.on("error", () => undefined); + try { + const body = await through.open(controller.signal); + for await (const raw of body) { + if (this.child !== child || this.closing || controller.signal.aborted) break; + const chunk = Buffer.isBuffer(raw) ? raw : Buffer.from(raw); + this.tap(chunk); + if (!stdin.write(chunk)) { + // Wait for ffmpeg to take it, or for the pipe to go: a pipe that + // closed never drains, and waiting on it would hold the read open. + await new Promise((done) => { + stdin.once("drain", done); + stdin.once("close", done); + }); + } + } + } catch (error) { + if (this.child === child && !controller.signal.aborted) this.info.error = (error as Error).message; + } finally { + try { + stdin.end(); + } catch { + // Already gone. + } + if (this.throughAbort === controller) this.throughAbort = null; + } + } + listen(listener: Listener): () => void { // What the stream is, before any of what it is currently saying. Without // this a listener who arrives after the first second gets fragments that @@ -735,6 +872,9 @@ export class Channel { if (said && !this.info.error) this.info.error = said; const child = this.child; this.child = null; + this.throughAbort?.abort(); + this.throughAbort = null; + this.endTaps(); try { child?.stdin?.end(); } catch { @@ -838,6 +978,7 @@ export class Channels { input: string[] = [], audio = "", resume: PullResume = { live: true, position: 0 }, + codecs?: ChannelInfo["codecs"], ): Channel | null { if (this.open.has(id)) return null; const channel = new Channel( @@ -851,6 +992,9 @@ export class Channels { listeners: 0, kind, source, + // Known before the first dial, so whether to read the source here + // can be decided from what it holds. + ...(codecs ? { codecs } : {}), }, this.options, (gone) => this.open.delete(gone), @@ -965,6 +1109,20 @@ export class Channels { return this.open.get(id)?.opening() ?? []; } + /** Hear a channel's source bytes, when it is read through us. Null otherwise. */ + tapSource(id: string, tap: Listener): (() => void) | null { + return this.open.get(id)?.tapSource(tap) ?? null; + } + + /** + * Who decides whether a pulled source is read here and piped to ffmpeg. + * Set once by whoever owns the policies; asked at every dial. + */ + setThrough(provider: ThroughProvider | null): void { + if (provider) this.options.through = provider; + else delete this.options.through; + } + /** The kind of a channel, for a relay to say what it is carrying. */ kindOf(id: string): "audio" | "video" | undefined { return this.open.get(id)?.info.kind; diff --git a/src/compression/receiver.ts b/src/compression/receiver.ts index 3d85aeb..e1d0dc0 100644 --- a/src/compression/receiver.ts +++ b/src/compression/receiver.ts @@ -10,13 +10,78 @@ * which is surfaced as an error naming the reason, never as a stream of * something else. */ -import { MEDIA_TYPE, type Mode, RelayError, type StreamHeader } from "./envelope.ts"; +import { type Boundary, MEDIA_TYPE, type Mode, RelayError, type StreamHeader } from "./envelope.ts"; import { RelayDecoder } from "./relay.ts"; export const CODECS_HEADER = "x-nixamp-stream-codecs"; export const KIND_HEADER = "x-nixamp-kind"; +export const BOUNDARY_HEADER = "x-nixamp-boundary"; +/** What the media inside is, as JSON: video, audio, container, duration. */ +export const MEDIA_HEADER = "x-nixamp-codecs"; export const KEY_HEADER = "x-nixamp-key"; +export interface Media { + video: string; + audio: string; + container: string; + duration?: number; +} + +/** What a server said it would send, read off its response headers. */ +export interface Accepted { + codecs: string; + kind: "audio" | "video" | ""; + boundary: Boundary | ""; + media: Media | null; +} + +function acceptedFrom(headers: Headers): Accepted { + const kindSaid = headers.get(KIND_HEADER); + const boundarySaid = headers.get(BOUNDARY_HEADER); + let media: Media | null = null; + try { + const raw = JSON.parse(headers.get(MEDIA_HEADER) ?? "null") as Partial | null; + if (raw && typeof raw.video === "string" && typeof raw.audio === "string" && typeof raw.container === "string") { + media = { video: raw.video, audio: raw.audio, container: raw.container }; + if (typeof raw.duration === "number") media.duration = raw.duration; + } + } catch { + // Not JSON: no media description, which the receiver copes with. + } + return { + codecs: headers.get(CODECS_HEADER) ?? "", + kind: kindSaid === "audio" || kindSaid === "video" ? kindSaid : "", + boundary: boundarySaid === "source" || boundarySaid === "channel" ? boundarySaid : "", + media, + }; +} + +/** + * Ask a relay what it would send, without taking it: the negotiation + * headers come back on the response and the body is cancelled at once. + * Costs one short connection; a receiver needs to know the boundary and + * the media before it can decide how to carry the stream. + */ +export async function probeRelay(url: string, key: string | null, fetchImpl?: typeof fetch): Promise { + const headers: Record = { accept: MEDIA_TYPE, [CODECS_HEADER]: "stored,zstd,ts-zstd" }; + if (key) headers[KEY_HEADER] = key; + const response = await (fetchImpl ?? fetch)(url, { headers }); + const type = response.headers.get("content-type") ?? ""; + if (response.status !== 200 || !type.startsWith(MEDIA_TYPE)) { + let reason = `${response.status}`; + try { + const body = (await response.json()) as { error?: string }; + if (typeof body.error === "string") reason = body.error; + } catch { + // Not JSON; the status is the message. + } + throw new RelayRefused(response.status, reason); + } + const accepted = acceptedFrom(response.headers); + await response.body?.cancel().catch(() => undefined); + return accepted; +} + export interface ReceiveOptions { url: string; key: string | null; @@ -24,7 +89,7 @@ export interface ReceiveOptions { modes?: Mode[]; maxFrameBytes?: number; /** The server said yes: what it will compress with, and what the channel carries. Before any byte. */ - onStart?: (accepted: { codecs: string; kind: "audio" | "video" | "" }) => void; + onStart?: (accepted: Accepted) => void; /** The stream header arrived: the generation this is. */ onHeader?: (header: StreamHeader) => void; onBytes: (bytes: Buffer) => void | Promise; @@ -74,11 +139,7 @@ export async function receiveRelay(options: ReceiveOptions): Promise answer.session.leave(); @@ -160,7 +165,7 @@ export async function handleChannelCompression( } const key = typeof body?.["key"] === "string" ? body["key"] : null; const name = typeof body?.["name"] === "string" ? body["name"] : id; - const started = service.pull(id, from, key, name); + const started = await service.pull(id, from, key, name); if (!started.ok) { json(response, started.status, { error: started.error }); return true; diff --git a/src/compression/service.ts b/src/compression/service.ts index 809ffc8..2003b7c 100644 --- a/src/compression/service.ts +++ b/src/compression/service.ts @@ -8,16 +8,17 @@ * translates flags into the routes. There is no second copy of the rules. */ import { statSync } from "node:fs"; -import { codecsOf } from "../audio.ts"; -import { type Channel, type Channels, GIVE_UP, REDIAL } from "../channels.ts"; +import { codecsOf, videoArgs } from "../audio.ts"; +import { type Channel, type ChannelInfo, type Channels, GIVE_UP, type PullThrough, REDIAL } from "../channels.ts"; import { type Analysis, analyzeSample, SAMPLE_MAX_BYTES, SAMPLE_MAX_SECONDS } from "./analyze.ts"; import { Pool } from "./codec.ts"; -import { type Mode, RelayError } from "./envelope.ts"; +import { type Boundary, type Mode, RelayError } from "./envelope.ts"; import { AnalysisJobs, type Job } from "./jobs.ts"; import { ChannelMetrics, type ChannelMetricsSnapshot } from "./metrics.ts"; import { type ChannelPolicy, type LosslessPolicy, variantOf } from "./policy.ts"; -import { receiveRelay, RelayRefused } from "./receiver.ts"; +import { type Accepted, probeRelay, receiveRelay, RelayRefused } from "./receiver.ts"; import { RelayEncoder, type RelayListener, type RelaySession } from "./relay.ts"; +import { relayThrough, sourceThrough, unreadable } from "./source.ts"; import { type GlobalSettings, PolicyStore } from "./store.ts"; import { type Prepared, StaticCache } from "./static.ts"; @@ -101,7 +102,16 @@ export interface ChannelStatus { } export type RelayAnswer = - | { ok: true; session: RelaySession; codecs: Mode[]; generation: number; kind: "audio" | "video" | "" } + | { + ok: true; + session: RelaySession; + codecs: Mode[]; + generation: number; + kind: "audio" | "video" | ""; + /** Which bytes the stream carries, and what is inside them. */ + boundary: Boundary; + media: ChannelInfo["codecs"] | null; + } | { ok: false; status: 404 | 406 | 409 | 503; code: string; error: string }; interface Running { @@ -110,7 +120,17 @@ interface Running { variant: string; } -/** An incoming relay: this server as the receiver, dialling again when it drops. */ +/** + * An incoming relay: this server as the receiver. + * + * Two shapes, decided by what the sender says it carries. A channel-boundary + * relay is the sender's finished output, so it is fed straight into a + * channel here and dialled again by this class when it drops. A + * source-boundary relay is the original transport stream, so it is handed + * to a channel's own ffmpeg as a source read through us -- which plays it + * exactly as the sender would have, lets it be relayed on again, and puts + * the redialling in the channel where it already lives. + */ class Incoming { generation = 0; reconnects = 0; @@ -223,8 +243,24 @@ export class CompressionService { this.statics = options.cacheDir ? new StaticCache(options.cacheDir, { pool: this.pool, ...(options.maxCacheBytes !== undefined ? { maxBytes: options.maxCacheBytes } : {}) }) : null; + // Whether a pulled source is read here rather than by ffmpeg, decided at + // every dial: only when the channel's policy asks for the source boundary + // and the source is one we can read exactly (a plain transport stream). + // A channel we brought in from another nixamp's source-boundary relay is + // itself read through us, so it can be relayed on. + this.channels.setThrough((info, from, input, audio) => { + const relayed = this.relaySource.get(info.id); + if (relayed) return relayThrough(relayed.from, relayed.key); + const policy = this.store.get(info.id).losslessCompression; + if (policy.mode === "off" || policy.boundary !== "source") return null; + if (unreadable(info, from, input, audio) !== null) return null; + return sourceThrough(info.source ?? ""); + }); } + /** Incoming source-boundary relays, so their channel is read through us. */ + private readonly relaySource = new Map(); + private get channels(): Channels { return this.options.channels; } @@ -239,8 +275,15 @@ export class CompressionService { lossless.mode = "off"; reason = "compression is off for the whole server"; } else if (lossless.mode !== "off" && lossless.boundary === "source") { - lossless.mode = "off"; - reason = "original source bytes are unavailable: this server's sources are read by ffmpeg, and only the channel boundary can be relayed"; + // The source boundary is available only for a source we can read here. + // A live channel that is not read through us cannot offer it; a channel + // that is not on yet is given the benefit of the doubt until it is. + const info = this.channels.info(id); + const why = info ? unreadable(info, 0, [], "") : null; + if (why !== null) { + lossless.mode = "off"; + reason = `original source bytes are unavailable: ${why}`; + } } return { losslessCompression: lossless, @@ -333,7 +376,9 @@ export class CompressionService { } // A receiver that cannot undo the transform gets a variant without it. const variantPolicy: LosslessPolicy = { ...policy, tsAware: policy.tsAware && codecs.includes("ts-zstd") }; - const variant = variantOf(variantPolicy); + // The boundary is part of the variant: a source relay and a channel + // relay are two different streams, and two different compressors. + const variant = `${policy.boundary}:${variantOf(variantPolicy)}`; let running = this.running.get(id); if (running && running.variant !== variant) { // One compressor per channel. A second variant would be a second @@ -341,17 +386,32 @@ export class CompressionService { return { ok: false, status: 409, code: "VARIANT_IN_USE", error: "this channel is already being relayed under a different codec set; try again when that relay ends" }; } if (!running) { - const started = this.startEncoder(id, variantPolicy, variant); - if (!started) return { ok: false, status: 503, code: "CHANNEL_GONE", error: "the channel ended before the relay could start" }; + const started = this.startEncoder(id, variantPolicy, variant, policy.boundary); + if (!started) { + return started === null + ? { ok: false, status: 503, code: "CHANNEL_GONE", error: "the channel ended before the relay could start" } + : { ok: false, status: 409, code: "SOURCE_BOUNDARY_UNAVAILABLE", error: "the source is not being read through this server, so its original bytes are not available to relay" }; + } running = started; } - // The opening bytes and the join point, in the same tick. - const preface = this.channels.opening(id); + // For the channel boundary, the opening bytes are the preface. The source + // boundary has none: a joining demuxer re-syncs on the next PAT and + // keyframe, exactly as it does when a live source is joined. + const preface = policy.boundary === "source" ? [] : this.channels.opening(id); const session = running.encoder.join(listener, preface); - return { ok: true, session, codecs, generation: running.encoder.generation, kind: this.channels.kindOf(id) ?? "" }; + return { + ok: true, + session, + codecs, + generation: running.encoder.generation, + kind: this.channels.kindOf(id) ?? "", + boundary: policy.boundary, + media: this.channels.info(id)?.codecs ?? null, + }; } - private startEncoder(id: string, policy: LosslessPolicy, variant: string): Running | null { + /** Null: the channel ended. False: the source boundary was asked for but is not being read through us. */ + private startEncoder(id: string, policy: LosslessPolicy, variant: string, boundary: Boundary): Running | null | false { this.generation = (this.generation + 1) % 0xffff_ffff; const metrics = this.metrics.get(id) ?? new ChannelMetrics(); this.metrics.set(id, metrics); @@ -361,7 +421,7 @@ export class CompressionService { policy, pool: this.pool, metrics, - boundary: "channel", + boundary, onAbort: () => { if (record && this.running.get(id) === record) { this.running.delete(id); @@ -378,22 +438,58 @@ export class CompressionService { } }, }); - const detach = this.channels.listen(id, { - write: (chunk) => encoder.write(chunk), - end: () => encoder.end(), - pending: () => metrics.queueBytes, - }); - if (detach === null) return null; + const face = { + write: (chunk: Buffer): boolean => encoder.write(chunk), + end: (): void => encoder.end(), + pending: (): number => metrics.queueBytes, + }; + // The source boundary taps the source's own bytes; the channel boundary + // is one more ordinary listener on the channel's output. + const detach = boundary === "source" ? this.channels.tapSource(id, face) : this.channels.listen(id, face); + if (detach === null) return boundary === "source" ? false : null; encoder.attach(); record = { encoder, detach, variant }; this.running.set(id, record); return record; } - /** Start listening to another nixamp's channel as one of ours. */ - pull(id: string, from: string, key: string | null, name: string): { ok: true } | { ok: false; status: 409 | 400; error: string } { + /** + * Start listening to another nixamp's channel as one of ours. Probes the + * relay first to learn its boundary: a channel-boundary relay is decoded + * and its bytes are the channel's output directly; a source-boundary relay + * is the original transport stream, so it is read through ffmpeg here + * exactly as a pulled source would be, and can be relayed on again. + */ + async pull(id: string, from: string, key: string | null, name: string): Promise<{ ok: true } | { ok: false; status: 409 | 400 | 502; error: string }> { if (!/^https?:\/\//i.test(from)) return { ok: false, status: 400, error: "the relay address must be http or https" }; if (this.channels.has(id) || this.incoming.has(id)) return { ok: false, status: 409, error: `channel "${id}" is already on` }; + let accepted: Accepted; + try { + accepted = await probeRelay(from, key); + } catch (error) { + const message = error instanceof RelayRefused ? `the relay refused: ${error.message}` : (error as Error).message; + return { ok: false, status: 502, error: message }; + } + if (accepted.boundary === "source") { + // The upstream sends the original transport stream. Read it through + // ffmpeg here: the through-provider sees this registration and hands + // ffmpeg the decoded bytes. It can then be relayed on at either + // boundary, since its source now passes through this process. + const media = accepted.media; + const kind: "audio" | "video" = accepted.kind === "audio" ? "audio" : "video"; + const encode = kind === "video" + ? videoArgs(media ?? { video: "h264", audio: "aac", container: "mpegts" }) + : ["-vn", "-c:a", "libmp3lame", "-b:a", "192k", "-f", "mp3"]; + this.relaySource.set(id, { from, key }); + // A placeholder source string: the through-provider supplies the bytes, + // so this is only what a listener would never see and a restart reads. + const channel = this.channels.pull(id, name, from, encode, kind, true, undefined, [], "", { live: true, position: 0 }, media ?? undefined); + if (!channel) { + this.relaySource.delete(id); + return { ok: false, status: 409, error: `channel "${id}" is already on` }; + } + return { ok: true }; + } const inbound = new Incoming(id, from, key, name, this.channels, this.options.onEvent ?? (() => undefined), (gone) => this.incoming.delete(gone)); this.incoming.set(id, inbound); inbound.start(); @@ -402,6 +498,7 @@ export class CompressionService { /** Stop an incoming relay, and the channel it feeds. */ stopPull(id: string): boolean { + if (this.relaySource.delete(id)) return this.channels.stop(id); const inbound = this.incoming.get(id); if (!inbound) return false; inbound.stop(); diff --git a/src/compression/source.ts b/src/compression/source.ts new file mode 100644 index 0000000..3587e4a --- /dev/null +++ b/src/compression/source.ts @@ -0,0 +1,130 @@ +/** + * Reading a source ourselves, so its bytes exist here to be relayed. + * + * ffmpeg normally dials a channel's source and the original bytes never + * pass through this process. For a source-boundary relay they have to. + * This is the narrow set of sources that can be read here and piped to + * ffmpeg without changing how they play: a transport stream from a plain + * http(s) URL or a file, from its beginning, with no special request + * headers and no separate sound file. Anything else is left to ffmpeg, + * and a relay that asks for its source boundary is told why not. + * + * The second half is the receiving end of a source-boundary relay: the + * decoded original bytes, made into a source ffmpeg can be handed the + * same way, so a relayed transport stream plays here exactly as the + * original would have, and can be relayed on again. + */ +import { createReadStream, statSync } from "node:fs"; +import type { ChannelInfo, PullThrough } from "../channels.ts"; +import { receiveRelay } from "./receiver.ts"; + +/** + * ffmpeg input flags that carry per-request authentication a plain fetch + * would not send: a user agent, a referer, cookies, arbitrary headers. A + * source that needs any of these is ffmpeg's to dial. Transport demux + * tuning (`-probesize`, `-analyzeduration`, `-fflags`) is not on this list: + * it applies to a piped read too and is no reason to refuse the tee. + */ +const HEADER_FLAGS = new Set(["-headers", "-user_agent", "-user-agent", "-referer", "-cookies", "-icy", "-http_proxy", "-http_persistent"]); + +/** The reason a source cannot be read here, or null when it can. */ +export function unreadable(info: ChannelInfo, from: number, input: string[], audio: string): string | null { + const source = info.source ?? ""; + if (info.via !== "pull" || source === "") return "only a source this server pulls can be read here"; + const container = info.codecs?.container ?? ""; + if (!container.split(",").includes("mpegts")) return `the source is ${container || "of unknown container"}, not a transport stream`; + if (input.some((arg) => HEADER_FLAGS.has(arg))) return "the source needs request headers ffmpeg sends and nixamp does not"; + if (audio !== "") return "the source keeps its sound in a second file"; + if (from > 0) return "a film picked up mid-way cannot be read from its start"; + if (/^https?:\/\//i.test(source)) return null; + if (/^[a-z][a-z0-9+.-]*:\/\//i.test(source)) return "only http(s) URLs and files can be read here"; + try { + if (!statSync(source).isFile()) return "the source is not a file"; + } catch { + return "the source file cannot be read"; + } + return null; +} + +/** Open a plain URL or a file as a stream of bytes, pulled shut by the signal. */ +export async function openSource(source: string, signal: AbortSignal): Promise> { + if (/^https?:\/\//i.test(source)) { + const response = await fetch(source, { signal, redirect: "follow", headers: { "user-agent": "nixamp" } }); + if (!response.ok || !response.body) throw new Error(`the source answered ${response.status}`); + return response.body as unknown as AsyncIterable; + } + const stream = createReadStream(source, { highWaterMark: 256 * 1024 }); + signal.addEventListener("abort", () => stream.destroy(), { once: true }); + return stream; +} + +/** A transport-stream source read by us, for `Channels.setThrough`. */ +export function sourceThrough(source: string): PullThrough { + return { format: "mpegts", open: (signal) => openSource(source, signal) }; +} + +/** How many decoded bytes may wait for ffmpeg before the relay is asked to pause. */ +const RELAY_QUEUE = 8 * 1024 * 1024; + +/** + * A relay's decoded bytes as an async iterable, at the pace ffmpeg takes + * them. The decoder is asked to wait when too much is queued, which holds + * the socket read, which is backpressure all the way to the sender's + * per-listener queue. A clean end ends the iterable; a broken stream + * throws, and the channel's own redial dials again. + */ +export async function* relayBytes(url: string, key: string | null, signal: AbortSignal): AsyncGenerator { + const queue: Buffer[] = []; + let queued = 0; + let wake: (() => void) | null = null; + let room: (() => void) | null = null; + let done = false; + let failure: Error | null = null; + void receiveRelay({ + url, + key, + signal, + onBytes: async (bytes) => { + queue.push(bytes); + queued += bytes.length; + wake?.(); + wake = null; + if (queued > RELAY_QUEUE) await new Promise((resume) => { room = resume; }); + }, + }).then( + () => { + done = true; + wake?.(); + }, + (error: Error) => { + failure = error; + done = true; + wake?.(); + }, + ); + for (;;) { + const next = queue.shift(); + if (next) { + queued -= next.length; + // Captured into a local: `room` is only ever assigned inside a callback, + // which the compiler cannot see, so it would narrow the field to null. + const resume = room as (() => void) | null; + if (resume && queued <= RELAY_QUEUE / 2) { + resume(); + room = null; + } + yield next; + continue; + } + if (done) { + if (failure) throw failure; + return; + } + await new Promise((resume) => { wake = resume; }); + } +} + +/** A source-boundary relay from another nixamp, as a source ffmpeg reads through us. */ +export function relayThrough(url: string, key: string | null): PullThrough { + return { format: "mpegts", open: async (signal) => relayBytes(url, key, signal) }; +} diff --git a/src/server.ts b/src/server.ts index 0da41df..df7341c 100644 --- a/src/server.ts +++ b/src/server.ts @@ -1136,6 +1136,9 @@ export async function pullChannel( const channel = channels.pull( id, name, source, encode, kind, true, undefined, opening, kind === "video" ? audio : "", { live, position: known.position ?? 0 }, + // Known before the first dial: whether the source is a transport stream + // decides whether it can be read here for a source-boundary relay. + assumed ? undefined : codecs, ); if (channel && !assumed) channel.info.codecs = codecs; // What comes out, as opposed to what went in. An H.265 source copied diff --git a/test/compression-source.test.ts b/test/compression-source.test.ts new file mode 100644 index 0000000..ae11187 --- /dev/null +++ b/test/compression-source.test.ts @@ -0,0 +1,230 @@ +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { spawnSync } from "node:child_process"; +import { fileURLToPath } from "node:url"; +import { randomBytes } from "node:crypto"; +import type { AddressInfo } from "node:net"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { unreadable } from "../src/compression/source.ts"; +import { Channels, type ChannelInfo, type Listener, type PullThrough } from "../src/channels.ts"; +import { CompressionService } from "../src/compression/service.ts"; +import { createServer, EmptyEngine, pullChannel } from "../src/server.ts"; +import { receiveRelay } from "../src/compression/receiver.ts"; +import { detectTools } from "../src/audio.ts"; + +const STUB = [process.execPath, fileURLToPath(new URL("./fixtures/stub-ffmpeg.mjs", import.meta.url))]; + +const FFMPEG = detectTools().ffmpeg; +const FFPROBE = detectTools().ffprobe; +const ffmpegHere = ((): boolean => { + const [cmd, ...rest] = FFMPEG; + if (!cmd) return false; + const r = spawnSync(cmd, [...rest, "-version"], { encoding: "utf8", timeout: 10_000 }); + return !r.error && r.status === 0; +})(); + +function info(over: Partial = {}): ChannelInfo { + return { id: "x", name: "x", format: "mp4", via: "pull", startedAt: 0, bytes: 0, listeners: 0, source: "https://h/s.ts", codecs: { video: "h264", audio: "aac", container: "mpegts" }, ...over }; +} + +test("unreadable names why a source cannot be read here, and passes a plain transport stream", () => { + assert.equal(unreadable(info(), 0, [], ""), null, "a plain http transport stream is readable"); + assert.equal(unreadable(info({ source: "/tmp/does-not-exist.ts" }), 0, [], ""), "the source file cannot be read"); + assert.match(unreadable(info({ codecs: { video: "h264", audio: "aac", container: "mov,mp4,m4a" } }), 0, [], "") ?? "", /not a transport stream/); + assert.match(unreadable(info(), 0, ["-headers", "x"], "") ?? "", /request headers/); + assert.match(unreadable(info(), 0, [], "https://h/audio.ts") ?? "", /second file/); + assert.match(unreadable(info(), 30, [], "") ?? "", /picked up mid-way/); + assert.match(unreadable(info({ source: "rtmp://h/s" }), 0, [], "") ?? "", /http\(s\) URLs and files/); + assert.match(unreadable(info({ via: "http", source: "" }), 0, [], "") ?? "", /source this server pulls/); +}); + +const CONTROL = "control-key"; +const LISTEN = "listen-key"; + +/** A source I feed by hand, so a tap can attach before the first byte. */ +function drip(): { through: PullThrough; push: (b: Buffer) => void; end: () => void } { + const queue: Buffer[] = []; + let wake: (() => void) | null = null; + let done = false; + return { + through: { + format: "mpegts", + open: async () => + (async function* () { + for (;;) { + const next = queue.shift(); + if (next) { yield next; continue; } + if (done) return; + await new Promise((r) => { wake = r; }); + } + })(), + }, + push: (b) => { queue.push(b); wake?.(); wake = null; }, + end: () => { done = true; wake?.(); wake = null; }, + }; +} + +const settle = (): Promise => new Promise((r) => setTimeout(r, 30)); + +test("a tapped source is byte-exact, sees exactly ffmpeg's input, and ends when the source starts over", async () => { + const source = drip(); + const channels = new Channels({ ffmpeg: STUB, through: () => source.through }); + const channel = channels.pull("clip", "A clip", "http://h/s.ts", ["-f", "mpegts", "-i", "pipe:0"], "audio", true, 30_000, [], "", { live: true, position: 0 }, { video: "", audio: "aac", container: "mpegts" }); + assert.ok(channel); + assert.equal(channel.info.teed, true, "the source is read through the channel"); + // Attach the tap before any byte is pushed: it must see all of them. + const tapped: Buffer[] = []; + let tapEnded = false; + const detach = channel.tapSource({ write: (b) => { tapped.push(Buffer.from(b)); return true; }, end: () => { tapEnded = true; } }); + assert.ok(detach, "a teed channel offers its source"); + const original = [randomBytes(1000), randomBytes(20_000), Buffer.from("the end ".repeat(100))]; + for (const chunk of original) source.push(chunk); + await settle(); + assert.ok(Buffer.concat(tapped).equals(Buffer.concat(original)), "the tap saw every source byte, in order"); + + // A source-read channel that is not teed offers nothing. + const plain = channels.pull("plain", "x", "http://h/s.ts", ["-f", "mpegts", "-i", "pipe:0"], "audio", true, 30_000, [], "", { live: true, position: 0 }); + // (the provider returns a through for every channel here, so `plain` is teed too; + // a channel with no provider is covered by the service tests.) + plain?.close(); + + // The source starting over ends the tap: a new generation must not follow + // the old middle. + channel.rollover(); + assert.equal(tapEnded, true, "the tap was ended when the source rolled over"); + detach?.(); + channels.stopAll(); +}); + +test("a source-boundary relay serves the original bytes to a receiver over HTTP, packet-aligned", async () => { + // A real transport stream: 188-byte packets, half null padding so it compresses. + const packets: Buffer[] = []; + for (let i = 0; i < 400; i += 1) { + const p = Buffer.alloc(188, i % 2 === 0 ? 0xff : (i & 0xff)); + p[0] = 0x47; + packets.push(p); + } + const original = Buffer.concat(packets); + + // A source server the upstream reads through its own tee. It holds the + // response open until released, so the receiver is attached before a byte + // flows and gets the stream from its very start. + const { createServer: httpServer } = await import("node:http"); + let release: (() => void) | null = null; + const held = new Promise((r) => { release = r; }); + const dataServer = httpServer((_req, res) => { + res.writeHead(200, { "content-type": "video/mp2t" }); + void held.then(async () => { + for (let at = 0; at < original.length; at += 2000) { + res.write(original.subarray(at, at + 2000)); + await new Promise((t) => setTimeout(t, 5)); + } + res.end(); + }); + }); + await new Promise((done) => dataServer.listen(0, "127.0.0.1", done)); + const sourceUrl = `http://127.0.0.1:${(dataServer.address() as AddressInfo).port}/s.ts`; + + const channels = new Channels({ ffmpeg: STUB }); + const service = new CompressionService({ channels, stateDir: null, port: 1 }); + const server = createServer(new EmptyEngine(), { web: null, media: true, version: "test", key: CONTROL, listenKey: LISTEN, channels, compression: service }); + await new Promise((done) => server.listen(0, "127.0.0.1", done)); + const base = `http://127.0.0.1:${(server.address() as AddressInfo).port}`; + try { + service.set("clip", { losslessCompression: { mode: "zstd", boundary: "source", maxBlockBytes: 8192, maxHoldMs: 10 } }); + // The service's own tee provider reads the source URL through this server. + const channel = channels.pull("clip", "A clip", sourceUrl, ["-f", "mpegts", "-i", "pipe:0"], "audio", true, 30_000, [], "", { live: true, position: 0 }, { video: "", audio: "aac", container: "mpegts" }); + assert.ok(channel?.info.teed); + assert.equal(service.effective("clip").losslessCompression.boundary, "source"); + + const pieces: Buffer[] = []; + let boundary = ""; + const controller = new AbortController(); + const relay = receiveRelay({ + url: `${base}/api/channels/clip/relay`, + key: LISTEN, + signal: controller.signal, + onStart: (a) => { boundary = a.boundary; }, + onBytes: (b) => { pieces.push(b); }, + }).catch(() => undefined); + await settle(); + // Receiver is attached: let the source flow, from the top. + release?.(); + await new Promise((t) => setTimeout(t, 700)); + controller.abort(); + await relay; + const got = Buffer.concat(pieces); + assert.equal(boundary, "source", "the stream declared the source boundary"); + assert.ok(got.length > original.length / 2, `got ${got.length} of ${original.length}`); + assert.ok(got.equals(original.subarray(0, got.length)), "the received bytes are the original, byte for byte from the start"); + assert.equal(got[0], 0x47, "starts on a packet"); + const metrics = service.status("clip").metrics!; + assert.ok(metrics.compressedBlocks > 0, "the padded packets compressed at the source boundary"); + } finally { + service.stopAll(); + channels.stopAll(); + await new Promise((done) => server.close(() => done())); + await new Promise((done) => dataServer.close(() => done())); + } +}); + +test("the real ffmpeg tee reads a transport-stream file byte-exact through the server", { skip: !ffmpegHere }, async () => { + const dir = mkdtempSync(join(tmpdir(), "nixamp-tee-")); + const file = join(dir, "clip.ts"); + const make = spawnSync(FFMPEG[0] as string, [ + ...FFMPEG.slice(1), "-hide_banner", "-loglevel", "error", "-y", + "-f", "lavfi", "-i", "testsrc=size=320x240:rate=25", "-t", "2", "-c:v", "libx264", "-preset", "ultrafast", "-g", "25", "-pix_fmt", "yuv420p", "-muxrate", "3000k", "-f", "mpegts", file, + ], { timeout: 60_000 }); + assert.equal(make.status, 0, make.stderr?.toString()); + const original = (await import("node:fs")).readFileSync(file); + + const channels = new Channels({ ffmpeg: FFMPEG }); + const service = new CompressionService({ channels, stateDir: null, port: 1, ffprobe: FFPROBE }); + try { + service.set("clip", { losslessCompression: { mode: "zstd", boundary: "source" } }); + // Tap through the channel directly, driven by the same provider the + // service installs; the tap collects every byte ffmpeg is fed. + const tapped: Buffer[] = []; + let attached: (() => void) | null = null; + // Start the pull; the tee opens on dial. Attach as soon as the channel exists. + const channel = await pullChannel(channels, FFPROBE, "clip", "A clip", file); + assert.ok(channel); + assert.equal(channel.info.teed, true, "a real transport-stream file is read through the server"); + attached = channel.tapSource({ write: (b) => { tapped.push(Buffer.from(b)); return true; }, end: () => undefined }); + // The tap may attach a beat after the first chunks on a fast disk; what it + // does catch must be a byte-exact run of the file, padding included. + await new Promise((t) => setTimeout(t, 1500)); + attached?.(); + const got = Buffer.concat(tapped); + if (got.length > 0) { + const at = original.indexOf(got.subarray(0, Math.min(2000, got.length))); + assert.notEqual(at, -1, "the tapped bytes appear verbatim in the file"); + assert.ok(original.subarray(at, at + got.length).equals(got), "and continue byte-exact: the original, not a re-mux"); + } + } finally { + service.stopAll(); + channels.stopAll(); + rmSync(dir, { recursive: true, force: true }); + } +}); + +test("a channel whose source ffmpeg reads cannot offer the source boundary, and says why", () => { + const channels = new Channels({ ffmpeg: ["ffmpeg"] }); + const service = new CompressionService({ channels, stateDir: null, port: 1 }); + try { + // A published channel (not pulled) has no source to read through us. + channels.attach("pub", "a device", "mp3", "http"); + service.set("pub", { losslessCompression: { mode: "zstd", boundary: "source" } }); + const eff = service.effective("pub"); + assert.equal(eff.losslessCompression.mode, "off"); + assert.match(eff.reason ?? "", /original source bytes are unavailable/); + const answer = service.relay("pub", { write: () => true, end: () => undefined } as Listener, new Set(["zstd"] as const)); + assert.equal(answer.ok, false); + assert.equal(!answer.ok && answer.code, "SOURCE_BOUNDARY_UNAVAILABLE"); + } finally { + service.stopAll(); + channels.stopAll(); + } +}); diff --git a/test/fixtures/stub-ffmpeg.mjs b/test/fixtures/stub-ffmpeg.mjs new file mode 100644 index 0000000..358afa3 --- /dev/null +++ b/test/fixtures/stub-ffmpeg.mjs @@ -0,0 +1,7 @@ +// A stand-in for ffmpeg in tests: copy stdin to stdout, ignore every flag, +// and stay alive until stdin ends. Enough for a channel to have a live +// "decoder" whose output flows to listeners while its input is tapped. +process.stdin.on("data", (chunk) => process.stdout.write(chunk)); +process.stdin.on("end", () => process.exit(0)); +process.stdin.on("error", () => process.exit(0)); +process.stdout.on("error", () => process.exit(0));