@remotion/webcodecs: Revamp queueing to fix Safari issues

package.json

@@ -12,7 +12,7 @@
 		"testssr": "turbo run testssr --no-update-notifier",
 		"testlambda": "turbo run testlambda --concurrency=1 --no-update-notifier",
 		"ci": "turbo run make test --concurrency=1 --no-update-notifier",
-		"watch": "turbo watch make",
+		"watch": "turbo --no-daemon run make && turbo --no-daemon watch make",
 		"release-alpha": "pnpm recursive publish --tag=alpha --no-git-checks && pnpm recursive run --sequential publishprivate",
 		"release": "pnpm recursive publish && pnpm recursive run --sequential publishprivate && git push --tags && git push",
 		"clean": "turbo run clean && rm -rf packages/**/dist && rm -rf packages/**/node_modules && rm -rf node_modules && rm -rf .cache && rm -rf packages/**/tsconfig.tsbuildinfo && rm -f packages/tsconfig.tsbuildinfo && rm -rf packages/**/.turbo && rm -rf .turbo"

packages/convert/app/components/ConvertUi.tsx

@@ -89,6 +89,7 @@ export default function ConvertUI({src}: {readonly src: Source}) {
 				// eslint-disable-next-line no-alert
 				alert((e as Error).stack);
 				console.error(e);
+				setState({type: 'idle'});
 			});
 
 		return () => {

packages/convert/app/lib/format-seconds.ts

@@ -2,5 +2,5 @@ export const formatSeconds = (seconds: number) => {
 	const minutes = Math.floor(seconds / 60);
 	const secondsLeft = seconds % 60;
 
-	return `${minutes}:${secondsLeft < 10 ? '0' : ''}${Math.round(secondsLeft)} min`;
+	return `${minutes}:${Math.round(secondsLeft) < 10 ? '0' : ''}${Math.round(secondsLeft)} min`;
 };

packages/media-parser/src/index.ts

@@ -1,4 +1,7 @@
 import {createMedia} from './create/create-media';
+import type {LogLevel} from './log';
+import {Log} from './log';
+export {WriterInterface} from './writers/writer';
 
 export {
 	AudioTrack,
@@ -33,6 +36,9 @@ export type {MediaFn} from './create/create-media';
 export {Dimensions} from './get-dimensions';
 export type {ReaderInterface} from './readers/reader';
 
+export type {LogLevel};
+
 export const MediaParserInternals = {
 	createMedia,
+	Log,
 };

packages/media-parser/src/log.ts

@@ -0,0 +1,41 @@
+/* eslint-disable no-console */
+export const logLevels = ['trace', 'verbose', 'info', 'warn', 'error'] as const;
+
+export type LogLevel = (typeof logLevels)[number];
+
+const getNumberForLogLevel = (level: LogLevel) => {
+	return logLevels.indexOf(level);
+};
+
+export const isEqualOrBelowLogLevel = (
+	currentLevel: LogLevel,
+	level: LogLevel,
+) => {
+	return getNumberForLogLevel(currentLevel) <= getNumberForLogLevel(level);
+};
+
+export const Log = {
+	trace: (logLevel: LogLevel, ...args: Parameters<typeof console.log>) => {
+		if (isEqualOrBelowLogLevel(logLevel, 'trace')) {
+			return console.log(...args);
+		}
+	},
+	verbose: (logLevel: LogLevel, ...args: Parameters<typeof console.log>) => {
+		if (isEqualOrBelowLogLevel(logLevel, 'verbose')) {
+			return console.log(...args);
+		}
+	},
+	info: (logLevel: LogLevel, ...args: Parameters<typeof console.log>) => {
+		if (isEqualOrBelowLogLevel(logLevel, 'info')) {
+			return console.log(...args);
+		}
+	},
+	warn: (logLevel: LogLevel, ...args: Parameters<typeof console.log>) => {
+		if (isEqualOrBelowLogLevel(logLevel, 'warn')) {
+			return console.warn(...args);
+		}
+	},
+	error: (...args: Parameters<typeof console.log>) => {
+		return console.error(...args);
+	},
+};

packages/webcodecs/src/audio-decoder.ts

@@ -1,10 +1,10 @@
-import type {AudioSample} from '@remotion/media-parser';
+import type {AudioSample, LogLevel} from '@remotion/media-parser';
+import {makeIoSynchronizer} from './io-manager/io-synchronizer';
 
 export type WebCodecsAudioDecoder = {
 	processSample: (audioSample: AudioSample) => Promise<void>;
 	waitForFinish: () => Promise<void>;
 	close: () => void;
-	getQueueSize: () => number;
 	flush: () => Promise<void>;
 };
 
@@ -13,29 +13,46 @@ export const createAudioDecoder = ({
 	onError,
 	signal,
 	config,
+	logLevel,
 }: {
 	onFrame: (frame: AudioData) => Promise<void>;
 	onError: (error: DOMException) => void;
 	signal: AbortSignal;
 	config: AudioDecoderConfig;
+	logLevel: LogLevel;
 }): WebCodecsAudioDecoder => {
 	if (signal.aborted) {
 		throw new Error('Not creating audio decoder, already aborted');
 	}
 
+	const ioSynchronizer = makeIoSynchronizer(logLevel, 'Audio decoder');
+
 	let outputQueue = Promise.resolve();
-	let outputQueueSize = 0;
-	let dequeueResolver = () => {};
 
 	const audioDecoder = new AudioDecoder({
 		output(inputFrame) {
-			outputQueueSize++;
+			ioSynchronizer.onOutput(inputFrame.timestamp);
+			const abortHandler = () => {
+				inputFrame.close();
+			};
+
+			signal.addEventListener('abort', abortHandler, {once: true});
 			outputQueue = outputQueue
-				.then(() => onFrame(inputFrame))
 				.then(() => {
-					dequeueResolver();
-					outputQueueSize--;
+					if (signal.aborted) {
+						return;
+					}
+
+					return onFrame(inputFrame);
+				})
+				.then(() => {
+					ioSynchronizer.onProcessed();
+					signal.removeEventListener('abort', abortHandler);
 					return Promise.resolve();
+				})
+				.catch((err) => {
+					inputFrame.close();
+					onError(err);
 				});
 		},
 		error(error) {
@@ -60,41 +77,20 @@ export const createAudioDecoder = ({
 
 	signal.addEventListener('abort', onAbort);
 
-	const getQueueSize = () => {
-		return audioDecoder.decodeQueueSize + outputQueueSize;
-	};
-
 	audioDecoder.configure(config);
 
-	const waitForDequeue = async () => {
-		await new Promise<void>((r) => {
-			dequeueResolver = r;
-			// @ts-expect-error exists
-			audioDecoder.addEventListener('dequeue', () => r(), {
-				once: true,
-			});
-		});
-	};
-
-	const waitForFinish = async () => {
-		while (getQueueSize() > 0) {
-			await waitForDequeue();
-		}
-	};
-
 	const processSample = async (audioSample: AudioSample) => {
 		if (audioDecoder.state === 'closed') {
 			return;
 		}
 
-		while (getQueueSize() > 10) {
-			await waitForDequeue();
-		}
+		await ioSynchronizer.waitFor({unemitted: 100, _unprocessed: 2});
 
 		// Don't flush, it messes up the audio
 
 		const chunk = new EncodedAudioChunk(audioSample);
 		audioDecoder.decode(chunk);
+		ioSynchronizer.inputItem(chunk.timestamp, audioSample.type === 'key');
 	};
 
 	let queue = Promise.resolve();
@@ -106,11 +102,10 @@ export const createAudioDecoder = ({
 		},
 		waitForFinish: async () => {
 			await audioDecoder.flush();
-			await waitForFinish();
+			await ioSynchronizer.waitForFinish();
 			await outputQueue;
 		},
 		close,
-		getQueueSize,
 		flush: async () => {
 			await audioDecoder.flush();
 		},

packages/webcodecs/src/audio-encoder.ts

@@ -1,10 +1,11 @@
+import type {LogLevel} from '@remotion/media-parser';
 import type {ConvertMediaAudioCodec} from './codec-id';
+import {makeIoSynchronizer} from './io-manager/io-synchronizer';
 
 export type WebCodecsAudioEncoder = {
 	encodeFrame: (audioData: AudioData) => Promise<void>;
 	waitForFinish: () => Promise<void>;
 	close: () => void;
-	getQueueSize: () => number;
 	flush: () => Promise<void>;
 };
 
@@ -14,30 +15,40 @@ export const createAudioEncoder = ({
 	codec,
 	signal,
 	config: audioEncoderConfig,
+	logLevel,
 }: {
 	onChunk: (chunk: EncodedAudioChunk) => Promise<void>;
 	onError: (error: DOMException) => void;
 	codec: ConvertMediaAudioCodec;
 	signal: AbortSignal;
 	config: AudioEncoderConfig;
+	logLevel: LogLevel;
 }): WebCodecsAudioEncoder => {
 	if (signal.aborted) {
 		throw new Error('Not creating audio encoder, already aborted');
 	}
 
+	const ioSynchronizer = makeIoSynchronizer(logLevel, 'Audio encoder');
+
 	let prom = Promise.resolve();
-	let outputQueue = 0;
-	let dequeueResolver = () => {};
 
 	const encoder = new AudioEncoder({
 		output: (chunk) => {
-			outputQueue++;
+			ioSynchronizer.onOutput(chunk.timestamp);
 			prom = prom
-				.then(() => onChunk(chunk))
 				.then(() => {
-					outputQueue--;
-					dequeueResolver();
+					if (signal.aborted) {
+						return;
+					}
+
+					return onChunk(chunk);
+				})
+				.then(() => {
+					ioSynchronizer.onProcessed();
 					return Promise.resolve();
+				})
+				.catch((err) => {
+					onError(err);
 				});
 		},
 		error(error) {
@@ -65,44 +76,22 @@ export const createAudioEncoder = ({
 		throw new Error('Only `codec: "opus"` is supported currently');
 	}
 
-	const getQueueSize = () => {
-		return encoder.encodeQueueSize + outputQueue;
-	};
-
 	encoder.configure(audioEncoderConfig);
 
-	const waitForDequeue = async () => {
-		await new Promise<void>((r) => {
-			dequeueResolver = r;
-
-			// @ts-expect-error exists
-			encoder.addEventListener('dequeue', () => r(), {
-				once: true,
-			});
-		});
-	};
-
-	const waitForFinish = async () => {
-		while (getQueueSize() > 0) {
-			await waitForDequeue();
-		}
-	};
-
 	const encodeFrame = async (audioData: AudioData) => {
 		if (encoder.state === 'closed') {
 			return;
 		}
 
-		while (getQueueSize() > 10) {
-			await waitForDequeue();
-		}
+		await ioSynchronizer.waitFor({unemitted: 2, _unprocessed: 2});
 
 		// @ts-expect-error - can have changed in the meanwhile
 		if (encoder.state === 'closed') {
 			return;
 		}
 
 		encoder.encode(audioData);
+		ioSynchronizer.inputItem(audioData.timestamp, true);
 	};
 
 	let queue = Promise.resolve();
@@ -114,11 +103,10 @@ export const createAudioEncoder = ({
 		},
 		waitForFinish: async () => {
 			await encoder.flush();
-			await waitForFinish();
+			await ioSynchronizer.waitForFinish();
 			await prom;
 		},
 		close,
-		getQueueSize,
 		flush: async () => {
 			await encoder.flush();
 		},

packages/webcodecs/src/auto-select-writer.ts

@@ -0,0 +1,28 @@
+import {bufferWriter} from '@remotion/media-parser/buffer';
+import {canUseWebFsWriter, webFsWriter} from '@remotion/media-parser/web-fs';
+
+import type {WriterInterface} from '@remotion/media-parser';
+import type {LogLevel} from './log';
+import {Log} from './log';
+
+export const autoSelectWriter = async (
+	writer: WriterInterface | undefined,
+	logLevel: LogLevel,
+): Promise<WriterInterface> => {
+	if (writer) {
+		Log.verbose(logLevel, 'Using writer provided by user');
+		return writer;
+	}
+
+	Log.verbose(logLevel, 'Determining best writer');
+	if (await canUseWebFsWriter()) {
+		Log.verbose(logLevel, 'Using WebFS writer because it is supported');
+		return webFsWriter;
+	}
+
+	Log.verbose(
+		logLevel,
+		'Using buffer writer because WebFS writer is not supported',
+	);
+	return bufferWriter;
+};