MORE CODE REVIEW

changes-and-fixes.txt

@@ -112,6 +112,31 @@ BEHAVIOR CHANGES
    Previously would have matched the aac prefix check. MP3 check now
    runs first and includes this codec string.
 
+8. UrlSource: servers without Content-Length now supported
+   Previously, if the server returned 200 without a Content-Length header,
+   UrlSource threw an error. Now it gracefully handles this by downloading
+   the entire resource with an unbounded worker (targetPos = Infinity,
+   strictTarget = false), determining file size once the stream ends.
+
+9. UrlSource: file size probing request removed
+   The old UrlSource always made a dedicated initial `Range: bytes=0-`
+   request solely to probe file size and range request support. File size
+   is now determined lazily from the response headers of the first actual
+   read, removing the extra round-trip.
+
+10. UrlSource: non-range-request server warnings deduplicated per origin
+    The "server did not respond with 206 Partial Content" warning is now
+    emitted at most once per origin instead of on every request.
+
+11. ReadableStreamSource: reader now canceled on dispose
+    ReadableStreamSource._dispose() now calls `this._reader?.cancel()`,
+    properly releasing the underlying stream resource.
+
+12. ReadOrchestrator: worker queue system
+    When the maximum worker count is reached, reads are now queued and
+    dispatched when a worker becomes free, instead of evicting running
+    workers which could abort in-flight fetches.
+
 
 ============================================================
 DEPRECATIONS

src/aes.ts

@@ -288,6 +288,10 @@ export const createAes128CbcDecryptStream = (
 			} else {
 				// This is the last chunk
 				const paddingLength = output[bytesToRead - 1]!;
+				if (paddingLength === 0 || paddingLength > 16) {
+					throw new Error('Invalid PKCS#7 padding. Incorrect key or corrupted data.');
+				}
+
 				const trimmedOutput = output.subarray(0, bytesToRead - paddingLength); // PKCS#7 padding
 
 				controller.enqueue(trimmedOutput);

src/hls/hls-segmented-input.ts

@@ -27,6 +27,7 @@ import {
 	TAG_PROGRAM_DATE_TIME,
 	TAG_TARGETDURATION,
 } from './hls-misc';
+import { HlsInputFormat } from '../input-format';
 
 const IV_STRING_REGEX = /^0[xX][0-9a-fA-F]+$/;
 
@@ -603,7 +604,8 @@ export class HlsSegmentedInput extends SegmentedInput {
 					return ref!;
 				},
 			),
-			formats: this.input._formats,
+			// Do not allow recursive HLS. Cool on paper, but allows for nasty infinite-depth request trees.
+			formats: this.input._formats.filter(x => !(x instanceof HlsInputFormat)),
 			initInput: initInput ?? undefined,
 		});
 

src/input.ts

@@ -64,8 +64,10 @@ let inputFinalizationRegistry: FinalizationRegistry<SourceRef[]> | null = null;
 if (typeof FinalizationRegistry !== 'undefined') {
 	inputFinalizationRegistry = new FinalizationRegistry((refs) => {
 		for (const ref of refs) {
+			if (!ref.freed) {
 				ref.free();
 			}
+		}
 	});
 }
 

src/misc.ts

@@ -1219,8 +1219,14 @@ export class EventEmitter<TEvents extends Record<string, unknown>> {
 		if (!listeners) {
 			return;
 		}
+
 		for (const entry of listeners) {
+			try {
 				(entry.fn as (data: unknown) => void)(data);
+			} catch (error) {
+				console.error(error);
+			}
+
 			if (entry.once) {
 				listeners.delete(entry);
 			}

src/reader.ts

@@ -388,91 +388,3 @@ export const readAllLines = (slice: FileSlice, length: number, options?: {
 
 	return lines;
 };
-
-export class LineReader {
-	getReader: () => MaybePromise<Reader>;
-	ignore?: (line: string) => boolean;
-	reader: Reader | null = null;
-	textDecoder = new TextDecoder();
-	currentLineNumber = 0; // 1-based
-	readPos = 0;
-	reachedEnd = false;
-	lineBuffer = '';
-
-	constructor(getReader: () => MaybePromise<Reader>, ignore?: (line: string) => boolean) {
-		this.getReader = getReader;
-		this.ignore = ignore;
-	}
-
-	readNextLine(): MaybePromise<string | null> {
-		if (this.reachedEnd) {
-			return null;
-		}
-
-		const line = this.extractLineFromBuffer();
-		if (line !== null) {
-			return line;
-		}
-
-		return (async () => {
-			if (!this.reader) {
-				let reader = this.getReader();
-				if (reader instanceof Promise) reader = await reader;
-
-				this.reader = reader;
-			}
-
-			while (true) {
-				let slice = this.reader.requestSliceRange(this.readPos, 0, 1024);
-				if (slice instanceof Promise) slice = await slice;
-
-				if (!slice || slice.length === 0) {
-					this.reachedEnd = true;
-					const line = this.lineBuffer.trim();
-					this.lineBuffer = '';
-
-					if (line) {
-						this.currentLineNumber++;
-					}
-
-					if (!line || this.ignore?.(line)) {
-						return null;
-					}
-
-					return line;
-				}
-
-				const bytes = readBytes(slice, slice.length);
-				this.readPos += bytes.length;
-
-				this.lineBuffer += this.textDecoder.decode(bytes, { stream: true });
-
-				const line = this.extractLineFromBuffer();
-				if (line !== null) {
-					return line;
-				}
-			}
-		})();
-	}
-
-	extractLineFromBuffer() {
-		assert(!this.reachedEnd);
-
-		while (true) {
-			const newlineIndex = this.lineBuffer.indexOf('\n');
-			if (newlineIndex === -1) {
-				return null;
-			}
-
-			const line = this.lineBuffer.slice(0, newlineIndex).trim();
-			this.lineBuffer = this.lineBuffer.slice(newlineIndex + 1);
-			this.currentLineNumber++;
-
-			if (this.ignore?.(line)) {
-				continue;
-			}
-
-			return line;
-		}
-	}
-}

src/source.ts

@@ -233,8 +233,10 @@ export class SourceRef<S extends Source = Source> implements Disposable {
 	 * Calls {@link SourceRef.free}.
 	 */
 	[Symbol.dispose]() {
+		if (!this.freed) {
 			this.free();
 		}
+	}
 }
 
 /**

test/node/hls-input.test.ts

@@ -1,7 +1,7 @@
 /* eslint-disable @stylistic/max-len */
 import { ALL_FORMATS, BufferSource, createInputFrom, EncodedPacketSink, Input, InputAudioTrack, InputVideoTrack, PathedSource } from 'mediabunny';
 import { expect, test } from 'vitest';
-import { HLS, HlsInputFormat } from '../../src/input-format.js';
+import { HLS, HLS_FORMATS, HlsInputFormat } from '../../src/input-format.js';
 import { assert, rejectAfter } from '../../src/misc.js';
 
 // A lot of test cases taken from:
@@ -775,3 +775,29 @@ test.concurrent('Missing media tag codec', async () => {
 
 	expect([...new Set(await Promise.all(tracks.map(x => x.getCodec())))]).toEqual(['aac', 'avc']);
 });
+
+test.concurrent('Circular/recursive HLS is forbidden', async () => {
+	const text = `#EXTM3U
+#EXT-X-VERSION=3
+#EXT-X-TARGETDURATION=10
+
+#EXTINF:5,
+root.m3u8
+
+#EXT-X-ENDLIST
+`;
+
+	const input = new Input({
+		source: new PathedSource(
+			'root.m3u8',
+			({ path }) => {
+				console.log(1, path);
+				assert(path === 'root.m3u8');
+				return new BufferSource(new TextEncoder().encode(text));
+			},
+		),
+		formats: HLS_FORMATS,
+	});
+
+	await expect(input.getTracks()).rejects.toThrow('unsupported');
+});