fix: surface StreamTarget write errors instead of swallowing them (#305)

* fix: await StreamTarget writes to prevent overlapping OPFS operations

* no need for flush

* preserve fire and forget writes

* Remove pendingWrites, add missing mutex acquire to ADTS muxer finalize method

---------

Co-authored-by: Vanilagy <1696106+Vanilagy@users.noreply.github.com>

Author: samohovets

src/adts/adts-muxer.ts

@@ -111,5 +111,8 @@ export class AdtsMuxer extends Muxer {
 		throw new Error('ADTS does not support subtitles.');
 	}
 
-	async finalize() {}
+	async finalize() {
+		const release = await this.mutex.acquire(); // Required so that finalize() can't resolve before other calls
+		release();
+	}
 }

src/writer.ts

@@ -215,6 +215,7 @@ export class StreamTargetWriter extends Writer {
 	private lastWriteEnd = 0;
 	private lastFlushEnd = 0;
 	private writer: WritableStreamDefaultWriter<StreamTargetChunk> | null = null;
+	private writeError: unknown = null;
 
 	// These variables regard chunked mode:
 	private chunked: boolean;
@@ -267,6 +268,11 @@ export class StreamTargetWriter extends Writer {
 	}
 
 	async flush() {
+		if (this.writeError !== null) {
+			// eslint-disable-next-line @typescript-eslint/only-throw-error
+			throw this.writeError;
+		}
+
 		if (this.pos > this.lastWriteEnd) {
 			// There's a "void" between the last written byte and the next byte we're about to write. Let's pad that
 			// void with zeroes explicitly.
@@ -329,11 +335,12 @@ export class StreamTargetWriter extends Writer {
 					throw new Error('Internal error: Monotonicity violation.');
 				}
 
-				// Write out the data immediately
 				void this.writer.write({
 					type: 'write',
 					data: chunk.data,
 					position: chunk.start,
+				}).catch((error) => {
+					this.writeError ??= error;
 				});
 
 				this.lastFlushEnd = chunk.start + chunk.data.byteLength;
@@ -440,6 +447,8 @@ export class StreamTargetWriter extends Writer {
 					type: 'write',
 					data: chunk.data.subarray(section.start, section.end),
 					position,
+				}).catch((error) => {
+					this.writeError ??= error;
 				});
 
 				this.lastFlushEnd = chunk.start + section.end;
@@ -449,12 +458,18 @@ export class StreamTargetWriter extends Writer {
 		}
 	}
 
-	finalize() {
+	async finalize() {
 		if (this.chunked) {
 			this.tryToFlushChunks(true);
 		}
 
+		if (this.writeError !== null) {
+			// eslint-disable-next-line @typescript-eslint/only-throw-error
+			throw this.writeError;
+		}
+
 		assert(this.writer);
+		await this.writer.ready;
 		return this.writer.close();
 	}
 

test/browser/adts-muxing.test.ts

@@ -0,0 +1,130 @@
+import { expect, test } from 'vitest';
+import { Input } from '../../src/input.js';
+import { BufferSource, UrlSource } from '../../src/source.js';
+import { ALL_FORMATS } from '../../src/input-format.js';
+import { EncodedPacketSink } from '../../src/media-sink.js';
+import { EncodedAudioPacketSource } from '../../src/media-source.js';
+import { Output } from '../../src/output.js';
+import { StreamTarget, type StreamTargetChunk } from '../../src/target.js';
+import { AdtsOutputFormat } from '../../src/output-format.js';
+import { assert } from '../../src/misc.js';
+
+const createBufferingStreamTarget = () => {
+	const written = new Map<number, Uint8Array>();
+
+	const stream = new WritableStream<StreamTargetChunk>({
+		async write(chunk: StreamTargetChunk) {
+			written.set(chunk.position, chunk.data.slice());
+		},
+	});
+
+	const toBuffer = () => {
+		let maxEnd = 0;
+		for (const [offset, data] of written) {
+			maxEnd = Math.max(maxEnd, offset + data.byteLength);
+		}
+		const buffer = new Uint8Array(maxEnd);
+		for (const [offset, data] of written) {
+			buffer.set(data, offset);
+		}
+		return buffer;
+	};
+
+	return { stream, toBuffer };
+};
+
+test('ADTS with metadata over StreamTarget', async () => {
+	const target = createBufferingStreamTarget();
+
+	const output = new Output({
+		format: new AdtsOutputFormat(),
+		target: new StreamTarget(target.stream),
+	});
+
+	output.setMetadataTags({ comment: 'Remotion' });
+
+	const audioSource = new EncodedAudioPacketSource('aac');
+	output.addAudioTrack(audioSource);
+
+	await output.start();
+
+	using input = new Input({
+		source: new UrlSource('/sample3.aac'),
+		formats: ALL_FORMATS,
+	});
+
+	const audioTrack = await input.getPrimaryAudioTrack();
+	assert(audioTrack);
+
+	const sink = new EncodedPacketSink(audioTrack);
+
+	let isFirst = true;
+	for await (const packet of sink.packets()) {
+		await audioSource.add(packet, {
+			decoderConfig: isFirst ? (await audioTrack.getDecoderConfig())! : undefined,
+		});
+		isFirst = false;
+	}
+
+	await output.finalize();
+
+	const buffer = target.toBuffer();
+	using outputAsInput = new Input({
+		source: new BufferSource(buffer.buffer),
+		formats: ALL_FORMATS,
+	});
+
+	const readTags = await outputAsInput.getMetadataTags();
+	expect(readTags.comment).toBe('Remotion');
+
+	const outputAudioTrack = await outputAsInput.getPrimaryAudioTrack();
+	assert(outputAudioTrack);
+	expect(outputAudioTrack.codec).toBe('aac');
+});
+
+// Previously, write handler rejections were silently swallowed and surfaced as
+// "Cannot write to a closing writable stream" instead of the actual error.
+test('StreamTarget write errors surface directly', async () => {
+	let writeCount = 0;
+	const stream = new WritableStream<StreamTargetChunk>({
+		async write() {
+			writeCount++;
+			if (writeCount === 2) {
+				throw new Error('OPFS write failed');
+			}
+		},
+	});
+
+	const output = new Output({
+		format: new AdtsOutputFormat(),
+		target: new StreamTarget(stream),
+	});
+
+	const audioSource = new EncodedAudioPacketSource('aac');
+	output.addAudioTrack(audioSource);
+
+	await output.start();
+
+	using input = new Input({
+		source: new UrlSource('/sample3.aac'),
+		formats: ALL_FORMATS,
+	});
+
+	const audioTrack = await input.getPrimaryAudioTrack();
+	assert(audioTrack);
+
+	const sink = new EncodedPacketSink(audioTrack);
+
+	const run = async () => {
+		let isFirst = true;
+		for await (const packet of sink.packets()) {
+			await audioSource.add(packet, {
+				decoderConfig: isFirst ? (await audioTrack.getDecoderConfig())! : undefined,
+			});
+			isFirst = false;
+		}
+		await output.finalize();
+	};
+
+	await expect(run()).rejects.toThrow('OPFS write failed');
+});