From 4f88d71277346b1b7b22da04a26ad3c5f3c342d5 Mon Sep 17 00:00:00 2001 From: Vanilagy <1696106+Vanilagy@users.noreply.github.com> Date: Tue, 8 Sep 2026 17:42:23 +0200 Subject: [PATCH] Add handleUnhandledError to most sources (closes #489) --- docs/guide/reading-media-files.md | 17 ++++++++++- src/source.ts | 50 +++++++++++++++++++++++++++++-- 2 files changed, 64 insertions(+), 3 deletions(-) diff --git a/docs/guide/reading-media-files.md b/docs/guide/reading-media-files.md index b07f1ab..c847cdc 100644 --- a/docs/guide/reading-media-files.md +++ b/docs/guide/reading-media-files.md @@ -607,6 +607,9 @@ type BlobSourceOptions = { // The maximum number of bytes the cache is allowed to hold // in memory. Defaults to 8 MiB. maxCacheSize?: number; + + // Handles errors that occur while no read is pending + handleUnhandledError?: (error: unknown) => unknown; }; ``` @@ -640,6 +643,9 @@ type UrlSourceOptions = { // Used to provide a custom fetch function fetchFn?: typeof fetch; + + // Handles errors that occur while no read is pending + handleUnhandledError?: (error: unknown) => unknown; }; ``` @@ -704,6 +710,9 @@ type FilePathSourceOptions = { // The maximum number of bytes the cache is allowed to hold // in memory. Defaults to 8 MiB. maxCacheSize?: number; + + // Handles errors that occur while no read is pending + handleUnhandledError?: (error: unknown) => unknown; }; ``` @@ -743,6 +752,7 @@ type CustomSourceOptions = { dispose?: () => unknown; maxCacheSize?: number; prefetchProfile?: 'none' | 'fileSystem' | 'network'; + handleUnhandledError?: (error: unknown) => unknown; }; type MaybePromise = T | Promise; @@ -761,6 +771,8 @@ type MaybePromise = T | Promise; - `'none'` (default): No prefetching; only the data needed in the moment is requested. - `'fileSystem'`: File system-optimized prefetching: a small amount of data is prefetched bidirectionally, aligned with page boundaries. - `'network'`: Network-optimized prefetching, or more generally, prefetching optimized for any high-latency environment: tries to minimize the amount of read calls and aggressively prefetches data when sequential access patterns are detected. +- `handleUnhandledError`\ + Handles errors that occur while no read is pending. By default, these become unhandled promise rejections. ::: info `CustomSource` was previously known as `StreamSource` and is still available under that alias, but usage of `StreamSource` is deprecated. @@ -790,6 +802,9 @@ type ReadableStreamSourceOptions = { // The maximum number of bytes the cache is allowed to hold // in memory. Defaults to 16 MiB. maxCacheSize?: number; + + // Handles errors that occur while no read is pending + handleUnhandledError?: (error: unknown) => unknown; }; ``` @@ -896,4 +911,4 @@ const input = new Input({ formats: ALL_FORMATS, initInput, }); -``` \ No newline at end of file +``` diff --git a/src/source.ts b/src/source.ts index 3edea96..ee4e7d6 100644 --- a/src/source.ts +++ b/src/source.ts @@ -494,6 +494,9 @@ export type BlobSourceOptions = { * field to `false` to try a slower but more stable reading method. */ useStreamReader?: boolean; + + /** Handles errors that occur while no read is pending. By default, these become unhandled rejections. */ + handleUnhandledError?: (error: unknown) => unknown; }; const blobReaderRegistry = typeof FinalizationRegistry !== 'undefined' @@ -541,6 +544,9 @@ export class BlobSource extends Source { if (options.useStreamReader !== undefined && typeof options.useStreamReader !== 'boolean') { throw new TypeError('options.useStreamReader, when provided, must be a boolean.'); } + if (options.handleUnhandledError !== undefined && typeof options.handleUnhandledError !== 'function') { + throw new TypeError('options.handleUnhandledError, when provided, must be a function.'); + } super(); @@ -563,6 +569,7 @@ export class BlobSource extends Source { } }, prefetchProfile: PREFETCH_PROFILES.fileSystem, + handleUnhandledError: options.handleUnhandledError, }); this._orchestrator.fileSize = blob.size; @@ -732,6 +739,9 @@ export type UrlSourceOptions = { * features, or use a custom implementation. */ fetchFn?: typeof fetch; + + /** Handles errors that occur while no read is pending. By default, these become unhandled rejections. */ + handleUnhandledError?: (error: unknown) => unknown; }; /** @@ -808,6 +818,9 @@ export class UrlSource extends PathedSource { throw new TypeError('options.fetchFn, when provided, must be a function.'); // Won't bother validating this function beyond this } + if (options.handleUnhandledError !== undefined && typeof options.handleUnhandledError !== 'function') { + throw new TypeError('options.handleUnhandledError, when provided, must be a function.'); + } const urlString = url instanceof Request ? url.url @@ -869,6 +882,7 @@ export class UrlSource extends PathedSource { maxWorkerCount: options.parallelism ?? DEFAULT_PARALLELISM, runWorker: this._runWorker.bind(this), prefetchProfile: PREFETCH_PROFILES.network, + handleUnhandledError: options.handleUnhandledError, }); } @@ -1191,6 +1205,7 @@ export class UrlSource extends PathedSource { const backing = new ReadableStreamSource(wrappedStream, { maxCacheSize: this._orchestrator.options.maxCacheSize, + handleUnhandledError: this._options.handleUnhandledError, }); 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' @@ -1286,6 +1301,9 @@ const parseByteRangeHeader = (value: string) => { export type FilePathSourceOptions = { /** The maximum number of bytes the cache is allowed to hold in memory. Defaults to 8 MiB. */ maxCacheSize?: number; + + /** Handles errors that occur while no read is pending. By default, these become unhandled rejections. */ + handleUnhandledError?: (error: unknown) => unknown; }; /** @@ -1349,6 +1367,7 @@ export class FilePathSource extends PathedSource { }, maxCacheSize: options.maxCacheSize, prefetchProfile: 'fileSystem', + handleUnhandledError: options.handleUnhandledError, }); } @@ -1419,6 +1438,9 @@ export type CustomSourceOptions = { * patterns are detected. */ prefetchProfile?: 'none' | 'fileSystem' | 'network'; + + /** Handles errors that occur while no read is pending. By default, these become unhandled rejections. */ + handleUnhandledError?: (error: unknown) => unknown; }; /** @@ -1447,6 +1469,9 @@ export class CustomSource extends Source { if (options.dispose !== undefined && typeof options.dispose !== 'function') { throw new TypeError('options.dispose, when provided, must be a function.'); } + if (options.handleUnhandledError !== undefined && typeof options.handleUnhandledError !== 'function') { + throw new TypeError('options.handleUnhandledError, when provided, must be a function.'); + } if ( options.maxCacheSize !== undefined && (!isNumber(options.maxCacheSize) || options.maxCacheSize < 0) @@ -1468,6 +1493,7 @@ export class CustomSource extends Source { maxWorkerCount: 2, // Fixed for now, *should* be fine prefetchProfile: PREFETCH_PROFILES[options.prefetchProfile ?? 'none'], runWorker: this._runWorker.bind(this), + handleUnhandledError: options.handleUnhandledError, }); } @@ -1617,6 +1643,9 @@ type ReadableStreamSourcePendingSlice = { export type ReadableStreamSourceOptions = { /** The maximum number of bytes the cache is allowed to hold in memory. Defaults to 32 MiB. */ maxCacheSize?: number; + + /** Handles errors that occur while no read is pending. By default, these become unhandled rejections. */ + handleUnhandledError?: (error: unknown) => unknown; }; /** @@ -1654,6 +1683,8 @@ export class ReadableStreamSource extends Source { _endIndex: number | null = null; /** @internal */ _pulling = false; + /** @internal */ + _handleUnhandledError: ((error: unknown) => void) | undefined; /** * Overridable for internal use. * @internal @@ -1669,6 +1700,9 @@ export class ReadableStreamSource extends Source { if (!options || typeof options !== 'object') { throw new TypeError('options must be an object.'); } + if (options.handleUnhandledError !== undefined && typeof options.handleUnhandledError !== 'function') { + throw new TypeError('options.handleUnhandledError, when provided, must be a function.'); + } if ( options.maxCacheSize !== undefined && (!isNumber(options.maxCacheSize) || options.maxCacheSize < 0) @@ -1680,6 +1714,7 @@ export class ReadableStreamSource extends Source { this._stream = stream; this._maxCacheSize = options.maxCacheSize ?? (32 * 2 ** 20 /* 32 MiB */); + this._handleUnhandledError = options.handleUnhandledError; } /** @internal */ @@ -1774,6 +1809,8 @@ export class ReadableStreamSource extends Source { if (this._pendingSlices.length > 0) { this._pendingSlices.forEach(x => x.reject(error)); // Make sure to propagate any errors this._pendingSlices.length = 0; + } else if (this._handleUnhandledError) { + this._handleUnhandledError(error); } else { throw error; // So it doesn't get swallowed } @@ -2000,6 +2037,7 @@ class ReadOrchestrator { prefetchProfile: PrefetchProfile; maxWorkerCount: number; onIdleWorkerRemoved?: (worker: ReadWorker) => void; + handleUnhandledError?: (error: unknown) => unknown; }) {} read( @@ -2214,7 +2252,11 @@ class ReadOrchestrator { } // Nobody's awaiting this result but an errored read is still notable - throw error; + if (this.options.handleUnhandledError) { + this.options.handleUnhandledError(error); + } else { + throw error; + } }); } @@ -2332,7 +2374,11 @@ class ReadOrchestrator { worker.pendingSlices.forEach(x => x.reject(error)); // Make sure to propagate any errors worker.pendingSlices.length = 0; } else if (!worker.aborted && !this.disposed) { - throw error; // So it doesn't get swallowed + if (this.options.handleUnhandledError) { + this.options.handleUnhandledError(error); + } else { + throw error; // So it doesn't get swallowed + } } }) .finally(() => {