From 783cfa6fbe7576adb1d523f74f4bf98dd2a26f0c Mon Sep 17 00:00:00 2001 From: Vanilagy <1696106+Vanilagy@users.noreply.github.com> Date: Tue, 21 Apr 2026 14:58:33 +0200 Subject: [PATCH] Add AppendOnlyStreamSource, adjust examples to use it --- docs/guide/quick-start.md | 9 +- docs/guide/writing-hls.md | 22 ++-- docs/guide/writing-media-files.md | 24 ++++ src/adts/adts-muxer.ts | 2 +- src/flac/flac-muxer.ts | 6 +- src/hls/hls-muxer.ts | 6 +- src/isobmff/isobmff-muxer.ts | 18 ++- src/matroska/matroska-muxer.ts | 6 +- src/mp3/mp3-muxer.ts | 2 +- src/mpeg-ts/mpeg-ts-muxer.ts | 3 +- src/ogg/ogg-muxer.ts | 3 +- src/output.ts | 4 +- src/target.ts | 97 +++++++++++++++- src/wave/wave-muxer.ts | 2 +- src/writer.ts | 8 +- test/node/hls-output.test.ts | 179 +++++++++++++++++++++++++++++- todo.txt | 5 - 17 files changed, 328 insertions(+), 68 deletions(-) diff --git a/docs/guide/quick-start.md b/docs/guide/quick-start.md index fc6f364..d7e5893 100644 --- a/docs/guide/quick-start.md +++ b/docs/guide/quick-start.md @@ -308,17 +308,14 @@ await output.finalize(); ```ts import { Output, - StreamTarget, - StreamTargetChunk, + AppendOnlyStreamTarget, Mp4OutputFormat, } from 'mediabunny'; -const { writable, readable } = new TransformStream({ - transform: (chunk, controller) => controller.enqueue(chunk.data), -}); +const { writable, readable } = new TransformStream(); const output = new Output({ - target: new StreamTarget(writable), + target: new AppendOnlyStreamTarget(writable), // We must use an append-only format here, such as fragmented MP4 format: new Mp4OutputFormat({ fastStart: 'fragmented' }), }); diff --git a/docs/guide/writing-hls.md b/docs/guide/writing-hls.md index 741909e..da030a6 100644 --- a/docs/guide/writing-hls.md +++ b/docs/guide/writing-hls.md @@ -71,14 +71,9 @@ const writtenFiles = new Map(); const output = new Output({ target: new PathedTarget( 'master.m3u8', - ({ path }) => { - const target = new BufferTarget(); - target.on('finalized', () => { - writtenFiles.set(path, target.buffer!); - }); - - return target; - }, + ({ path }) => new BufferTarget({ + onFinalized: buffer => writtenFiles.set(path, buffer), + }), ), // ... }); @@ -127,12 +122,8 @@ const output = new Output({ 'master.m3u8', async ({ path, mimeType }) => { const { writable, readable } = new TransformStream< - StreamTargetChunk, - Uint8Array - >({ - transform: (chunk, controller) => - controller.enqueue(chunk.data), - }); + Uint8Array, Uint8Array, + >(); const url = `/upload?file=${encodeURIComponent(path)}`; const promise = fetch(url, { @@ -145,7 +136,8 @@ const output = new Output({ }); promises.push(promise); - return new StreamTarget(writable); + // Requires that all segments use an append-only format + return new AppendOnlyStreamTarget(writable); }, ), onFinalize: () => Promise.all(promises), diff --git a/docs/guide/writing-media-files.md b/docs/guide/writing-media-files.md index 256e9c5..9cbbdea 100644 --- a/docs/guide/writing-media-files.md +++ b/docs/guide/writing-media-files.md @@ -388,6 +388,30 @@ const output = new Output({ await output.finalize(); // Will automatically close the writable stream ``` +### `AppendOnlyStreamTarget` + +Similar to a `StreamTarget` but for writing files in a purely append-only way: +```ts +import { Output, AppendOnlyStreamTarget } from 'mediabunny'; + +const writable = new WritableStream({ + write(data: Uint8Array) { + // Do something with the data... + }, +}); + +const output = new Output({ + target: new AppendOnlyStreamTarget(writable), + // ... +}); +``` + +Useful for consumers that can only read sequentially, like an HTTP server processing an incoming upload. + +::: warning +The underlying data source doesn't magically become append-only just because you use this source. Instead, you can only use this source when the underlying format is *append-only*. See [Output formats](./output-formats) to see which formats are append-only. +::: + ### `FilePathTarget` This target writes to a file at the specified path. It is intended for server-side usage in Node, Bun, or Deno, and offers a simpler API than `StreamTarget` when you just want to write directly to a file path. diff --git a/src/adts/adts-muxer.ts b/src/adts/adts-muxer.ts index 2e28864..def99c2 100644 --- a/src/adts/adts-muxer.ts +++ b/src/adts/adts-muxer.ts @@ -34,7 +34,7 @@ export class AdtsMuxer extends Muxer { async start() { const release = await this.mutex.acquire(); - this.writer = await this.output._getRootWriter(); + this.writer = await this.output._getRootWriter(true); if (!metadataTagsAreEmpty(this.output._metadataTags)) { const id3Writer = new Id3V2Writer(this.writer); diff --git a/src/flac/flac-muxer.ts b/src/flac/flac-muxer.ts index e3bcb5a..b022dbe 100644 --- a/src/flac/flac-muxer.ts +++ b/src/flac/flac-muxer.ts @@ -54,11 +54,7 @@ export class FlacMuxer extends Muxer { async start() { const release = await this.mutex.acquire(); - this.writer = await this.output._getRootWriter(); - if (this.format._options.appendOnly) { - this.writer.ensureMonotonicity(); - } - + this.writer = await this.output._getRootWriter(!!this.format._options.appendOnly); this.writer.write(FLAC_HEADER); release(); diff --git a/src/hls/hls-muxer.ts b/src/hls/hls-muxer.ts index adb3498..a79c383 100644 --- a/src/hls/hls-muxer.ts +++ b/src/hls/hls-muxer.ts @@ -1225,7 +1225,7 @@ export class HlsMuxer extends Muxer { isRoot: false, mimeType: HLS_MIME_TYPE, }); - const writer = new Writer(target); + const writer = new Writer(target, true); writer.start(); writer.write(textEncoder.encode(playlistText)); @@ -1426,7 +1426,7 @@ export class HlsMuxer extends Muxer { if (this.numWrittenMasterPlaylists === 0) { // For the first master playlist write, we use the normal root writer getter, so that the target // returned by Output.target emits valid write events. - writer = await this.output._getRootWriter(); + writer = await this.output._getRootWriter(true); } else { // For subsequent master playlist writes, we *must* obtain a different target in order to overwrite // the file. @@ -1435,7 +1435,7 @@ export class HlsMuxer extends Muxer { isRoot: true, mimeType: HLS_MIME_TYPE, }); - writer = new Writer(target); + writer = new Writer(target, true); writer.start(); } diff --git a/src/isobmff/isobmff-muxer.ts b/src/isobmff/isobmff-muxer.ts index 0edc52e..e40e21e 100644 --- a/src/isobmff/isobmff-muxer.ts +++ b/src/isobmff/isobmff-muxer.ts @@ -176,7 +176,7 @@ export class IsobmffMuxer extends Muxer { isCmaf: boolean; private auxTarget = new BufferTarget(); - private auxWriter = new Writer(this.auxTarget); + private auxWriter = new Writer(this.auxTarget, false); private auxBoxWriter = new IsobmffBoxWriter(this.auxWriter); private mdat: Box | null = null; @@ -210,7 +210,11 @@ export class IsobmffMuxer extends Muxer { const release = await this.mutex.acquire(); if (!this.isCmaf) { - this.writer = await this.output._getRootWriter(); + this.writer = await this.output._getRootWriter(target => ( + this.format._options.fastStart !== undefined + ? this.format._options.fastStart === 'fragmented' + : target instanceof BufferTarget // Since if this is the case we'll use 'in-memory' + )); this.boxWriter = new IsobmffBoxWriter(this.writer); // If the fastStart option isn't defined, enable in-memory fast start if the target is an ArrayBuffer, as @@ -223,10 +227,6 @@ export class IsobmffMuxer extends Muxer { this.isFragmented = true; } - if (this.fastStart === 'in-memory' || this.isFragmented) { - this.writer?.ensureMonotonicity(); - } - if (this.isCmaf) { if (!this.output._hasInitTarget()) { throw new Error( @@ -237,7 +237,7 @@ export class IsobmffMuxer extends Muxer { // Set up the init writer to which we'll write the init segment const initTarget = await this.output._getInitTarget(); - const initWriter = new Writer(initTarget); + const initWriter = new Writer(initTarget, true); initWriter.start(); this.initWriter = initWriter; @@ -1165,11 +1165,9 @@ export class IsobmffMuxer extends Muxer { // Only now, init the main writer; this way the init writer is fully done before the main writer is // even acquired - this.writer = await this.output._getRootWriter(); + this.writer = await this.output._getRootWriter(true); this.boxWriter = new IsobmffBoxWriter(this.writer); - this.writer.ensureMonotonicity(); - const stypSize = this.boxWriter.measureBox(styp()); const sidxSize = this.boxWriter.measureBox(sidx(this, 0)); this.segmentHeaderSize = stypSize + sidxSize; diff --git a/src/matroska/matroska-muxer.ts b/src/matroska/matroska-muxer.ts index 9df5e1f..da60565 100644 --- a/src/matroska/matroska-muxer.ts +++ b/src/matroska/matroska-muxer.ts @@ -165,13 +165,9 @@ export class MatroskaMuxer extends Muxer { async start() { const release = await this.mutex.acquire(); - this.writer = await this.output._getRootWriter(); + this.writer = await this.output._getRootWriter(!!this.format._options.appendOnly); this.ebmlWriter = new EBMLWriter(this.writer); - if (this.format._options.appendOnly) { - this.writer.ensureMonotonicity(); - } - this.writeEBMLHeader(); this.createSegmentInfo(); diff --git a/src/mp3/mp3-muxer.ts b/src/mp3/mp3-muxer.ts index 17046cc..97546b3 100644 --- a/src/mp3/mp3-muxer.ts +++ b/src/mp3/mp3-muxer.ts @@ -35,7 +35,7 @@ export class Mp3Muxer extends Muxer { async start() { const release = await this.mutex.acquire(); - this.writer = await this.output._getRootWriter(); + this.writer = await this.output._getRootWriter(this.format._options.xingHeader === false); this.mp3Writer = new Mp3Writer(this.writer); if (!metadataTagsAreEmpty(this.output._metadataTags)) { diff --git a/src/mpeg-ts/mpeg-ts-muxer.ts b/src/mpeg-ts/mpeg-ts-muxer.ts index 3cc7fc9..dfa0f7e 100644 --- a/src/mpeg-ts/mpeg-ts-muxer.ts +++ b/src/mpeg-ts/mpeg-ts-muxer.ts @@ -96,8 +96,7 @@ export class MpegTsMuxer extends Muxer { async start() { const release = await this.mutex.acquire(); - this.writer = await this.output._getRootWriter(); - this.writer.ensureMonotonicity(); + this.writer = await this.output._getRootWriter(true); release(); } diff --git a/src/ogg/ogg-muxer.ts b/src/ogg/ogg-muxer.ts index 931f087..987eef8 100644 --- a/src/ogg/ogg-muxer.ts +++ b/src/ogg/ogg-muxer.ts @@ -79,8 +79,7 @@ export class OggMuxer extends Muxer { async start() { const release = await this.mutex.acquire(); - this.writer = await this.output._getRootWriter(); - this.writer.ensureMonotonicity(); // Ogg is always monotonically written! + this.writer = await this.output._getRootWriter(true); // Ogg is always monotonically written! release(); } diff --git a/src/output.ts b/src/output.ts index 927dc18..e1f7b67 100644 --- a/src/output.ts +++ b/src/output.ts @@ -580,11 +580,11 @@ export class Output< } /** @internal */ - _getRootWriter() { + _getRootWriter(isMonotonic: boolean | ((target: Target) => boolean)) { return this._rootWriterPromise ??= (async () => { const target = await this._getRootTarget(); - const writer = new Writer(target); + const writer = new Writer(target, typeof isMonotonic === 'boolean' ? isMonotonic : isMonotonic(target)); writer.start(); return writer; })(); diff --git a/src/target.ts b/src/target.ts index 05cfe2f..ee008cb 100644 --- a/src/target.ts +++ b/src/target.ts @@ -42,7 +42,7 @@ export abstract class Target extends EventEmitter { _output: Output | null = null; /** @internal */ - _ensureMonotonicity = false; + _monotonicity: boolean | null = null; // null = unknown /** @internal */ abstract _start(): void; @@ -65,6 +65,15 @@ export abstract class Target extends EventEmitter { */ onwrite: ((start: number, end: number) => unknown) | null = null; + /** @internal */ + _setMonotonicity(monotonicity: boolean) { + if (this._monotonicity !== false) { + this._monotonicity = monotonicity; + } else { + // Once false, it's locked + } + } + /** @internal */ _dispatchWrite(start: number, end: number) { // eslint-disable-next-line @typescript-eslint/no-deprecated @@ -418,7 +427,7 @@ export class StreamTarget extends Target { this._writeDataIntoChunks(chunk.data, chunk.start); this._tryToFlushChunks(); } else { - if (this._ensureMonotonicity && chunk.start !== this._lastFlushEnd) { + if (this._monotonicity === true && chunk.start !== this._lastFlushEnd) { throw new Error('Internal error: Monotonicity violation.'); } @@ -530,7 +539,7 @@ export class StreamTarget extends Target { for (const section of chunk.written) { const position = chunk.start + section.start; - if (this._ensureMonotonicity && position !== this._lastFlushEnd) { + if (this._monotonicity === true && position !== this._lastFlushEnd) { throw new Error('Internal error: Monotonicity violation.'); } @@ -573,6 +582,76 @@ export class StreamTarget extends Target { } } +export class AppendOnlyStreamTarget extends Target { + /** @internal */ + _writable: WritableStream; + /** @internal */ + _streamTarget: StreamTarget; + /** @internal */ + _writer: WritableStreamDefaultWriter | null = null; + /** @internal */ + _nextWritePos = 0; + + constructor(writable: WritableStream) { + super(); + + this._writable = writable; + this._streamTarget = new StreamTarget(new WritableStream({ + start: () => { + this._writer = this._writable.getWriter(); + }, + write: (chunk) => { + if (this._monotonicity !== true) { + throw new Error( + 'AppendOnlyStreamTarget requires that data be written monotonically (always appended to the' + + ' end). You must use a format that guarantees this behavior.', + ); + } + + assert(chunk.position === this._nextWritePos); + this._nextWritePos += chunk.data.byteLength; + + assert(this._writer); + return this._writer.write(chunk.data); + }, + close: () => { + return this._writer?.close(); + }, + })); + } + + /** @internal */ + _start(): void { + this._streamTarget._start(); + } + + /** @internal */ + _write(data: Uint8Array, pos: number): void { + this._streamTarget._write(data, pos); + } + + /** @internal */ + _flush(): Promise { + return this._streamTarget._flush(); + } + + /** @internal */ + _finalize(): Promise { + return this._streamTarget._finalize(); + } + + /** @internal */ + _close(): Promise { + return this._streamTarget._close(); + } + + /** @internal */ + override _setMonotonicity(monotonicity: boolean): void { + super._setMonotonicity(monotonicity); + this._streamTarget._setMonotonicity(monotonicity); + } +} + /** * Options for {@link FilePathTarget}. * @group Output targets @@ -654,6 +733,12 @@ export class FilePathTarget extends Target { async _close() { return this._streamTarget._close(); } + + /** @internal */ + override _setMonotonicity(monotonicity: boolean): void { + super._setMonotonicity(monotonicity); + this._streamTarget._setMonotonicity(monotonicity); + } } /** @@ -726,6 +811,12 @@ export class RangedTarget extends Target { /** @internal */ async _close() {} + + /** @internal */ + override _setMonotonicity(monotonicity: boolean): void { + super._setMonotonicity(monotonicity); + this._baseTarget._setMonotonicity(monotonicity); + } } /** diff --git a/src/wave/wave-muxer.ts b/src/wave/wave-muxer.ts index 26319d7..7e43a0f 100644 --- a/src/wave/wave-muxer.ts +++ b/src/wave/wave-muxer.ts @@ -44,7 +44,7 @@ export class WaveMuxer extends Muxer { async start() { const release = await this.mutex.acquire(); - this.writer = await this.output._getRootWriter(); + this.writer = await this.output._getRootWriter(false); this.riffWriter = new RiffWriter(this.writer); // No writing needed here - we'll write the header with the first sample diff --git a/src/writer.ts b/src/writer.ts index 43c1622..2f610fd 100644 --- a/src/writer.ts +++ b/src/writer.ts @@ -16,8 +16,9 @@ export class Writer { private pos = 0; - constructor(target: Target) { + constructor(target: Target, isMonotonic: boolean) { this.target = target; + target._setMonotonicity(isMonotonic); } start() { @@ -26,11 +27,6 @@ export class Writer { this.started = true; } - ensureMonotonicity() { - this.target._ensureMonotonicity = true; - // Note that this currently is without effect for RangedTarget. But, should be fine since its use is rare - } - /** Writes the given data to the target, at the current position. */ write(data: Uint8Array) { assert(this.started && !this.finalized); diff --git a/test/node/hls-output.test.ts b/test/node/hls-output.test.ts index 3454537..6777b80 100644 --- a/test/node/hls-output.test.ts +++ b/test/node/hls-output.test.ts @@ -4,9 +4,17 @@ import { CmafOutputFormat, HlsOutputFormat, HlsOutputSegmentInfo, + Mp4OutputFormat, MpegTsOutputFormat, } from '../../src/output-format.js'; -import { BufferTarget, NullTarget, PathedTarget, StreamTarget, StreamTargetChunk } from '../../src/target.js'; +import { + AppendOnlyStreamTarget, + BufferTarget, + NullTarget, + PathedTarget, + StreamTarget, + StreamTargetChunk, +} from '../../src/target.js'; import { EncodedAudioPacketSource, EncodedVideoPacketSource } from '../../src/media-source.js'; import { HlsMuxer } from '../../src/hls/hls-muxer.js'; import { AudioCodec, VideoCodec } from '../../src/codec.js'; @@ -2726,3 +2734,172 @@ test('Live mode, maxLiveSegmentCount with singleFilePerPlaylist', async () => { const extinfCount = (lastPlaylistText.match(/#EXTINF:/g) ?? []).length; expect(extinfCount).toBe(2); }); + +test('Append-only stream', async () => { + const writes = new Map(); + + const output = new Output({ + format: new HlsOutputFormat({ + segmentFormat: new MpegTsOutputFormat(), + }), + target: new PathedTarget('master.m3u8', (request) => { + writes.set(request.path, 0); + + const writable = new WritableStream({ + write: () => { + writes.set(request.path, writes.get(request.path)! + 1); + }, + }); + const target = new AppendOnlyStreamTarget(writable); + return target; + }), + }); + + const source = videoSource(); + output.addVideoTrack(source); + + await output.start(); + + await source.add(new EncodedPacket(avcPacketData, 'key', 0, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 0.5, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 1, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 1.5, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'key', 2, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 2.5, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 3, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 3.5, 0), avcMetadata); + + await output.finalize(); + + // Each segment file should have been written to at least once + for (const [, count] of writes) { + expect(count).toBeGreaterThanOrEqual(1); + } + + expect(writes.size).toBe(1 + 1 + 2); // Master playlist + media playlist + 2 segments +}); + +test('Append-only stream, single file', async () => { + const writes = new Map(); + + const output = new Output({ + format: new HlsOutputFormat({ + segmentFormat: new MpegTsOutputFormat(), + singleFilePerPlaylist: true, + }), + target: new PathedTarget('master.m3u8', (request) => { + writes.set(request.path, 0); + + const writable = new WritableStream({ + write: () => { + writes.set(request.path, writes.get(request.path)! + 1); + }, + }); + const target = new AppendOnlyStreamTarget(writable); + return target; + }), + }); + + const source = videoSource(); + output.addVideoTrack(source); + + await output.start(); + + await source.add(new EncodedPacket(avcPacketData, 'key', 0, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 0.5, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 1, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 1.5, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'key', 2, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 2.5, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 3, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 3.5, 0), avcMetadata); + + await output.finalize(); + + // Each segment file should have been written to at least once + for (const [, count] of writes) { + expect(count).toBeGreaterThanOrEqual(1); + } + + expect(writes.size).toBe(1 + 1 + 1); // Master playlist + media playlist + 1 segments file +}); + +test('Append-only stream, single file with CMAF', async () => { + const writes = new Map(); + + const output = new Output({ + format: new HlsOutputFormat({ + segmentFormat: new CmafOutputFormat(), + singleFilePerPlaylist: true, + }), + target: new PathedTarget('master.m3u8', (request) => { + writes.set(request.path, 0); + + const writable = new WritableStream({ + write: () => { + writes.set(request.path, writes.get(request.path)! + 1); + }, + }); + const target = new AppendOnlyStreamTarget(writable); + return target; + }), + }); + + const source = videoSource(); + output.addVideoTrack(source); + + await output.start(); + + await source.add(new EncodedPacket(avcPacketData, 'key', 0, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 0.5, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 1, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 1.5, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'key', 2, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 2.5, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 3, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 3.5, 0), avcMetadata); + + await output.finalize(); + + // Each segment file should have been written to at least once + for (const [, count] of writes) { + expect(count).toBeGreaterThanOrEqual(1); + } + + expect(writes.size).toBe(1 + 1 + 1); // Master playlist + media playlist + 1 segments file +}); + +test('Append-only stream with monotonicity violation', async () => { + const writes = new Map(); + + const output = new Output({ + format: new HlsOutputFormat({ + segmentFormat: new Mp4OutputFormat(), + singleFilePerPlaylist: true, + }), + target: new PathedTarget('master.m3u8', (request) => { + writes.set(request.path, 0); + + const writable = new WritableStream({ + write: () => { + writes.set(request.path, writes.get(request.path)! + 1); + }, + }); + const target = new AppendOnlyStreamTarget(writable); + return target; + }), + }); + + const source = videoSource(); + output.addVideoTrack(source); + + await output.start(); + + await source.add(new EncodedPacket(avcPacketData, 'key', 0, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 0.5, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 1, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 1.5, 0), avcMetadata); + + await expect(source.add(new EncodedPacket(avcPacketData, 'key', 2, 0), avcMetadata)) + .rejects.toThrow('AppendOnlyStreamTarget'); +}); diff --git a/todo.txt b/todo.txt index 2c6b2f8..ae966ed 100644 --- a/todo.txt +++ b/todo.txt @@ -1,8 +1,3 @@ -MENTION APPEND-ONLY IN UPLOAD EXAMPLE IN WRITING HLS - -also I wish there was a more explicit way this was enforced and would error at runtime. -writablestream target? - Thoughts: So, a certain "lookahead" logic is definitely needed. The question is if this is a per-demuxer thing or a general thing instead. The demuxer could get in a "packet query" that specifies things like "I am interested in the next 20 seconds guaranteed", allowing the demuxer to pre-fetch more intelligently. The alternative would be some sort of demuxer-agnostic approach where there is a magical "packet requester" that has to be segment-aware. I'm actually not sure if that's good.