Add AppendOnlyStreamSource, adjust examples to use it

docs/guide/quick-start.md

@@ -308,17 +308,14 @@ await output.finalize();
 ```ts
 import {
 	Output,
-	StreamTarget,
+	AppendOnlyStreamTarget,
-	StreamTargetChunk,
 	Mp4OutputFormat,
 } from 'mediabunny';
 
-const { writable, readable } = new TransformStream<StreamTargetChunk, Uint8Array>({
+const { writable, readable } = new TransformStream<Uint8Array, Uint8Array>();
-	transform: (chunk, controller) => controller.enqueue(chunk.data),
-});
 
 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' }),
 });

docs/guide/writing-hls.md

@@ -71,14 +71,9 @@ const writtenFiles = new Map<string, ArrayBuffer>();
 const output = new Output({
 	target: new PathedTarget(
 		'master.m3u8',
-		({ path }) => {
+		({ path }) => new BufferTarget({
-			const target = new BufferTarget();
+			onFinalized: buffer => writtenFiles.set(path, buffer),
-			target.on('finalized', () => {
+		}),
-				writtenFiles.set(path, target.buffer!);
-			});
-
-			return target;
-		},
 	),
 	// ...
 });
@@ -127,12 +122,8 @@ const output = new Output({
 		'master.m3u8',
 		async ({ path, mimeType }) => {
 			const { writable, readable } = new TransformStream<
-				StreamTargetChunk,
+				Uint8Array, Uint8Array,
-				Uint8Array
+			>();
-			>({
-				transform: (chunk, controller) =>
-					controller.enqueue(chunk.data),
-			});
 
 			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),

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.

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);

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();
+		this.writer = await this.output._getRootWriter(!!this.format._options.appendOnly);
-		if (this.format._options.appendOnly) {
-			this.writer.ensureMonotonicity();
-		}
-
 		this.writer.write(FLAC_HEADER);
 
 		release();

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();
 			}
 

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;

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();

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)) {

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 = await this.output._getRootWriter(true);
-		this.writer.ensureMonotonicity();
 
 		release();
 	}

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 = await this.output._getRootWriter(true); // Ogg is always monotonically written!
-		this.writer.ensureMonotonicity(); // Ogg is always monotonically written!
 
 		release();
 	}

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;
 		})();

src/target.ts

@@ -42,7 +42,7 @@ export abstract class Target extends EventEmitter<TargetEvents> {
 	_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<TargetEvents> {
 	 */
 	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<Uint8Array>;
+	/** @internal */
+	_streamTarget: StreamTarget;
+	/** @internal */
+	_writer: WritableStreamDefaultWriter<Uint8Array> | null = null;
+	/** @internal */
+	_nextWritePos = 0;
+
+	constructor(writable: WritableStream<Uint8Array>) {
+		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<void> {
+		return this._streamTarget._flush();
+	}
+
+	/** @internal */
+	_finalize(): Promise<void> {
+		return this._streamTarget._finalize();
+	}
+
+	/** @internal */
+	_close(): Promise<void> {
+		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);
+	}
 }
 
 /**

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

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);

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<string, number>();
+
+	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<Uint8Array>({
+				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<string, number>();
+
+	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<Uint8Array>({
+				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<string, number>();
+
+	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<Uint8Array>({
+				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<string, number>();
+
+	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<Uint8Array>({
+				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');
+});

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.