From bf87ce682df1f54666be7180f93a5c43fc14e405 Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Sat, 12 Sep 2026 11:45:17 +0000 Subject: [PATCH] Relay a live source at the source boundary, byte-exact The source boundary was documented but unbuilt: a live channel could only be relayed as its post-ffmpeg output, and asking for the source bytes always answered SOURCE_BOUNDARY_UNAVAILABLE. Now, for the sources nixamp can read itself, it can. A Channel may read its own source and pipe it to ffmpeg rather than let ffmpeg dial it. Every chunk that goes down that pipe is first handed to whoever has tapped the source, so a tap sees exactly the bytes ffmpeg does -- the original transport stream, padding and all, not the re-muxed output that drops null packets. The tap is ended when the source starts over, so a new generation never follows the old middle, and the pipe is paced by ffmpeg's own reading through ordinary backpressure. Only the sources that can be read exactly are read this way: 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. Anything else stays ffmpeg's to dial, and a source-boundary relay of it is refused with the reason. The readability gate is one function, tested against each way a source can fail it. The relay encoder now carries its boundary in the envelope and the negotiation headers; a source relay taps the source, a channel relay listens to the output, and the two are different variants so a channel can serve both at once without one compressor doing the other's work. A receiver learns the boundary and the media codecs before the first byte, so a source-boundary receiver can hand the bytes to its own ffmpeg without a probe. Bringing a relay in now probes it first: a source-boundary upstream is read through this server too, so it plays here and can be relayed on again at either boundary. A source-boundary relay is a live join with no preface: the receiver's demuxer re-syncs on the next program table and keyframe, the same way a listener joining a live channel does. Tests: the readability gate; a deterministic byte-exact tap with a controlled source and a stub ffmpeg, including rollover ending the tap; an end-to-end source relay over HTTP that reassembles the original packets byte for byte from the start; and a real-ffmpeg tee reading a transport stream file. Full suite 509 pass; the one remaining failure is the pre-existing @profullstack/player web-test module resolution on this branch, unchanged from main. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01MxNif5tsYq4LczgG7aE8Jp --- docs/stream-compression.md | 30 ++++- src/channels.ts | 202 +++++++++++++++++++++++++--- src/compression/receiver.ts | 75 ++++++++++- src/compression/routes.ts | 9 +- src/compression/service.ts | 145 ++++++++++++++++---- src/compression/source.ts | 130 ++++++++++++++++++ src/server.ts | 3 + test/compression-source.test.ts | 230 ++++++++++++++++++++++++++++++++ test/fixtures/stub-ffmpeg.mjs | 7 + 9 files changed, 770 insertions(+), 61 deletions(-) create mode 100644 src/compression/source.ts create mode 100644 test/compression-source.test.ts create mode 100644 test/fixtures/stub-ffmpeg.mjs 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));