Change pump reuse rule: enforce queued commands are resolved in the order in which they are queued

src/cursors.ts

@@ -374,11 +374,11 @@ export abstract class SampleCursor<
 	pumpStopQueued = false;
 	pumpStopped = promiseWithResolvers();
 
-	decodedTimestamps: number[] = [];
-	maxDecodedSequenceNumber = -1;
 	pumpTarget: EncodedPacket | null = null;
+	lastTarget: EncodedPacket | null = null;
 	pumpMutex = new AsyncMutex4();
 	_closed = false;
+	queuedResets = 0;
 
 	error: unknown = null;
 	errorSet = false;
@@ -447,10 +447,6 @@ export abstract class SampleCursor<
 				}
 			}
 
-			while (this.decodedTimestamps.length > 0 && this.decodedTimestamps[0]! <= sample.timestamp) {
-				this.decodedTimestamps.shift();
-			}
-
 			if (this.pendingRequests.length === 0) {
 				if (this.pumpStopQueued) {
 					sample.close();
@@ -525,6 +521,14 @@ export abstract class SampleCursor<
 		}
 	}
 
+	_ensureWillBeOpen() {
+		if (this.queuedResets > 0) {
+			return;
+		}
+
+		this._ensureNotClosed();
+	}
+
 	async _seekToPacket(
 		res: ResultValue<TransformedSample | null>,
 		targetPacketPromise: MaybePromise<EncodedPacket | null>,
@@ -565,9 +569,7 @@ export abstract class SampleCursor<
 
 		let setNewPump = true;
 
-		if (this.sampleQueue.length > 0 && targetPacket.timestamp < this.sampleQueue[0]!.timestamp) {
+		if (this.lastTarget && this.lastTarget.timestamp <= targetPacket.timestamp) {
-
-		} else {
 			while (this.sampleQueue.length > 0) {
 				const nextSample = this.sampleQueue.shift()!;
 				this.queueDequeue.resolve();
@@ -581,31 +583,17 @@ export abstract class SampleCursor<
 				}
 			}
 
-			if (this.pumpTarget) {
+			if (targetPacket.timestamp - this.lastTarget.timestamp < 0.1) {
-				const max = Math.max(this.pumpTarget.sequenceNumber, this.maxDecodedSequenceNumber);
+				setNewPump = false;
+			} else {
+				let key = this.packetReader.readNextKey(this.lastTarget, { verifyKeyPackets: true });
+				if (key instanceof Promise) key = await key;
 
-				if (targetPacket.sequenceNumber <= max) {
+				if (
-					const nextExpectedTimestamp = this.decodedTimestamps[0] ?? this.packetCursor.current?.timestamp;
+					!key
-					if (nextExpectedTimestamp === undefined || nextExpectedTimestamp > targetPacket.timestamp) {
+					|| targetPacket.sequenceNumber < key.sequenceNumber
-						// yeah
+				) {
-						// formulate the "no nextExpectedTimestamp" case
+					setNewPump = false;
-					} else {
-						setNewPump = false;
-					}
-				} else {
-					if (targetPacket.timestamp - this.pumpTarget.timestamp < 0.1) {
-						setNewPump = false;
-					} else {
-						let key = this.packetReader.readNextKey(this.pumpTarget, { verifyKeyPackets: true });
-						if (key instanceof Promise) key = await key;
-
-						if (
-							!key
-							|| targetPacket.sequenceNumber < key.sequenceNumber
-						) {
-							setNewPump = false;
-						}
-					}
 				}
 			}
 		}
@@ -614,6 +602,8 @@ export abstract class SampleCursor<
 			await this.stopPump();
 		}
 
+		this.lastTarget = targetPacket;
+
 		if (!this.pumpTarget || targetPacket.sequenceNumber > this.pumpTarget.sequenceNumber) {
 			this.pumpTarget = targetPacket;
 		}
@@ -648,7 +638,7 @@ export abstract class SampleCursor<
 	}
 
 	seekToFirst(): MaybePromise<TransformedSample | null> {
-		this._ensureNotClosed();
+		this._ensureWillBeOpen();
 
 		try {
 			const result = new ResultValue<TransformedSample | null>();
@@ -667,7 +657,7 @@ export abstract class SampleCursor<
 	}
 
 	seekTo(timestamp: number): MaybePromise<TransformedSample | null> {
-		this._ensureNotClosed();
+		this._ensureWillBeOpen();
 
 		try {
 			const result = new ResultValue<TransformedSample | null>();
@@ -686,7 +676,7 @@ export abstract class SampleCursor<
 	}
 
 	seekToKey(timestamp: number): MaybePromise<TransformedSample | null> {
-		this._ensureNotClosed();
+		this._ensureWillBeOpen();
 
 		try {
 			const result = new ResultValue<TransformedSample | null>();
@@ -751,21 +741,13 @@ export abstract class SampleCursor<
 		}
 		this.lastPendingRequest = pendingRequest;
 
-		// Note that the next packet we get here is not necessarily the packet belonging to the next sample, since we
-		// can have out of order timestamps when B-frames are at play. However, if next() is called a sufficiently
-		// large amount of times, then the pump target will stay roughly in sync with the desired next sample.
-		const next = this.pumpTarget && await this.packetReader.readNext(this.pumpTarget, { metadataOnly: true });
-		if (next) {
-			this.pumpTarget = next;
-		}
-
 		lock.release(); // Waiting for the return would be too long
 
 		return res.set(await request.promise);
 	}
 
 	next(): MaybePromise<TransformedSample | null> {
-		this._ensureNotClosed();
+		this._ensureWillBeOpen();
 
 		try {
 			const result = new ResultValue<TransformedSample | null>();
@@ -786,7 +768,7 @@ export abstract class SampleCursor<
 	async iterate(
 		callback: (sample: TransformedSample, stop: () => void) => MaybePromise<unknown>,
 	) {
-		this._ensureNotClosed();
+		this._ensureWillBeOpen();
 
 		let stopped = false;
 		const stop = () => stopped = true;
@@ -794,6 +776,12 @@ export abstract class SampleCursor<
 		const waitPromise = this.waitUntilIdle();
 		if (waitPromise) await waitPromise;
 
+		const lock = this.pumpMutex.lock();
+		if (lock.pending) await lock.ready;
+		lock.release();
+
+		this._ensureNotClosed();
+
 		while (true) {
 			if (this.current) {
 				const result = callback(this.current, stop);
@@ -815,11 +803,17 @@ export abstract class SampleCursor<
 
 	// eslint-disable-next-line @stylistic/generator-star-spacing
 	async *[Symbol.asyncIterator]() {
-		this._ensureNotClosed();
+		this._ensureWillBeOpen();
 
 		const waitPromise = this.waitUntilIdle();
 		if (waitPromise) await waitPromise;
 
+		const lock = this.pumpMutex.lock();
+		if (lock.pending) await lock.ready;
+		lock.release();
+
+		this._ensureNotClosed();
+
 		while (true) {
 			if (this.current) {
 				yield this.current;
@@ -932,8 +926,6 @@ export abstract class SampleCursor<
 					continue;
 				}
 
-				insertSorted(this.decodedTimestamps, this.packetCursor.current.timestamp, x => x);
-				this.maxDecodedSequenceNumber = this.packetCursor.current.sequenceNumber;
 				this.decoder.decode(this.packetCursor.current);
 
 				if (this.debugInfo.enabled) {
@@ -974,10 +966,9 @@ export abstract class SampleCursor<
 			this.pumpStopped.resolve();
 			this.pumpStopped = promiseWithResolvers();
 			this.pumpRunning = false;
-			this.maxDecodedSequenceNumber = -1;
-			this.decodedTimestamps.length = 0;
 			this.pumpTarget = null;
 			this.pumpStopQueued = false;
+			this.lastTarget = null;
 		}
 	}
 
@@ -1009,6 +1000,9 @@ export abstract class SampleCursor<
 	async reset() {
 		this.predictedRequests++;
 
+		this.queuedResets++;
+		using _ = defer(() => this.queuedResets--);
+
 		using lock = this.pumpMutex.lock();
 		if (lock.pending) await lock.ready;
 
@@ -1023,8 +1017,6 @@ export abstract class SampleCursor<
 		assert(this.sampleQueue.length === 0);
 		assert(this.pendingRequests.length === 0);
 		assert(this.lastPendingRequest === null);
-		assert(this.decodedTimestamps.length === 0);
-		assert(this.maxDecodedSequenceNumber === -1);
 		assert(this.pumpTarget === null);
 		assert(!this.decoder || this.decoder.closed);
 

test/browser/sample-cursor.test.ts

@@ -10,7 +10,36 @@ import {
 	VideoSampleCursor,
 } from '../../src/cursors.js';
 import { AudioSample, VideoSample } from '../../src/sample.js';
-import { promiseIterateAll } from '../../src/misc.js';
+import { promiseIterateAll, promiseWithResolvers } from '../../src/misc.js';
+
+const promiseAllEnsureOrder = async <T>(promises: T[]) => {
+	const results: Awaited<T>[] = [];
+	const { promise, resolve, reject } = promiseWithResolvers();
+
+	const onValue = (value: Awaited<T>, i: number) => {
+		if (results.length === i) {
+			results.push(value);
+
+			if (results.length === promises.length) {
+				resolve();
+			}
+		} else {
+			reject(new Error('Order violation'));
+		}
+	};
+
+	for (let i = 0; i < promises.length; i++) {
+		const value = promises[i]!;
+		if (value instanceof Promise) {
+			void value.then(x => onValue(x as Awaited<T>, i));
+		} else {
+			onValue(value as Awaited<T>, i);
+		}
+	}
+
+	await promise;
+	return results;
+};
 
 test('Sample cursor seeking', async () => {
 	using input = new Input({
@@ -392,8 +421,7 @@ test('Sample cursor reset', async () => {
 		cursor.next(),
 	];
 
-	// eslint-disable-next-line @typescript-eslint/await-thenable
+	const results = await promiseAllEnsureOrder(commands);
-	const results = await Promise.all(commands);
 
 	expect(cursor.closed).toBe(false);
 	expect(results[0]!.timestamp).toBeGreaterThan(firstSample!.timestamp);
@@ -410,8 +438,7 @@ test('Sample cursor reset', async () => {
 		cursor2.reset(),
 	];
 
-	// eslint-disable-next-line @typescript-eslint/await-thenable
+	await promiseAllEnsureOrder(commands2);
-	await Promise.all(commands2);
 
 	expect(cursor2.debugInfo.decodedPackets).toHaveLength(1);
 });
@@ -588,8 +615,7 @@ test('Command queuing', async () => {
 	];
 	expect(commands0.every(x => x instanceof Promise)).toBe(true);
 
-	// eslint-disable-next-line @typescript-eslint/await-thenable
+	await promiseAllEnsureOrder(commands0);
-	await Promise.all(commands0);
 
 	expect(cursor0.debugInfo.pumpsStarted).toBe(1);
 
@@ -602,8 +628,7 @@ test('Command queuing', async () => {
 	];
 	expect(commands1.every(x => x instanceof Promise)).toBe(true);
 
-	// eslint-disable-next-line @typescript-eslint/await-thenable
+	const results1 = await promiseAllEnsureOrder(commands1);
-	const results1 = await Promise.all(commands1);
 	expect(results1[0]!.timestamp).toBe(0);
 	expect(cursor1.debugInfo.decodedPackets.map(x => x.timestamp)).toEqual([0]);
 
@@ -621,8 +646,7 @@ test('Command queuing', async () => {
 	];
 	expect(commands2.every(x => x instanceof Promise)).toBe(true);
 
-	// eslint-disable-next-line @typescript-eslint/await-thenable
+	const results2 = await promiseAllEnsureOrder(commands2);
-	const results2 = await Promise.all(commands2);
 	expect(results2[0]!.timestamp).toBe(0);
 	expect(results2[1]!.timestamp).toBe(1);
 	expect(results2[2]!.timestamp).toBe(2);
@@ -632,6 +656,7 @@ test('Command queuing', async () => {
 	expect(cursor2.debugInfo.decodedPackets.map(x => x.timestamp)).toEqual([
 		0, 1, 2, 3, 4, 5,
 	]);
+	expect(cursor2.debugInfo.pumpsStarted).toBe(6);
 
 	const cursor3 = new VideoSampleCursor(reader);
 	cursor3.debugInfo.enabled = true;
@@ -647,8 +672,7 @@ test('Command queuing', async () => {
 	];
 	expect(commands3.every(x => x instanceof Promise)).toBe(true);
 
-	// eslint-disable-next-line @typescript-eslint/await-thenable
+	const results3 = await promiseAllEnsureOrder(commands3);
-	const results3 = await Promise.all(commands3);
 
 	expect(results3[0]!.timestamp).toBeLessThanOrEqual(0.5);
 	expect(results3[1]!.timestamp).toBeLessThanOrEqual(0.4);
@@ -657,7 +681,7 @@ test('Command queuing', async () => {
 	expect(results3[4]!.timestamp).toBeLessThanOrEqual(0.1);
 	expect(results3[5]!.timestamp).toBe(0);
 	expect(cursor3.debugInfo.decodedPackets.every(x => x.timestamp <= 0.5)).toBe(true);
-	expect(cursor3.debugInfo.pumpsStarted).toBe(1);
+	expect(cursor3.debugInfo.pumpsStarted).toBe(6);
 
 	const cursor4 = new VideoSampleCursor(reader);
 	cursor4.debugInfo.enabled = true;
@@ -671,8 +695,7 @@ test('Command queuing', async () => {
 
 	expect(commands4.every(x => x instanceof Promise)).toBe(true);
 
-	// eslint-disable-next-line @typescript-eslint/await-thenable
+	const results4 = await promiseAllEnsureOrder(commands4);
-	const results4 = await Promise.all(commands4);
 
 	expect(results4[0]!.timestamp).toBe(0);
 	expect(results4[1]!.timestamp).toBeGreaterThan(results4[0]!.timestamp);
@@ -697,8 +720,7 @@ test('Command queuing', async () => {
 		cursor5.close(),
 	];
 
-	// eslint-disable-next-line @typescript-eslint/await-thenable
+	const results5 = await promiseAllEnsureOrder(commands5);
-	const results5 = await Promise.all(commands5);
 
 	expect(results5[0]!.timestamp).toBe(0);
 	expect(results5[1]!.timestamp).toBeGreaterThan(results5[0]!.timestamp);
@@ -714,7 +736,7 @@ test('Command queuing', async () => {
 	expect(results5[9]!.timestamp).toBe(0);
 	expect(results5[10]!.timestamp).toBeGreaterThan(results5[9]!.timestamp);
 
-	expect(cursor5.debugInfo.pumpsStarted).toBe(1);
+	expect(cursor5.debugInfo.pumpsStarted).toBe(2);
 
 	const cursor6 = new VideoSampleCursor(reader);
 	cursor6.debugInfo.enabled = true;
@@ -729,8 +751,7 @@ test('Command queuing', async () => {
 		cursor6.close(),
 	];
 
-	// eslint-disable-next-line @typescript-eslint/await-thenable
+	const results6 = await promiseAllEnsureOrder(commands6);
-	const results6 = await Promise.all(commands6);
 
 	expect(results6[0]!.timestamp).toBe(0);
 	expect(results6[1]!.timestamp).toBeLessThanOrEqual(0.4);
@@ -748,7 +769,7 @@ test('Command queuing', async () => {
 	];
 	expect(commands7[0]).toBe(commands7[1]); // Same Promise
 
-	await Promise.all(commands7);
+	await promiseAllEnsureOrder(commands7);
 
 	const cursor8 = new VideoSampleCursor(reader, { autoClose: false });
 
@@ -881,8 +902,7 @@ test('AudioSampleCursor', async () => {
 		cursor.seekTo(0.2),
 	];
 
-	// eslint-disable-next-line @typescript-eslint/await-thenable
+	const result = await promiseAllEnsureOrder(commands);
-	const result = await Promise.all(commands);
 
 	expect(result[0]!.timestamp).toBeLessThanOrEqual(0);
 	expect(result[1]!.timestamp).toBeLessThanOrEqual(0.05);