From 3023c5f7bd7c129918ff367680589e56b4c04e8b Mon Sep 17 00:00:00 2001 From: Vanilagy <1696106+Vanilagy@users.noreply.github.com> Date: Thu, 6 Aug 2026 14:02:28 +0200 Subject: [PATCH] Adjust UrlSource behavior when reading non-range responses, it now honors the cache size and behaves like a ReadableStreamSource (fixes #457) --- src/source.ts | 279 +++++++++++++++++++++++-------- test/node/url-source.test.ts | 309 +++++++++++++++++++++++++++++++++++ 2 files changed, 517 insertions(+), 71 deletions(-) create mode 100644 test/node/url-source.test.ts diff --git a/src/source.ts b/src/source.ts index cfa7cdb..1cd4a45 100644 --- a/src/source.ts +++ b/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) { + // 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({ + pull: async (controller) => { + while (true) { + let readResult: ReadableStreamReadResult; + + 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(); + + 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`. */ constructor(stream: ReadableStream, 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); diff --git a/test/node/url-source.test.ts b/test/node/url-source.test.ts new file mode 100644 index 0000000..fce8dd0 --- /dev/null +++ b/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(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; + }, + }; +};