Merge pull request #5342 from remotion-dev/decoder-async-methods

Author: JonnyBurger

packages/docs/docs/webcodecs/create-audio-decoder.mdx

@@ -36,7 +36,7 @@ const decoder = createAudioDecoder({
 ## Differences to `AudioDecoder`
 
 - Samples with a `codec` of `pcm-s16` are accepted and passed through, even if the `AudioDecoder` object does not exist or support it.
-- Two new methods are added: [`.waitForQueueToBeLessThan()`](#waitforqueuetobelessthanqueue-size-number) and [`.waitForFinish()`](#waitforfinish).
+- Two new methods are added: [`.waitForQueueToBeLessThan()`](#waitforqueuetobelessthan) and [`.waitForFinish()`](#waitforfinish).
 - The [`dequeue`](https://developer.mozilla.org/en-US/docs/Web/API/AudioDecoder/dequeue_event) event is not supported as it is not reliable across browsers.
 - In addition to [`EncodedAudioChunk`](https://developer.mozilla.org/en-US/docs/Web/API/EncodedAudioChunk), [`EncodedAudioChunkInit`](https://www.w3.org/TR/webcodecs/#dictdef-encodedaudiochunkinit) objects are also accepted for [`.decode()`](#decode).
 - A [`webcodecsController()`](/docs/webcodecs/webcodecs-controller) instance can be passed in to the function, allowing for decoding to be paused, resumed and aborted.
@@ -96,7 +96,9 @@ Decodes a sample. Same as [`AudioDecoder.decode()`](https://developer.mozilla.or
 You can pass in a [`MediaParserAudioSample`](/docs/media-parser/types#mediaparseraudiosample) object from [`parseMedia()`](/docs/media-parser/parse-media), which also satisfies the [`EncodedAudioChunkInit`](https://www.w3.org/TR/webcodecs/#dictdef-encodedaudiochunkinit) interface.
 
 
-### `waitForQueueToBeLessThan(queueSize: number)`
+### `waitForQueueToBeLessThan()`
+
+Pass a number to wait for the queue to be less than the given number.
 
 A promise that resolves when the queue size is less than the given number.  
 The queue is only decremented when the[ `onFrame`](#onframe) callback resolves.
@@ -113,6 +115,15 @@ Clears the queue and resets the decoder. Same as [`AudioDecoder.reset()`](https:
 
 Closes the decoder. Same as [`AudioDecoder.close()`](https://developer.mozilla.org/en-US/docs/Web/API/AudioDecoder/close).
 
+### `checkReset()`<AvailableFrom v="4.0.312" />
+
+Returns a handle with a `wasReset()` function. If the decoder was reset inbetween the call to `.checkReset()` and the call to `wasReset()`, `wasReset()` will return `true`. See [below](#checking-if-the-decoder-was-reset) for an example.
+
+### `getMostRecentSampleInput()`<AvailableFrom v="4.0.312" />
+
+Return the `.timestamp` of the most recently input sample. 
+
+
 ## Example usage with `@remotion/media-parser`
 
 In this example, the whole audio track is decoded and the decoder is closed when the track is done.  
@@ -146,6 +157,52 @@ await parseMedia({
 });
 ```
 
+## Checking if the decoder was reset
+
+A potential race condition you may face is that `decoder.reset()` is called while a sample is waiting for the queue to be less than a certain number. Use `.checkReset()` to check if the decoder was reset after any asynchronous operation, and abort the processing of the sample if needed.
+
+```tsx twoslash title="Check if the decoder was reset"
+import {parseMedia} from '@remotion/media-parser';
+import {createAudioDecoder} from '@remotion/webcodecs';
+
+await parseMedia({
+  src: 'https://parser.media/video.mp4',
+  onAudioTrack: ({track, container}) => {
+
+    const decoder = createAudioDecoder({
+      track,
+      onFrame: console.log,
+      onError: console.error,
+    });
+
+    return async (sample) => {
+      const {wasReset} = decoder.checkReset();
+
+      await decoder.waitForQueueToBeLessThan(10);
+      if (wasReset()) {
+        return
+      }
+
+      await decoder.decode(sample);
+      if (wasReset()) {
+        return;
+      }
+
+
+      return async () => {
+        if (wasReset()) {
+          return;
+        }
+
+        // Called when the track is done
+        await decoder.flush();
+        decoder.close()
+      };
+    };
+  },
+});
+```
+
 
 ## See also
 

packages/docs/docs/webcodecs/create-video-decoder.mdx

@@ -35,7 +35,7 @@ const decoder = createVideoDecoder({
 
 ## Differences to `VideoDecoder`
 
-- Two new methods are added: [`.waitForQueueToBeLessThan()`](#waitforqueuetobelessthanqueue-size-number) and [`.waitForFinish()`](#waitforfinish).
+- Two new methods are added: [`.waitForQueueToBeLessThan()`](#waitforqueuetobelessthan) and [`.waitForFinish()`](#waitforfinish).
 - The [`dequeue`](https://developer.mozilla.org/en-US/docs/Web/API/VideoDecoder/dequeue_event) event is not supported as it is not reliable across browsers.
 - In addition to [`EncodedVideoChunk`](https://developer.mozilla.org/en-US/docs/Web/API/EncodedVideoChunk), [`EncodedVideoChunkInit`](https://www.w3.org/TR/webcodecs/#dictdef-encodedvideochunkinit) objects are also accepted for [`.decode()`](#decode).
 - A [`webcodecsController()`](/docs/webcodecs/webcodecs-controller) instance can be passed in to the function, allowing for decoding to be paused, resumed and aborted.
@@ -86,7 +86,9 @@ Default value: `"info"`, which logs only important information.
 
 Returns an object with the following properties:
 
-### `waitForQueueToBeLessThan(queueSize: number)`
+### `waitForQueueToBeLessThan()`
+
+Pass a number to wait for the queue to be less than the given number.
 
 A promise that resolves when the queue size is less than the given number.  
 The queue is only decremented when the[ `onFrame`](#onframe) callback resolves.
@@ -103,6 +105,14 @@ Clears the queue and resets the decoder. Same as [`VideoDecoder.reset()`](https:
 
 Closes the decoder. Same as [`AudioDecoder.close()`](https://developer.mozilla.org/en-US/docs/Web/API/AudioDecoder/close).
 
+### `checkReset()`<AvailableFrom v="4.0.312" />
+
+Returns a handle with a `wasReset()` function. If the decoder was reset inbetween the call to `.checkReset()` and the call to `wasReset()`, `wasReset()` will return `true`. See [below](#checking-if-the-decoder-was-reset) for an example.
+
+### `getMostRecentSampleInput()`<AvailableFrom v="4.0.312" />
+
+Return the `.timestamp` of the most recently input sample. 
+
 ## Example usage with `@remotion/media-parser`
 
 ```tsx twoslash title="Decode a video track"
@@ -133,6 +143,50 @@ await parseMedia({
 });
 ```
 
+## Checking if the decoder was reset
+
+A potential race condition you may face is that `decoder.reset()` is called while a sample is waiting for the queue to be less than a certain number. Use `.checkReset()` to check if the decoder was reset after any asynchronous operation, and abort the processing of the sample if needed.
+
+```tsx twoslash title="Check if the decoder was reset"
+import {parseMedia} from '@remotion/media-parser';
+import {createVideoDecoder} from '@remotion/webcodecs';
+
+await parseMedia({
+  src: 'https://parser.media/video.mp4',
+  onVideoTrack: ({track, container}) => {
+    const decoder = createVideoDecoder({
+      track,
+      onFrame: console.log,
+      onError: console.error,
+    });
+
+    return async (sample) => {
+      const {wasReset} = decoder.checkReset();
+
+      await decoder.waitForQueueToBeLessThan(10);
+      if (wasReset()) {
+        return
+      }
+
+      await decoder.decode(sample);
+      if (wasReset()) {
+        return;
+      }
+
+      return async () => {
+        if (wasReset()) {
+          return;
+        }
+
+        // Called when the track is done
+        await decoder.flush();
+        decoder.close()
+      };
+    };
+  },
+});
+```
+
 ## See also
 
 - [Source code for this function](https://github.com/remotion-dev/remotion/blob/main/packages/webcodecs/src/create-video-decoder.ts)

packages/media-parser/src/webcodec-sample-types.ts

@@ -3,11 +3,19 @@ import type {MediaParserContainer} from './options';
 
 export type MediaParserOnAudioSample = (
 	sample: MediaParserAudioSample,
-) => void | Promise<void> | OnTrackDoneCallback | Promise<OnTrackDoneCallback>;
+) =>
+	| void
+	| Promise<OnTrackDoneCallback | void>
+	| Promise<void>
+	| OnTrackDoneCallback;
 
 export type MediaParserOnVideoSample = (
 	sample: MediaParserVideoSample,
-) => void | Promise<void> | OnTrackDoneCallback | Promise<OnTrackDoneCallback>;
+) =>
+	| void
+	| Promise<OnTrackDoneCallback | void>
+	| Promise<void>
+	| OnTrackDoneCallback;
 
 export type OnTrackDoneCallback = () => void | Promise<void>;
 

packages/webcodecs/src/create-audio-decoder.ts

@@ -1,4 +1,6 @@
 import type {MediaParserLogLevel} from '@remotion/media-parser';
+import type {FlushPending} from './flush-pending';
+import {makeFlushPending} from './flush-pending';
 import {getWaveAudioDecoder} from './get-wave-audio-decoder';
 import {makeIoSynchronizer} from './io-manager/io-synchronizer';
 import type {WebCodecsController} from './webcodecs-controller';
@@ -11,6 +13,10 @@ export type WebCodecsAudioDecoder = {
 	flush: () => Promise<void>;
 	waitForQueueToBeLessThan: (items: number) => Promise<void>;
 	reset: () => void;
+	checkReset: () => {
+		wasReset: () => boolean;
+	};
+	getMostRecentSampleInput: () => number | null;
 };
 
 export type CreateAudioDecoderInit = {
@@ -41,6 +47,8 @@ export const internalCreateAudioDecoder = ({
 		controller,
 	});
 
+	let mostRecentSampleReceived: number | null = null;
+
 	if (config.codec === 'pcm-s16') {
 		return getWaveAudioDecoder({
 			onFrame,
@@ -111,6 +119,8 @@ export const internalCreateAudioDecoder = ({
 			return;
 		}
 
+		mostRecentSampleReceived = audioSample.timestamp;
+
 		// Don't flush, it messes up the audio
 
 		const chunk =
@@ -129,22 +139,47 @@ export const internalCreateAudioDecoder = ({
 		}
 	};
 
+	let flushPending: FlushPending | null = null;
+	const lastReset: number | null = null;
+
 	return {
 		decode,
 		close,
-		flush: async () => {
-			// Firefox might throw "Needs to be configured first"
-			try {
-				await audioDecoder.flush();
-			} catch {}
+		flush: () => {
+			if (flushPending) {
+				throw new Error('Flush already pending');
+			}
 
-			await ioSynchronizer.waitForQueueSize(0);
+			const pendingFlush = makeFlushPending();
+			flushPending = pendingFlush;
+			Promise.resolve()
+				.then(() => {
+					return audioDecoder.flush();
+				})
+				.catch(() => {
+					// Firefox might throw "Needs to be configured first"
+				})
+				.finally(() => {
+					pendingFlush.resolve();
+					flushPending = null;
+				});
+
+			return pendingFlush.promise;
 		},
 		waitForQueueToBeLessThan: ioSynchronizer.waitForQueueSize,
 		reset: () => {
 			audioDecoder.reset();
 			audioDecoder.configure(config);
 		},
+		checkReset: () => {
+			const initTime = Date.now();
+			return {
+				wasReset: () => lastReset !== null && lastReset > initTime,
+			};
+		},
+		getMostRecentSampleInput() {
+			return mostRecentSampleReceived;
+		},
 	};
 };
 

packages/webcodecs/src/create-video-decoder.ts

@@ -1,4 +1,6 @@
 import type {MediaParserLogLevel} from '@remotion/media-parser';
+import type {FlushPending} from './flush-pending';
+import {makeFlushPending} from './flush-pending';
 import {makeIoSynchronizer} from './io-manager/io-synchronizer';
 import type {WebCodecsController} from './webcodecs-controller';
 
@@ -10,6 +12,10 @@ export type WebCodecsVideoDecoder = {
 	flush: () => Promise<void>;
 	waitForQueueToBeLessThan: (items: number) => Promise<void>;
 	reset: () => void;
+	checkReset: () => {
+		wasReset: () => boolean;
+	};
+	getMostRecentSampleInput: () => number | null;
 };
 
 export const internalCreateVideoDecoder = ({
@@ -25,12 +31,21 @@ export const internalCreateVideoDecoder = ({
 	config: VideoDecoderConfig;
 	logLevel: MediaParserLogLevel;
 }): WebCodecsVideoDecoder => {
+	if (
+		controller &&
+		controller._internals._mediaParserController._internals.signal.aborted
+	) {
+		throw new Error('Not creating audio decoder, already aborted');
+	}
+
 	const ioSynchronizer = makeIoSynchronizer({
 		logLevel,
 		label: 'Video decoder',
 		controller,
 	});
 
+	let mostRecentSampleReceived: number | null = null;
+
 	const videoDecoder = new VideoDecoder({
 		async output(frame) {
 			try {
@@ -88,6 +103,8 @@ export const internalCreateVideoDecoder = ({
 			return;
 		}
 
+		mostRecentSampleReceived = sample.timestamp;
+
 		const encodedChunk =
 			sample instanceof EncodedVideoChunk
 				? sample
@@ -96,22 +113,50 @@ export const internalCreateVideoDecoder = ({
 		ioSynchronizer.inputItem(sample.timestamp);
 	};
 
+	let flushPending: FlushPending | null = null;
+	let lastReset: number | null = null;
+
 	return {
 		decode,
 		close,
-		flush: async () => {
-			// Firefox might throw "Needs to be configured first"
-			try {
-				await videoDecoder.flush();
-			} catch {}
+		flush: () => {
+			if (flushPending) {
+				throw new Error('Flush already pending');
+			}
 
-			await ioSynchronizer.waitForQueueSize(0);
+			const pendingFlush = makeFlushPending();
+			flushPending = pendingFlush;
+			Promise.resolve()
+				.then(() => {
+					return videoDecoder.flush();
+				})
+				.catch(() => {
+					// Firefox might throw "Needs to be configured first"
+				})
+				.finally(() => {
+					pendingFlush.resolve();
+					flushPending = null;
+				});
+
+			return pendingFlush.promise;
 		},
 		waitForQueueToBeLessThan: ioSynchronizer.waitForQueueSize,
 		reset: () => {
+			lastReset = Date.now();
+			flushPending?.resolve();
+			ioSynchronizer.clearQueue();
 			videoDecoder.reset();
 			videoDecoder.configure(config);
 		},
+		checkReset: () => {
+			const initTime = Date.now();
+			return {
+				wasReset: () => lastReset !== null && lastReset > initTime,
+			};
+		},
+		getMostRecentSampleInput() {
+			return mostRecentSampleReceived;
+		},
 	};
 };