Adjust UrlSource behavior when reading non-range responses, it now honors the cache size and behaves like a ReadableStreamSource (fixes #457)

src/source.ts

@@ -712,7 +712,8 @@ export type UrlSourceOptions = {
 
 /**
  * A source backed by a URL. This is useful for reading data from the network. Requests will be made using an optimized
- * reading and prefetching pattern to minimize request count and latency.
+ * reading and prefetching pattern to minimize request count and latency. Works best with servers that support HTTP
+ * range requests; otherwise, resources must be streamed and read sequentially.
  * @group Input sources
  * @public
  */
@@ -737,6 +738,12 @@ export class UrlSource extends PathedSource {
 	 * @internal
 	 */
 	_fileSizeDetermined = false;
+	/**
+	 * When the server doesn't support range requests, we abandon the orchestrator and instead defer to an internal
+	 * ReadableStreamSource wrapping the response body, which pulls new data only when reads demand it.
+	 * @internal
+	 */
+	_sequentialBacking: ReadableStreamSource | null = null;
 
 	/**
 	 * Creates a new {@link UrlSource} backed by the resource at the specified URL.
@@ -847,7 +854,9 @@ export class UrlSource extends PathedSource {
 			return this._length !== null ? this._length : undefined;
 		}
 
-		const baseSize = this._orchestrator.fileSize;
+		const baseSize = this._sequentialBacking
+			? this._sequentialBacking._endIndex
+			: this._orchestrator.fileSize;
 		if (baseSize === null) {
 			return this._length !== null ? this._length : null;
 		}
@@ -867,12 +876,14 @@ export class UrlSource extends PathedSource {
 		}
 
 		const offset = this._offset;
-		const result = this._orchestrator.read(
-			offset + start,
-			offset + end,
-			Math.max(offset + minReadPosition, offset),
-			offset + Math.min(maxReadPosition, this._length ?? Infinity),
-		);
+		const result = this._sequentialBacking
+			? this._sequentialBacking._read(offset + start, offset + end)
+			: this._orchestrator.read(
+					offset + start,
+					offset + end,
+					Math.max(offset + minReadPosition, offset),
+					offset + Math.min(maxReadPosition, this._length ?? Infinity),
+				);
 
 		const processResult = (result: ReadResult | null) => {
 			if (!result) {
@@ -937,13 +948,29 @@ export class UrlSource extends PathedSource {
 					// Note: For range requests, this is _technically_ not correct, as the range response could contain
 					// less data than was requested. In practice, it seems most servers don't do this though, and the
 					// Content-Length header actually contains the length until the end of the file.
-					this._orchestrator.supplyFileSize(worker.currentPos + Number(contentLength));
+					// A non-206 response always spans the entire resource, no matter what range we asked for.
+					const basePos = response.status === 206 ? worker.currentPos : 0;
+					this._orchestrator.supplyFileSize(basePos + Number(contentLength));
 				}
 			}
 
 			this._fileSizeDetermined = true; // Yes, this is correct even if file size is still null
 
+			if (!response.body) {
+				throw new Error(
+					'Missing HTTP response body stream. The used fetch function must provide the response body as a'
+					+ ' ReadableStream.',
+				);
+			}
+
 			if (response.status !== 206) {
+				if (this._sequentialBacking) {
+					// Another worker already discovered the missing range request support and initiated the
+					// transition into sequential mode; this response is of no use anymore
+					void response.body.cancel();
+					return;
+				}
+
 				if (!this._usedForHls) {
 					const url = new URL(
 						this._url instanceof Request ? this._url.url : this._url,
@@ -958,34 +985,19 @@ export class UrlSource extends PathedSource {
 						if (!warnedOrigins.has(url.origin)) {
 							Logging._warn(
 								`HTTP server (origin ${url.origin}) did not respond to a range request with 206 Partial`
-								+ ' Content, meaning the entire resource will now be downloaded. To enable efficient'
-								+ ' media file streaming across a network, please make sure your server supports'
-								+ ' range requests.',
+								+ ' Content, meaning the resource will now be streamed sequentially, with old data'
+								+ ' being evicted from the cache. Reads into evicted regions will throw. To enable'
+								+ ' efficient media file streaming across a network, please make sure your server'
+								+ ' supports range requests. Alternatively, set maxCacheSize to Infinity in the'
+								+ ' UrlSource options to keep the entire resource in memory.',
 							);
 							warnedOrigins.add(url.origin);
 						}
 					}
 				}
 
-				worker.currentPos = 0;
-				this._orchestrator.options.maxCacheSize = Infinity; // 🤷
-
-				if (this._orchestrator.fileSize !== null) {
-					worker.targetPos = this._orchestrator.fileSize;
-				} else {
-					// The server is dumb, doesn't even surface the content length, but we'll work with it.
-					worker.targetPos = Infinity;
-					worker.strictTarget = false;
-				}
-
-				this._orchestrator.consolidateEverythingIntoOneWorker(worker);
-			}
-
-			if (!response.body) {
-				throw new Error(
-					'Missing HTTP response body stream. The used fetch function must provide the response body as a'
-					+ ' ReadableStream.',
-				);
+				this._transitionToSequentialMode(response.body);
+				return;
 			}
 
 			const reader = response.body.getReader();
@@ -1054,9 +1066,165 @@ export class UrlSource extends PathedSource {
 		// logic for that has vanished for now. Leaving a comment here if this becomes relevant again.
 	}
 
+	/** @internal */
+	private _transitionToSequentialMode(body: ReadableStream<Uint8Array>) {
+		// The server ignored our range request and is sending the entire resource from byte 0. Instead of downloading
+		// and caching the whole thing, we hand the response over to an internal ReadableStreamSource, which pulls new
+		// data only when reads demand it and evicts old data as usual. The response body is wrapped in a stream that
+		// transparently resumes when the connection dies.
+
+		let currentReader = body.getReader();
+		let streamPosition = 0;
+		let skipRemaining = 0;
+
+		const wrappedStream = new ReadableStream<Uint8Array>({
+			pull: async (controller) => {
+				while (true) {
+					let readResult: ReadableStreamReadResult<Uint8Array>;
+
+					try {
+						readResult = await currentReader.read();
+					} catch (error) {
+						if (this._disposed) {
+							throw error;
+						}
+
+						const retryDelayInSeconds = this._getRetryDelay(1, error, this._url);
+						if (retryDelayInSeconds === null) {
+							throw error;
+						}
+
+						Logging._error('Error while reading response stream. Attempting to resume.', error);
+						await wait(1000 * retryDelayInSeconds);
+
+						const newResponse = await retriedFetch(
+							this._options.fetchFn ?? fetch,
+							this._url,
+							mergeRequestInit(this._requestInit, {
+								headers: {
+									// Who knows, maybe the server honors range requests this time
+									Range: `bytes=${streamPosition}-`,
+								},
+							}),
+							this._getRetryDelay,
+							() => this._disposed,
+						);
+
+						if (!newResponse.ok) {
+							throw new Error(
+								// eslint-disable-next-line @typescript-eslint/no-base-to-string
+								`Error fetching ${String(this._url)}:`
+								+ ` ${newResponse.status} ${newResponse.statusText}`,
+							);
+						}
+
+						if (!newResponse.body) {
+							throw new Error(
+								'Missing HTTP response body stream. The used fetch function must provide the'
+								+ ' response body as a ReadableStream.',
+							);
+						}
+
+						currentReader = newResponse.body.getReader();
+						// If the server still doesn't do ranges, the new response starts at byte 0 again and
+						// we need to skip over everything we already delivered. Cursed!
+						skipRemaining = newResponse.status === 206 ? 0 : streamPosition;
+
+						continue;
+					}
+
+					if (readResult.done) {
+						controller.close();
+						return;
+					}
+
+					let chunk = readResult.value;
+
+					if (skipRemaining > 0) {
+						const skippedAmount = Math.min(skipRemaining, chunk.length);
+						skipRemaining -= skippedAmount;
+						chunk = chunk.subarray(skippedAmount);
+					}
+
+					if (chunk.length === 0) {
+						continue;
+					}
+
+					streamPosition += chunk.length;
+					controller.enqueue(chunk);
+
+					return;
+				}
+			},
+			cancel: () => currentReader.cancel(),
+		});
+
+		const backing = new ReadableStreamSource(wrappedStream, {
+			maxCacheSize: this._orchestrator.options.maxCacheSize,
+		});
+		backing._endIndex = this._orchestrator.fileSize; // Might still be null
+		backing._cacheMissErrorMessage = 'Attempted to read data from an already-evicted part of the cache. Because the'
+			+ ' HTTP server did not honor the range request, data can only be read sequentially, with old data being'
+			+ ' evicted from the cache. To fix this issue, either ensure your server responds to range requests with'
+			+ ' 206 Partial Content, or set maxCacheSize to Infinity in the UrlSource options. Note that the latter'
+			+ ' will store the entire file in the cache if needed, no matter how large.';
+		backing.on('read', ({ start, end }) => this._dispatchRead(start, end));
+
+		this._sequentialBacking = backing;
+
+		// Everything still pending in the orchestrator must now be served by the backing instead. Gather all
+		// pending slices, then retire the orchestrator's workers and queued reads for good; _read will only
+		// consult the backing from now on.
+		const uniqueSlices = new Set<PendingSlice>();
+
+		for (const otherWorker of this._orchestrator.workers) {
+			for (const slice of otherWorker.pendingSlices) {
+				uniqueSlices.add(slice);
+			}
+
+			otherWorker.aborted = true;
+			otherWorker.pendingSlices.length = 0;
+		}
+
+		for (const queuedRead of this._orchestrator.queuedReads) {
+			for (const slice of queuedRead.pendingSlices) {
+				uniqueSlices.add(slice);
+			}
+		}
+
+		this._orchestrator.workers.length = 0;
+		this._orchestrator.queuedReads.length = 0;
+
+		for (const slice of uniqueSlices) {
+			const result = backing._read(slice.start, slice.start + slice.bytes.length);
+
+			if (result instanceof Promise) {
+				result.then((readResult) => {
+					if (readResult) {
+						// The backing's cache is empty at this point, so the read is guaranteed to produce
+						// exactly the requested range
+						assert(readResult.offset === slice.start);
+						slice.resolve(readResult.bytes);
+					} else {
+						slice.resolve(null);
+					}
+				}, (error: unknown) => slice.reject(error));
+			} else {
+				// Can only happen synchronously when the slice lies beyond the known file size
+				assert(result === null);
+				slice.resolve(null);
+			}
+		}
+	}
+
 	/** @internal */
 	_dispose() {
 		this._orchestrator.dispose();
+
+		if (this._sequentialBacking) {
+			this._sequentialBacking._disposed = true;
+			this._sequentialBacking._dispose();
+		}
 	}
 }
 
@@ -1456,6 +1624,12 @@ export class ReadableStreamSource extends Source {
 	_endIndex: number | null = null;
 	/** @internal */
 	_pulling = false;
+	/**
+	 * Overridable for internal use.
+	 * @internal
+	 */
+	_cacheMissErrorMessage = 'Attempted to read data from an already-evicted part of the cache. With'
+		+ ' ReadableStreamSource, you must access the data more sequentially or increase the size of its cache.';
 
 	/** Creates a new {@link ReadableStreamSource} backed by the specified `ReadableStream<Uint8Array>`. */
 	constructor(stream: ReadableStream<Uint8Array>, options: ReadableStreamSourceOptions = {}) {
@@ -1581,10 +1755,7 @@ export class ReadableStreamSource extends Source {
 
 	/** @internal */
 	_throwDueToCacheMiss() {
-		throw new Error(
-			'Read is before the cached region. With ReadableStreamSource, you must access the data more'
-			+ ' sequentially or increase the size of its cache.',
-		);
+		throw new Error(this._cacheMissErrorMessage);
 	}
 
 	/** @internal */
@@ -1644,8 +1815,8 @@ export class ReadableStreamSource extends Source {
 			});
 
 			// Do cache eviction, based on the distance from the last-requested index. It's important that we do it like
-			// this and not based on where the reader is at, because if the reader is fast, we'll unnecessarily evict
-			// data that we still might need.
+			// this and not based on how far we've pulled the stream, because if the stream supplies data faster than it
+			// is being requested, we'd unnecessarily evict data that we still might need.
 			while (this._cache.length > 0) {
 				const firstEntry = this._cache[0]!;
 				const distance = this._maxRequestedIndex - firstEntry.end;
@@ -2167,40 +2338,6 @@ class ReadOrchestrator {
 			});
 	}
 
-	consolidateEverythingIntoOneWorker(worker: ReadWorker) {
-		// Here we merge everything into one "megaworker" that spans the entire file. We assume the passed-in worker
-		// is already configured to be a megaworker.
-
-		const uniqueSlices = new Set(worker.pendingSlices);
-
-		for (let i = 0; i < this.workers.length; i++) {
-			const otherWorker = this.workers[i]!;
-			if (otherWorker === worker) {
-				continue;
-			}
-
-			for (const slice of otherWorker.pendingSlices) {
-				uniqueSlices.add(slice);
-			}
-
-			otherWorker.aborted = true;
-			otherWorker.pendingSlices.length = 0;
-			this.workers.splice(i, 1);
-			i--;
-		}
-
-		for (let i = 0; i < this.queuedReads.length; i++) {
-			const queuedRead = this.queuedReads[i]!;
-
-			for (const slice of queuedRead.pendingSlices) {
-				uniqueSlices.add(slice);
-			}
-		}
-
-		worker.pendingSlices = [...uniqueSlices];
-		this.queuedReads.length = 0;
-	}
-
 	/** Called by a worker when it has read some data. */
 	supplyWorkerData(worker: ReadWorker, bytes: Uint8Array) {
 		assert(!worker.aborted);

test/node/url-source.test.ts

@@ -0,0 +1,309 @@
+import { expect, test } from 'vitest';
+import http from 'node:http';
+import fs from 'node:fs';
+import path from 'node:path';
+import {
+	ALL_FORMATS,
+	EncodedPacket,
+	EncodedPacketSink,
+	FilePathSource,
+	Input,
+	Logging,
+	LogLevel,
+	UrlSource,
+} from '../../src/index.js';
+
+const __dirname = new URL('.', import.meta.url).pathname;
+const videoFilePath = path.join(__dirname, '..', 'public/video.mp4');
+
+test('UrlSource works against a server without range request support', async () => {
+	const server = await startRangelessServer();
+	const logs = captureLogs();
+
+	try {
+		using input = new Input({
+			source: new UrlSource(server.url),
+			formats: ALL_FORMATS,
+		});
+
+		const track = await input.getPrimaryVideoTrack();
+		if (!track) throw new Error('No video track found');
+
+		const sink = new EncodedPacketSink(track);
+
+		const timestamps: number[] = [];
+		for await (const packet of sink.packets()) {
+			timestamps.push(packet.timestamp);
+		}
+
+		expect(timestamps).toHaveLength(125);
+
+		// The default cache size exceeds the file size, so random access back to the start of the file still works
+		const firstPacket = await sink.getFirstPacket();
+		if (!firstPacket) throw new Error('No first packet found');
+
+		expect(firstPacket.timestamp).toBe(0);
+		expect(firstPacket.data.byteLength).toBeGreaterThan(0);
+
+		expect(logs.warnings.filter(
+			x => x.includes('did not respond to a range request with 206 Partial Content'),
+		)).toHaveLength(1);
+	} finally {
+		logs.stop();
+		server.close();
+	}
+});
+
+test('UrlSource throws when reading from an evicted region in sequential mode', async () => {
+	const server = await startRangelessServer();
+	const logs = captureLogs();
+
+	try {
+		using input = new Input({
+			// Much smaller than the file, so the start of the file will get evicted during full iteration
+			source: new UrlSource(server.url, { maxCacheSize: 2 ** 20 /* 1 MiB */ }),
+			formats: ALL_FORMATS,
+		});
+
+		const track = await input.getPrimaryVideoTrack();
+		if (!track) throw new Error('No video track found');
+
+		const sink = new EncodedPacketSink(track);
+
+		const timestamps: number[] = [];
+		for await (const packet of sink.packets()) {
+			timestamps.push(packet.timestamp);
+		}
+
+		expect(timestamps).toHaveLength(125);
+
+		await expect(sink.getFirstPacket()).rejects.toThrow(/already-evicted part of the cache/);
+
+		expect(logs.warnings.filter(
+			x => x.includes('did not respond to a range request with 206 Partial Content'),
+		)).toHaveLength(1);
+	} finally {
+		logs.stop();
+		server.close();
+	}
+});
+
+test('UrlSource resumes with correct data when the connection dies in sequential mode', async () => {
+	// The server kills the connection partway through the first two responses. Since it doesn't support range
+	// requests, each resume response starts back at byte 0 and the already-delivered prefix must be skipped over
+	// exactly; any off-by-one would corrupt the packet data.
+	const server = await startRangelessServer({ responseByteLimits: [700_000, 1_500_000] });
+	const logs = captureLogs();
+
+	try {
+		using referenceInput = new Input({
+			source: new FilePathSource(videoFilePath),
+			formats: ALL_FORMATS,
+		});
+
+		const referenceTrack = await referenceInput.getPrimaryVideoTrack();
+		if (!referenceTrack) throw new Error('No video track found');
+
+		const referencePackets: EncodedPacket[] = [];
+		for await (const packet of new EncodedPacketSink(referenceTrack).packets()) {
+			referencePackets.push(packet);
+		}
+
+		using input = new Input({
+			source: new UrlSource(server.url, { getRetryDelay: () => 0 }),
+			formats: ALL_FORMATS,
+		});
+
+		const track = await input.getPrimaryVideoTrack();
+		if (!track) throw new Error('No video track found');
+
+		let packetIndex = 0;
+		for await (const packet of new EncodedPacketSink(track).packets()) {
+			const referencePacket = referencePackets[packetIndex]!;
+
+			expect(packet.timestamp).toBe(referencePacket.timestamp);
+			expect(Buffer.from(packet.data).equals(Buffer.from(referencePacket.data))).toBe(true);
+
+			packetIndex++;
+		}
+
+		expect(packetIndex).toBe(referencePackets.length);
+
+		// Initial request plus one resume per killed response
+		expect(server.requestCount()).toBe(3);
+		expect(logs.errors.filter(x => x.includes('Attempting to resume'))).toHaveLength(2);
+	} finally {
+		logs.stop();
+		server.close();
+	}
+});
+
+test('UrlSource with maxCacheSize: Infinity allows random access against a rangeless server', async () => {
+	const server = await startRangelessServer();
+	const logs = captureLogs();
+
+	try {
+		using input = new Input({
+			// This is the escape hatch recommended by the eviction error message: with an infinite cache, nothing
+			// is ever evicted and random access keeps working
+			source: new UrlSource(server.url, { maxCacheSize: Infinity }),
+			formats: ALL_FORMATS,
+		});
+
+		const track = await input.getPrimaryVideoTrack();
+		if (!track) throw new Error('No video track found');
+
+		const sink = new EncodedPacketSink(track);
+
+		const timestamps: number[] = [];
+		for await (const packet of sink.packets()) {
+			timestamps.push(packet.timestamp);
+		}
+
+		expect(timestamps).toHaveLength(125);
+
+		const firstPacket = await sink.getFirstPacket();
+		if (!firstPacket) throw new Error('No first packet found');
+
+		expect(firstPacket.timestamp).toBe(0);
+		expect(firstPacket.data.byteLength).toBeGreaterThan(0);
+
+		expect(logs.warnings.filter(
+			x => x.includes('did not respond to a range request with 206 Partial Content'),
+		)).toHaveLength(1);
+	} finally {
+		logs.stop();
+		server.close();
+	}
+});
+
+test('UrlSource in sequential mode downloads lazily and aborts the response on dispose', async () => {
+	const fileSize = fs.statSync(videoFilePath).size;
+	const server = await startRangelessServer();
+	const logs = captureLogs();
+
+	const input = new Input({
+		source: new UrlSource(server.url),
+		formats: ALL_FORMATS,
+	});
+
+	try {
+		const track = await input.getPrimaryVideoTrack();
+		if (!track) throw new Error('No video track found');
+
+		// If the client were downloading eagerly, the entire file would easily arrive during this window. Instead,
+		// the server is expected to stall, since data is only pulled down when reads demand it.
+		await new Promise(resolve => setTimeout(resolve, 3000));
+
+		expect(server.bytesSent()).toBeLessThan(fileSize / 2);
+
+		// The response is merely suspended, not dead: reading still works
+		const sink = new EncodedPacketSink(track);
+		const firstPacket = await sink.getFirstPacket();
+		if (!firstPacket) throw new Error('No first packet found');
+
+		expect(firstPacket.timestamp).toBe(0);
+		expect(firstPacket.data.byteLength).toBeGreaterThan(0);
+
+		input.dispose();
+
+		// Give the abort a moment to propagate to the server
+		await new Promise(resolve => setTimeout(resolve, 250));
+
+		// Disposal terminated the response early, long before the entire file was sent
+		expect(server.abortedResponses()).toBe(1);
+		expect(server.bytesSent()).toBeLessThan(fileSize / 2);
+	} finally {
+		input.dispose();
+		logs.stop();
+		server.close();
+	}
+}, 10_000);
+
+/**
+ * Spins up a server that ignores Range headers, always responding with 200 and the full file. If responseByteLimits
+ * is provided, the n-th response gets its connection killed after roughly that many bytes.
+ */
+const startRangelessServer = async (options: { responseByteLimits?: number[] } = {}) => {
+	const fileSize = fs.statSync(videoFilePath).size;
+	let requestCount = 0;
+	let bytesSent = 0;
+	let abortedResponses = 0;
+
+	const server = http.createServer((req, res) => {
+		const byteLimit = options.responseByteLimits?.[requestCount] ?? Infinity;
+		requestCount++;
+
+		res.on('error', () => {}); // The client may abort the connection at any time
+		res.on('close', () => {
+			if (!res.writableFinished) {
+				abortedResponses++;
+			}
+		});
+		res.writeHead(200, {
+			'Content-Type': 'video/mp4',
+			'Content-Length': fileSize,
+		});
+
+		// Stream the file with backpressure and a small chunk size, so that the amount of sent bytes closely
+		// tracks how much the client actually consumes
+		const stream = fs.createReadStream(videoFilePath, { highWaterMark: 2 ** 14 });
+		let bytesSentThisResponse = 0;
+
+		stream.on('data', (chunk) => {
+			const canContinue = res.write(chunk);
+			bytesSent += chunk.length;
+			bytesSentThisResponse += chunk.length;
+
+			if (bytesSentThisResponse >= byteLimit) {
+				stream.destroy();
+				res.destroy(); // Kill the connection mid-response
+
+				return;
+			}
+
+			if (!canContinue) {
+				stream.pause();
+				res.once('drain', () => stream.resume());
+			}
+		});
+		stream.on('end', () => res.end());
+	});
+
+	await new Promise<void>(resolve => server.listen(0, resolve));
+
+	const address = server.address();
+	if (!address || typeof address === 'string') throw new Error('Unexpected server address');
+
+	return {
+		url: `http://localhost:${address.port}/video.mp4`,
+		requestCount: () => requestCount,
+		bytesSent: () => bytesSent,
+		abortedResponses: () => abortedResponses,
+		close: () => {
+			server.closeAllConnections();
+			server.close();
+		},
+	};
+};
+
+/** Collects logged warnings and errors, keeping the console clean while doing so. */
+const captureLogs = () => {
+	const warnings: string[] = [];
+	const errors: string[] = [];
+	const unsubscribeWarn = Logging.on('warn', args => warnings.push(args.map(String).join(' ')));
+	const unsubscribeError = Logging.on('error', args => errors.push(args.map(String).join(' ')));
+
+	const previousLogLevel = Logging.level;
+	Logging.level = LogLevel.Silent;
+
+	return {
+		warnings,
+		errors,
+		stop: () => {
+			unsubscribeWarn();
+			unsubscribeError();
+			Logging.level = previousLogLevel;
+		},
+	};
+};