diff --git a/docs/developer-guide/concepts/javascript-apis.md b/docs/developer-guide/concepts/javascript-apis.md index 2229c114ba..3af75674ff 100644 --- a/docs/developer-guide/concepts/javascript-apis.md +++ b/docs/developer-guide/concepts/javascript-apis.md @@ -37,6 +37,30 @@ The preferred way to provide random-access data to loaders.gl is through `Readab `ReadableFile` classes replace the deprecated `FileProvider` utilities; new code should use the `ReadableFile` wrappers exported from `@loaders.gl/loader-utils` (and `DataViewReadableFile` from `@loaders.gl/zip`) to keep loader interactions consistent across platforms. +### Validated HTTP ranges + +`HttpFile.open()` pins the remote object's byte length and available `ETag`/`Last-Modified` +validators. Supplying identity from a trusted manifest avoids the opening one-byte probe: + +```ts +import {HttpFile} from '@loaders.gl/loader-utils'; + +const file = await HttpFile.open('https://example.com/data.parquet', { + byteLength: manifest.byteLength, + etag: manifest.etag, + consistency: 'strict' +}); + +const bytes = await file.read(offset, length, abortController.signal); +console.log(file.getIdentitySnapshot(), file.getTelemetry()); +``` + +Every read requires an exact `206` response and validates `Content-Range`, response length, and the +pinned object identity before returning bytes. `strict` consistency requires validators to remain +visible; the default `best-effort` mode still rejects changed validators but permits servers whose +CORS policy does not expose them. A shared `RangeRequestScheduler` can coalesce nearby reads while +keeping different authentication and validator contexts isolated. + ## Saving data Saving data from a browser is either done by POST requests to a server, or via local downloads. diff --git a/docs/docs-sidebar.json b/docs/docs-sidebar.json index 5ec151804c..dc786043af 100644 --- a/docs/docs-sidebar.json +++ b/docs/docs-sidebar.json @@ -251,6 +251,7 @@ "modules/loader-utils/README", "modules/loader-utils/api-reference/data-source-manager", "modules/loader-utils/api-reference/readable-file", + "modules/loader-utils/api-reference/http-file", "modules/loader-utils/api-reference/request-scheduler", "modules/loader-utils/api-reference/range-request-scheduler", "modules/loader-utils/api-reference/parse-with-context" diff --git a/docs/modules/loader-utils/README.md b/docs/modules/loader-utils/README.md index 9e008035f2..bf016197a3 100644 --- a/docs/modules/loader-utils/README.md +++ b/docs/modules/loader-utils/README.md @@ -1,3 +1,10 @@ # Overview The `@loaders.gl/loader-utils` contains utilities for creating loaders. + +## API reference + +- [`ReadableFile`](/docs/modules/loader-utils/api-reference/readable-file) provides the common random-access file contract. +- [`HttpFile`](/docs/modules/loader-utils/api-reference/http-file) validates random-access HTTP reads and remote object identity. +- [`RequestScheduler`](/docs/modules/loader-utils/api-reference/request-scheduler) limits asynchronous request concurrency. +- [`RangeRequestScheduler`](/docs/modules/loader-utils/api-reference/range-request-scheduler) coalesces compatible byte ranges. diff --git a/docs/modules/loader-utils/api-reference/http-file.md b/docs/modules/loader-utils/api-reference/http-file.md new file mode 100644 index 0000000000..1947eda0a5 --- /dev/null +++ b/docs/modules/loader-utils/api-reference/http-file.md @@ -0,0 +1,163 @@ +# HttpFile + +

+ From-v5.0 + experimental +

+ +`HttpFile` provides validated random access to a remote object in browsers and Node.js. It sends +exact HTTP range requests, pins the object's length and available validators, and rejects responses +that no longer describe the same object. + +```typescript +import {HttpFile} from '@loaders.gl/loader-utils'; + +const file = await HttpFile.open('https://example.com/data.parquet', { + consistency: 'strict' +}); + +const footer = await file.read(file.size - 8, 8); +console.log(new Uint8Array(footer), file.getTelemetry()); +await file.close(); +``` + +The server must support byte ranges. A normal request must return `206 Partial Content` with an +exact `Content-Range`; `HttpFile` deliberately rejects a `200 OK` full-object fallback. A valid +zero-byte object may respond to the opening probe with `416 Range Not Satisfiable` and +`Content-Range: bytes */0`. + +## Opening a file + +### `HttpFile.open(url, options?, signal?): Promise` + +Creates a file and immediately pins its identity. Unless a complete identity is supplied, opening +sends `Range: bytes=0-0` to discover the object length, `ETag`, and `Last-Modified` value. + +Supplying a known length together with either validator avoids that opening request: + +```typescript +const file = await HttpFile.open(url, { + byteLength: manifest.byteLength, + etag: manifest.etag, + consistency: 'strict' +}); +``` + +### `new HttpFile(url, options?)` + +Creates a lazy file. The first call to `open()`, `stat()`, or `read()` discovers and pins the +identity. Concurrent callers share one discovery request, while each caller can independently +cancel its own wait. + +Prefer the static `HttpFile.open()` form when code needs to use `size` synchronously. + +## Options + +| Option | Type | Default | Description | +| ---- | ---- | ------- | ----------- | +| `fetch` | `(url, init?) => Promise` | `globalThis.fetch` | Custom fetch implementation. | +| `fetchOptions` | `RequestInit` | none | Headers, credentials, and other options copied to every request. `HttpFile` supplies the `GET` method, `Range` header, and combined signal. | +| `byteLength` | `number` | discovered | Trusted non-negative object length. | +| `etag` | `string` | discovered | Trusted object ETag. | +| `lastModified` | `string` | discovered | Trusted Last-Modified value. | +| `consistency` | `'best-effort' \| 'strict'` | `'best-effort'` | Controls how missing validators are handled. Changed visible validators are always rejected. | +| `rangeScheduler` | `RangeRequestScheduler` | private scheduler | Shared scheduler used to coalesce compatible reads. Each `HttpFile` remains an isolated request context. | +| `rangeSchedulerProps` | `RangeRequestSchedulerProps` | `{batchDelayMs: 0}` | Configuration for the private scheduler. Ignored when `rangeScheduler` is supplied. | + +`fetchOptions.headers` are preserved except for `Range`, which is set for each read. A signal in +`fetchOptions` applies to every request; a signal passed to `open()` or `read()` applies only to that +operation. + +## Consistency modes + +### `best-effort` + +Rejects changed validators whenever the server exposes them. It permits a response that omits a +previously visible validator, which is useful when a server or CORS policy does not expose headers +consistently. If an ETag disappears but both responses expose `Last-Modified`, that fallback must +still match. + +### `strict` + +Requires every response to expose the pinned validator. If no validator was supplied, the opening +response must expose either `ETag` or `Last-Modified`. + +For cross-origin URLs, expose `Content-Range`, `ETag`, and `Last-Modified` through the server's CORS +configuration when strict validation is required. + +## Properties + +### `size: number` + +Pinned object length. A lazy file reports a supplied `byteLength`, or zero until identity discovery +has completed. + +### `bigsize: bigint` + +The same object length represented as a bigint. + +### `url: string` + +Remote object URL. `handle` contains the same value for the `ReadableFile` interface. + +## Methods + +### `open(signal?): Promise` + +Pins the identity of a lazily constructed file. Repeated calls reuse the cached identity. + +### `read(offset?, length?, signal?): Promise` + +Reads exactly `length` bytes starting at `offset`. Both values must be non-negative safe integers, +and the range must not extend past `size`. A zero-length read returns an empty buffer without an +HTTP request after identity is known. + +Each non-empty response is accepted only when all of the following are true: + +- the status is `206`; +- `Content-Range` exactly matches the requested offsets and pinned object length; +- visible validators are consistent with the pinned identity; and +- the consumed response body contains exactly the requested number of bytes. + +### `stat(): Promise` + +Returns the pinned `size`, `bigsize`, and `isDirectory: false`. It discovers identity first when the +file is lazy. + +### `getIdentitySnapshot(): HttpFileIdentity | null` + +Returns the frozen pinned `{byteLength, etag, lastModified}` object, or `null` before a lazy file has +opened. + +### `getTelemetry(): HttpFileTelemetry` + +Returns a frozen point-in-time snapshot: + +```typescript +type HttpFileTelemetry = { + requestedBytes: number; + downloadedBytes: number; + requestCount: number; + networkTimeMs: number; + abortCount: number; + errorCount: number; +}; +``` + +The counters belong to this `HttpFile`, even when its scheduler is shared with other files. + +### `fetchRange(offset, length, signal?): Promise` + +Compatibility method for `ReadableFile` consumers that expect a `Response`. New code should prefer +`read()`. + +### `close(): Promise` + +Prevents new operations. Active requests remain controlled by their per-operation or persistent +abort signals. + +## Sharing a range scheduler + +Several files may share a [`RangeRequestScheduler`](./range-request-scheduler) to centralize queue +configuration and stats. `HttpFile` assigns every instance a private isolation key, so requests +with different credentials, validators, or fetch implementations are never coalesced together. diff --git a/docs/modules/loader-utils/api-reference/range-request-scheduler.md b/docs/modules/loader-utils/api-reference/range-request-scheduler.md index 0c94461ba4..26a1cff493 100644 --- a/docs/modules/loader-utils/api-reference/range-request-scheduler.md +++ b/docs/modules/loader-utils/api-reference/range-request-scheduler.md @@ -20,10 +20,14 @@ const scheduler = new RangeRequestScheduler({ stats }); +// Reuse one key only for requests with the same URL, credentials, and fetch behavior. +const transportContext = {}; + const arrayBuffer = await scheduler.fetch({ url, offset: 1_000_000, length: 4096, + isolationKey: transportContext, fetchOptions: { headers: {Authorization: 'Bearer token'} } @@ -56,6 +60,12 @@ creates the `Range` header, preserves caller headers from `fetchOptions`, aborts `200 OK` full-object responses, handles `416` size probes for offset `0`, and records transport diagnostics in `stats`. +HTTP fetch calls are isolated by default because separate calls may use different credentials, +headers, or fetch implementations. To coalesce compatible calls, pass the same stable +`isolationKey` object to each call. Keys are compared by identity (`===`), so creating a new object +for every request does not enable coalescing. Never reuse a key across different authentication or +validator contexts. + ### `scheduleRequest(request): Promise` Enqueues one exact range using a caller-supplied transport callback. The returned promise @@ -63,6 +73,15 @@ resolves to the exact requested byte slice, not the merged transport response. `request.fetchRange` must return the bytes for the offset and length it receives. Those may be larger than the original request when several child requests are merged. +If a server legitimately clamps the final range at end of file, return a transport result with +`arrayBuffer` and the authoritative `sourceByteLength`. The scheduler accepts a short response only +when its end offset exactly matches that declared length; unmarked and mismatched short responses +are rejected. + +`scheduleRequest()` coalesces requests with the same `sourceId` by default. Pass distinct +`isolationKey` values when one source identifier can refer to different transport, credential, or +validator contexts. Conversely, pass the same stable key to state explicitly that those contexts +are compatible. Use `scheduleRequest` for non-HTTP transports or sources that need custom response handling. @@ -94,6 +113,7 @@ type RangeStats = { requestedBytes: number; transportBytes: number; responseBytes: number; + networkTimeMs: number; overfetchBytes: number; failedTransportRanges: number; abortedLogicalRanges: number; diff --git a/docs/modules/loader-utils/api-reference/readable-file.md b/docs/modules/loader-utils/api-reference/readable-file.md index 7363e480d6..2d9fa6718b 100644 --- a/docs/modules/loader-utils/api-reference/readable-file.md +++ b/docs/modules/loader-utils/api-reference/readable-file.md @@ -4,12 +4,12 @@ ## Available classes -- `HttpFile` (browser & Node.js) – wraps a URL and downloads byte ranges with HTTP range requests when supported. +- [`HttpFile`](./http-file) (browser & Node.js) – validates HTTP byte-range reads and pins remote object identity. - `BlobFile` (browser & Node.js) – provides random access reads on `Blob` or `File` instances via the standard slicing APIs. - `NodeFile` (Node.js) – exposes random access reads backed by the local file system without importing `fs` directly in application code. - `DataViewReadableFile` (browser & Node.js) – adapts an in-memory `ArrayBuffer`/`DataView` into the `ReadableFile` interface for archive parsing or other buffer-first workflows. -All implementations satisfy the `ReadableFile` interface exported from `@loaders.gl/loader-utils` and support `slice`/`read` helpers for incremental processing of large files. +All implementations satisfy the `ReadableFile` interface exported from `@loaders.gl/loader-utils` and support exact `read` operations for incremental processing of large files. :::info Legacy `FileProvider` classes have been removed from the default `@loaders.gl/loader-utils` exports. Use the `ReadableFile` implementations above instead. @@ -22,8 +22,8 @@ Legacy `FileProvider` classes have been removed from the default `@loaders.gl/lo ```typescript import {HttpFile} from '@loaders.gl/loader-utils'; -const file = new HttpFile('https://example.com/archive.3tz'); -const header = await file.slice(0, 1024).arrayBuffer(); +const file = await HttpFile.open('https://example.com/archive.3tz'); +const header = await file.read(0, 1024); ``` ### Reading browser `File` drops @@ -33,7 +33,7 @@ import {BlobFile} from '@loaders.gl/loader-utils'; async function inspectUpload(fileInput: File) { const blobFile = new BlobFile(fileInput); - const signature = await blobFile.slice(0, 8).arrayBuffer(); + const signature = await blobFile.read(0, 8); return new Uint8Array(signature); } ``` @@ -44,7 +44,9 @@ async function inspectUpload(fileInput: File) { import {NodeFile} from '@loaders.gl/loader-utils'; const nodeFile = new NodeFile('/data/tileset.slpk'); -const footerBytes = await nodeFile.slice(-4096).arrayBuffer(); +const {size} = await nodeFile.stat(); +const footerLength = Math.min(size, 4096); +const footerBytes = await nodeFile.read(size - footerLength, footerLength); ``` ### Adapting an `ArrayBuffer` @@ -54,6 +56,7 @@ import {DataViewReadableFile} from '@loaders.gl/zip'; const archiveBuffer = await fetch(url).then((response) => response.arrayBuffer()); const archiveFile = new DataViewReadableFile(new DataView(archiveBuffer)); +const header = await archiveFile.read(0, 8); ``` These adapters can be passed anywhere a loader expects a `ReadableFile`, ensuring consistent random access across browser and Node.js environments. diff --git a/docs/modules/parquet/api-reference/parquet-source-loader.md b/docs/modules/parquet/api-reference/parquet-source-loader.md index 516154dca3..798b675559 100644 --- a/docs/modules/parquet/api-reference/parquet-source-loader.md +++ b/docs/modules/parquet/api-reference/parquet-source-loader.md @@ -143,11 +143,19 @@ await Array.fromAsync(source.read()); console.log(source.getTelemetry()); ``` -The snapshot reports exact transport counts and bytes, range-cache hits, cumulative +The frozen snapshot reports exact transport counts and bytes, range-cache hits, cumulative network/decode/Arrow durations, candidate/pruned/decoded row groups, emitted batches and rows, retries, cancellations, and failures. `retryCount` remains zero while the source uses its fail-fast range policy. +### `capabilities: ParquetSourceCapabilities` + +The source exposes the frozen `PARQUET_SOURCE_CAPABILITIES` descriptor synchronously, before any +network or decoding work starts. It reports support for cached immutable metadata, row-group and +column selection, provenance, cancellation, custom range transport, object-version validation, +statistics, transport/decode telemetry, and package-local WASM delivery. Source worker decoding is +the remaining deferred capability. + ### `close(): Promise` Aborts active requests, closes the range-backed file, and permanently closes the source. Calling @@ -177,6 +185,14 @@ individual read. | `rangeRequests.stats` | `Stats` | scheduler default | probe.gl range-request counters. | | `rangeRequests.onEvent` | `(event) => void` | `undefined` | Range scheduling diagnostic callback. | +## Package-local WASM + +`@loaders.gl/parquet/wasm` exports `PARQUET_WASM_URL`, a bundler-resolvable URL for the packaged +`parquet_wasm_bg.wasm` asset. The raw file is also exported as +`@loaders.gl/parquet/parquet_wasm_bg.wasm` for explicit copy or self-hosting workflows. The current +`ParquetSource` uses the TypeScript range decoder and does not initialize WASM; these entry points +serve the package's WASM loader and writer paths. + ## Current limitations - Decoding runs on the caller thread; worker-backed decoding and transferable Arrow buffers are not diff --git a/docs/whats-new.mdx b/docs/whats-new.mdx index ee8f3cc3a6..beb0c98325 100644 --- a/docs/whats-new.mdx +++ b/docs/whats-new.mdx @@ -88,8 +88,11 @@ Release Date: 2026 - [`ParquetSourceLoader`](/docs/modules/parquet/api-reference/parquet-source-loader) NEW - adds reusable range-backed Parquet metadata and selective row-group/column reads as cancellable Arrow batches with source provenance and object-version validation. Its lightweight root export dynamically preloads the runtime implementation. - `ParquetSourceLoader` exposes normalized column-chunk statistics, predicate-based row-group pruning, and cumulative transport/decode/Arrow telemetry with exact request and byte counts. +- `ParquetSourceLoader` now publishes an immutable capability descriptor, deep-freezes cached schema metadata and batch provenance, and preserves caller abort reasons. - `ParquetSourceLoader` now materializes selected columns directly into Arrow batches without an intermediate object-row table. - The wasm-backed `ParquetLoader` can decode in a cancellable worker, transfers Arrow output through Arrow IPC, and resolves its packaged worker and WASM assets without an implicit CDN dependency. +- The Parquet WASM backend is updated to 0.7.2, and `@loaders.gl/parquet/wasm` exposes its package-local binary URL for explicit asset workflows. +- `HttpFile` now supports pinned object identity, strict or best-effort validator consistency, cancellable exact range reads, shared scheduling, and immutable request/byte/time telemetry. - The Parquet TypeScript backend now reads Data Page V2, `DELTA_BINARY_PACKED`, `DELTA_LENGTH_BYTE_ARRAY`, `DELTA_BYTE_ARRAY`, and legacy Hadoop-framed LZ4 data. - [`ParquetJSLoader`](/docs/modules/parquet/api-reference/parquet-js-loader) and [`ParquetJSWriter`](/docs/modules/parquet/api-reference/parquet-js-writer) NEW - add the experimental parquetjs plain-row and plain-table APIs. - `parquet.shape` on [`ParquetLoader`](/docs/modules/parquet/api-reference/parquet-loader) is now documented for selecting object-row or Arrow output from the canonical wasm-backed loader. diff --git a/modules/core/test/lib/api/create-raster-source.spec.ts b/modules/core/test/lib/api/create-raster-source.spec.ts index d6f5916848..8fdfd35651 100644 --- a/modules/core/test/lib/api/create-raster-source.spec.ts +++ b/modules/core/test/lib/api/create-raster-source.spec.ts @@ -228,6 +228,43 @@ test('GeoTIFFRasterSource uses RangeRequestScheduler for remote byte-range reads expect(raster.data).toBeInstanceOf(Float32Array); }); +test('GeoTIFFRasterSource isolates clients that share a range scheduler', async () => { + const file = await readFixtureBytes(TIFF_URL); + const rangeScheduler = new RangeRequestScheduler({batchDelayMs: 0}); + const firstRanges: string[] = []; + const secondRanges: string[] = []; + + const makeFetch = + (requestedRanges: string[]): typeof globalThis.fetch => + async (_url, options) => { + const rangeHeader = new Headers(options?.headers).get('Range'); + const match = rangeHeader?.match(/^bytes=(\d+)-(\d+)$/); + if (!match) { + return new Response(file, {status: 200}); + } + + const start = Number(match[1]); + const end = Math.min(Number(match[2]), file.byteLength - 1); + requestedRanges.push(match[0]); + return new Response(file.subarray(start, end + 1), { + status: 206, + headers: {'Content-Range': `bytes ${start}-${end}/${file.byteLength}`} + }); + }; + const makeSource = (fetch: typeof globalThis.fetch) => + new GeoTIFFRasterSource('https://example.com/shared.tif', { + core: {loadOptions: {core: {fetch}}}, + geotiff: {rangeScheduler} + }); + const firstSource = makeSource(makeFetch(firstRanges)); + const secondSource = makeSource(makeFetch(secondRanges)); + + await Promise.all([firstSource.getMetadata(), secondSource.getMetadata()]); + + expect(firstRanges.length).toBeGreaterThan(0); + expect(secondRanges.length).toBeGreaterThan(0); +}); + test('GeoTIFFRasterSource preserves rangeSchedulerProps object references', async () => { const file = await readFixtureBytes(TIFF_URL); const rangeStats = createRangeStats('geotiff-range-scheduler-props'); diff --git a/modules/geotiff/src/geotiff-source-loader.ts b/modules/geotiff/src/geotiff-source-loader.ts index 54a3250b5e..d3d0d34d48 100644 --- a/modules/geotiff/src/geotiff-source-loader.ts +++ b/modules/geotiff/src/geotiff-source-loader.ts @@ -15,7 +15,8 @@ import type { RasterData, RasterChannelDataType, RasterBoundingBox, - RangeRequestSchedulerProps + RangeRequestSchedulerProps, + RangeRequestTransportResult } from '@loaders.gl/loader-utils'; import { DataSource, @@ -576,6 +577,7 @@ class GeoTIFFRangeSchedulerClient { private readonly sourceId: string; private readonly defaultHeaders?: HeadersInit; private readonly rangeScheduler: RangeRequestScheduler; + private readonly schedulerIsolationKey = {}; private fileSize: number | null = null; /** Creates a new range-scheduled GeoTIFF client. */ @@ -605,6 +607,7 @@ class GeoTIFFRangeSchedulerClient { const arrayBuffer = await this.rangeScheduler.scheduleRequest({ sourceId: this.sourceId, + isolationKey: this.schedulerIsolationKey, offset, length, signal, @@ -626,7 +629,7 @@ class GeoTIFFRangeSchedulerClient { length: number, headers: Headers, signal?: AbortSignal - ): Promise { + ): Promise { let response = await this.fetch(this.url, { headers: createRangeRequestHeaders(headers, offset, length), signal @@ -660,7 +663,13 @@ class GeoTIFFRangeSchedulerClient { } } - return await response.arrayBuffer(); + const arrayBuffer = await response.arrayBuffer(); + return { + arrayBuffer, + status: response.status, + sourceByteLength: this.fileSize ?? undefined, + transportBytes: arrayBuffer.byteLength + }; } } diff --git a/modules/loader-utils/src/index.ts b/modules/loader-utils/src/index.ts index 9fb56ae160..dfc5a1d9b0 100644 --- a/modules/loader-utils/src/index.ts +++ b/modules/loader-utils/src/index.ts @@ -220,6 +220,13 @@ export {stream}; export type {ReadableFile, WritableFile, Stat} from './lib/files/file'; export {BlobFile} from './lib/files/blob-file'; export {HttpFile} from './lib/files/http-file'; +export type { + HttpFileConsistency, + HttpFileFetch, + HttpFileIdentity, + HttpFileOptions, + HttpFileTelemetry +} from './lib/files/http-file'; export {NodeFileFacade as NodeFile} from './lib/files/node-file-facade'; export type {FileSystem, RandomAccessFileSystem} from './lib/filesystems/filesystem'; diff --git a/modules/loader-utils/src/lib/files/http-file-transport.ts b/modules/loader-utils/src/lib/files/http-file-transport.ts new file mode 100644 index 0000000000..cdd589cb66 --- /dev/null +++ b/modules/loader-utils/src/lib/files/http-file-transport.ts @@ -0,0 +1,432 @@ +// loaders.gl +// SPDX-License-Identifier: MIT +// Copyright (c) vis.gl contributors + +import type {RangeRequestTransportResult} from '../request-utils/range-request-scheduler'; +import type { + HttpFileConsistency, + HttpFileFetch, + HttpFileIdentity, + HttpFileOptions, + HttpFileTelemetry +} from './http-file-types'; + +type MutableHttpFileTelemetry = { + /** Bytes requested from HTTP. */ + requestedBytes: number; + /** Response body bytes consumed. */ + downloadedBytes: number; + /** HTTP requests started. */ + requestCount: number; + /** Aggregate request duration. */ + networkTimeMs: number; + /** Cancelled operations. */ + abortCount: number; + /** Non-cancellation failures. */ + errorCount: number; +}; + +type HttpFileIdentityHints = { + /** Expected object length, when supplied by the caller. */ + byteLength?: number; + /** Expected ETag, when supplied by the caller. */ + etag?: string; + /** Expected Last-Modified value, when supplied by the caller. */ + lastModified?: string; +}; + +type ParsedContentRange = { + /** First returned byte. */ + offset: number; + /** Last returned byte. */ + endOffset: number; + /** Authoritative object length. */ + byteLength: number; +}; + +/** Result of one validated HTTP range request. */ +export type HttpFileRangeResult = RangeRequestTransportResult & { + /** Identity observed or confirmed by the response. */ + identity: HttpFileIdentity; +}; + +/** Owns HTTP request construction, response validation, identity checks, and transport telemetry. */ +export class HttpFileTransport { + /** Remote object URL. */ + private readonly url: string; + /** Fetch implementation used for range requests. */ + private readonly fetchFunction: HttpFileFetch; + /** Persistent fetch options copied from the caller. */ + private readonly fetchOptions?: RequestInit; + /** Validator enforcement mode. */ + private readonly consistency: HttpFileConsistency; + /** Caller-supplied identity information. */ + private readonly identityHints: HttpFileIdentityHints; + /** Failure objects already included in telemetry. */ + private readonly countedFailures = new WeakSet(); + /** Mutable counters backing immutable telemetry snapshots. */ + private readonly telemetry: MutableHttpFileTelemetry = { + requestedBytes: 0, + downloadedBytes: 0, + requestCount: 0, + networkTimeMs: 0, + abortCount: 0, + errorCount: 0 + }; + + /** Creates the transport for one remote object. */ + constructor(url: string, options: HttpFileOptions) { + validateKnownByteLength(options.byteLength); + + this.url = url; + this.fetchFunction = options.fetch || ((input, init) => globalThis.fetch(input, init)); + this.fetchOptions = options.fetchOptions + ? {...options.fetchOptions, headers: new Headers(options.fetchOptions.headers)} + : undefined; + this.consistency = options.consistency || 'best-effort'; + this.identityHints = { + byteLength: options.byteLength, + etag: normalizeHeaderValue(options.etag), + lastModified: normalizeHeaderValue(options.lastModified) + }; + } + + /** Returns an identity that is complete enough to avoid an opening request. */ + getInitialIdentity(): HttpFileIdentity | null { + const {byteLength, etag, lastModified} = this.identityHints; + if (byteLength !== undefined && (etag !== undefined || lastModified !== undefined)) { + return createIdentity(byteLength, etag, lastModified); + } + if (byteLength === 0 && this.consistency === 'best-effort') { + return createIdentity(0); + } + return null; + } + + /** Returns an immutable point-in-time copy of the transport counters. */ + getTelemetry(): HttpFileTelemetry { + return Object.freeze({...this.telemetry}); + } + + /** Counts each failure object once even when it passes through several async layers. */ + trackFailure(error: unknown): void { + if (typeof error === 'object' && error !== null) { + if (this.countedFailures.has(error)) { + return; + } + this.countedFailures.add(error); + } + if (isAbortError(error)) { + this.telemetry.abortCount++; + } else { + this.telemetry.errorCount++; + } + } + + /** Performs and validates one exact HTTP range request. */ + async requestRange( + offset: number, + length: number, + signal: AbortSignal | undefined, + expectedIdentity: HttpFileIdentity | null + ): Promise { + const requestStartTime = getTimestamp(); + const abortContext = createCombinedAbortContext(signal, this.fetchOptions?.signal); + this.telemetry.requestCount++; + this.telemetry.requestedBytes += length; + + try { + const response = await this.fetchFunction( + this.url, + createFetchOptions(this.fetchOptions, offset, length, abortContext.signal) + ); + if (isEmptyIdentityProbeResponse(response, offset, length, expectedIdentity)) { + return await this.readEmptyIdentityProbe(response, requestStartTime); + } + if (response.status !== 206) { + await cancelResponse(response); + throw new Error(`HTTP byte-range request expected 206, received ${response.status}`); + } + + const contentRange = await readContentRange(response); + const responseIdentity = getResponseIdentity(response, contentRange.byteLength); + const responseError = + getIdentityError( + expectedIdentity, + responseIdentity, + this.identityHints, + this.consistency + ) || getRangeError(contentRange, offset, length); + if (responseError) { + await cancelResponse(response); + throw new Error(responseError); + } + + const arrayBuffer = await response.arrayBuffer(); + this.telemetry.downloadedBytes += arrayBuffer.byteLength; + if (arrayBuffer.byteLength !== length) { + throw new Error( + `HTTP byte-range response contained ${arrayBuffer.byteLength} bytes; expected ${length}` + ); + } + + return { + arrayBuffer, + identity: resolveIdentity(expectedIdentity, responseIdentity, this.identityHints), + status: response.status, + transportBytes: arrayBuffer.byteLength, + networkTimeMs: getTimestamp() - requestStartTime + }; + } catch (error) { + if (expectedIdentity === null || !isAbortError(error)) { + this.trackFailure(error); + } + throw error; + } finally { + abortContext.removeAbortListeners(); + this.telemetry.networkTimeMs += getTimestamp() - requestStartTime; + } + } + + /** Validates and consumes the special empty-object identity response. */ + private async readEmptyIdentityProbe( + response: Response, + requestStartTime: number + ): Promise { + const responseIdentity = getResponseIdentity(response, 0); + const identityError = getIdentityError( + null, + responseIdentity, + this.identityHints, + this.consistency + ); + await cancelResponse(response); + if (identityError) { + throw new Error(identityError); + } + return { + arrayBuffer: new ArrayBuffer(0), + identity: resolveIdentity(null, responseIdentity, this.identityHints), + status: response.status, + transportBytes: 0, + networkTimeMs: getTimestamp() - requestStartTime + }; + } +} + +/** Creates fetch options while preserving caller headers and forcing an exact GET range. */ +function createFetchOptions( + fetchOptions: RequestInit | undefined, + offset: number, + length: number, + signal?: AbortSignal +): RequestInit { + const headers = new Headers(fetchOptions?.headers); + headers.set('Range', `bytes=${offset}-${offset + length - 1}`); + return { + ...fetchOptions, + method: 'GET', + body: undefined, + headers, + signal + }; +} + +/** Returns whether a response is the valid empty-object form of an opening probe. */ +function isEmptyIdentityProbeResponse( + response: Response, + offset: number, + length: number, + expectedIdentity: HttpFileIdentity | null +): boolean { + return ( + response.status === 416 && + expectedIdentity === null && + offset === 0 && + length === 1 && + /^bytes \*\/0$/i.test(response.headers.get('Content-Range') || '') + ); +} + +/** Parses Content-Range and cancels the body when the header is invalid. */ +async function readContentRange(response: Response): Promise { + try { + return parseContentRange(response.headers.get('Content-Range')); + } catch (error) { + await cancelResponse(response); + throw error; + } +} + +/** Parses an exact byte Content-Range response header. */ +function parseContentRange(contentRange: string | null): ParsedContentRange { + const match = contentRange?.match(/^bytes (\d+)-(\d+)\/(\d+)$/); + if (!match) { + throw new Error('HTTP byte-range response has an invalid Content-Range header'); + } + return { + offset: Number(match[1]), + endOffset: Number(match[2]), + byteLength: Number(match[3]) + }; +} + +/** Reads a normalized object identity from response headers. */ +function getResponseIdentity(response: Response, byteLength: number): HttpFileIdentity { + return createIdentity( + byteLength, + normalizeHeaderValue(response.headers.get('ETag')), + normalizeHeaderValue(response.headers.get('Last-Modified')) + ); +} + +/** Resolves the pinned identity from an expected identity, response, and caller hints. */ +function resolveIdentity( + expectedIdentity: HttpFileIdentity | null, + responseIdentity: HttpFileIdentity, + identityHints: HttpFileIdentityHints +): HttpFileIdentity { + return ( + expectedIdentity || + createIdentity( + responseIdentity.byteLength, + responseIdentity.etag ?? identityHints.etag, + responseIdentity.lastModified ?? identityHints.lastModified + ) + ); +} + +/** Returns an error message when a response does not cover the requested range. */ +function getRangeError( + contentRange: ParsedContentRange, + offset: number, + length: number +): string | null { + if ( + !Number.isSafeInteger(contentRange.offset) || + !Number.isSafeInteger(contentRange.endOffset) || + !Number.isSafeInteger(contentRange.byteLength) || + contentRange.offset < 0 || + contentRange.endOffset < contentRange.offset || + contentRange.endOffset >= contentRange.byteLength + ) { + return 'HTTP byte-range response has invalid range bounds'; + } + if (contentRange.offset !== offset || contentRange.endOffset !== offset + length - 1) { + return 'HTTP byte-range response does not match the requested range'; + } + return null; +} + +/** Returns an error message when response identity is inconsistent with the open file. */ +function getIdentityError( + expectedIdentity: HttpFileIdentity | null, + responseIdentity: HttpFileIdentity, + identityHints: HttpFileIdentityHints, + consistency: HttpFileConsistency +): string | null { + const expectedByteLength = expectedIdentity?.byteLength ?? identityHints.byteLength; + if (expectedByteLength !== undefined && responseIdentity.byteLength !== expectedByteLength) { + return 'HTTP file length changed while the file was open'; + } + + const expectedEtag = expectedIdentity?.etag ?? identityHints.etag; + const expectedLastModified = expectedIdentity?.lastModified ?? identityHints.lastModified; + if (expectedEtag) { + if (responseIdentity.etag) { + return expectedEtag === responseIdentity.etag + ? null + : 'HTTP file ETag changed while the file was open'; + } + if (consistency === 'strict') { + return 'HTTP byte-range response is missing the pinned ETag'; + } + if ( + expectedLastModified && + responseIdentity.lastModified && + expectedLastModified !== responseIdentity.lastModified + ) { + return 'HTTP file Last-Modified value changed while the file was open'; + } + return null; + } + if (expectedLastModified) { + if (responseIdentity.lastModified && expectedLastModified !== responseIdentity.lastModified) { + return 'HTTP file Last-Modified value changed while the file was open'; + } + if (consistency === 'strict' && !responseIdentity.lastModified) { + return 'HTTP byte-range response is missing the pinned Last-Modified value'; + } + return null; + } + if (consistency === 'strict' && !responseIdentity.etag && !responseIdentity.lastModified) { + return 'HTTP byte-range response does not provide an ETag or Last-Modified validator'; + } + return null; +} + +/** Creates a frozen normalized identity. */ +function createIdentity( + byteLength: number, + etag?: string, + lastModified?: string +): HttpFileIdentity { + return Object.freeze({byteLength, etag, lastModified}); +} + +/** Validates a caller-provided object length. */ +function validateKnownByteLength(byteLength?: number): void { + if (byteLength !== undefined && (!Number.isSafeInteger(byteLength) || byteLength < 0)) { + throw new Error('HttpFile byteLength must be a non-negative safe integer'); + } +} + +/** Cancels a response body without obscuring the validation error. */ +async function cancelResponse(response: Response): Promise { + await response.body?.cancel().catch(() => {}); +} + +/** Normalizes empty response-header values to undefined. */ +function normalizeHeaderValue(value: string | null | undefined): string | undefined { + return value || undefined; +} + +/** Returns a high-resolution timestamp when available. */ +function getTimestamp(): number { + return globalThis.performance?.now() ?? Date.now(); +} + +/** Returns whether an arbitrary failure represents cancellation. */ +function isAbortError(error: unknown): boolean { + return ( + (error instanceof Error && error.name === 'AbortError') || + (error instanceof Error && /aborted/i.test(error.message)) + ); +} + +/** Creates one fetch signal that follows both a per-read and persistent source signal. */ +function createCombinedAbortContext(...signals: (AbortSignal | null | undefined)[]): { + signal: AbortSignal; + removeAbortListeners: () => void; +} { + const abortController = new AbortController(); + const uniqueSignals = [ + ...new Set(signals.filter((signal): signal is AbortSignal => Boolean(signal))) + ]; + const abortListener = () => abortController.abort(); + for (const signal of uniqueSignals) { + if (signal.aborted) { + abortController.abort(); + } else { + signal.addEventListener('abort', abortListener, {once: true}); + } + } + return { + signal: abortController.signal, + removeAbortListeners: () => { + for (const signal of uniqueSignals) { + signal.removeEventListener('abort', abortListener); + } + } + }; +} diff --git a/modules/loader-utils/src/lib/files/http-file-types.ts b/modules/loader-utils/src/lib/files/http-file-types.ts new file mode 100644 index 0000000000..6bd32b2ae9 --- /dev/null +++ b/modules/loader-utils/src/lib/files/http-file-types.ts @@ -0,0 +1,60 @@ +// loaders.gl +// SPDX-License-Identifier: MIT +// Copyright (c) vis.gl contributors + +import type { + RangeRequestScheduler, + RangeRequestSchedulerProps +} from '../request-utils/range-request-scheduler'; + +/** Controls how rigorously an HTTP file verifies response validators. */ +export type HttpFileConsistency = 'strict' | 'best-effort'; + +/** Fetch implementation accepted by `HttpFile`. */ +export type HttpFileFetch = (url: string, options?: RequestInit) => Promise; + +/** Options for opening a byte-range-addressable HTTP file. */ +export type HttpFileOptions = { + /** Optional fetch implementation. */ + fetch?: HttpFileFetch; + /** Fetch options applied to every request. The Range header and signal are supplied per read. */ + fetchOptions?: RequestInit; + /** Known object length. Supplying a validator as well avoids an opening probe request. */ + byteLength?: number; + /** Known object ETag. */ + etag?: string; + /** Known object Last-Modified value. */ + lastModified?: string; + /** Whether missing validators are rejected. Defaults to best-effort. */ + consistency?: HttpFileConsistency; + /** Optional shared byte-range scheduler. */ + rangeScheduler?: RangeRequestScheduler; + /** Configuration used when creating a private scheduler. */ + rangeSchedulerProps?: RangeRequestSchedulerProps; +}; + +/** Immutable identity pinned for an open HTTP file. */ +export type HttpFileIdentity = Readonly<{ + /** Authoritative object length in bytes. */ + byteLength: number; + /** Pinned ETag, when exposed by the server. */ + etag?: string; + /** Pinned Last-Modified value, when exposed by the server. */ + lastModified?: string; +}>; + +/** Immutable snapshot of one HTTP file's transport counters. */ +export type HttpFileTelemetry = Readonly<{ + /** Bytes requested from HTTP after range coalescing, including the opening probe. */ + requestedBytes: number; + /** Response body bytes consumed by the file. */ + downloadedBytes: number; + /** Actual HTTP requests started by the file. */ + requestCount: number; + /** Aggregate wall time spent awaiting HTTP requests and response bodies. */ + networkTimeMs: number; + /** Failed or cancelled operations caused by an abort signal. */ + abortCount: number; + /** Failed operations not caused by cancellation. */ + errorCount: number; +}>; diff --git a/modules/loader-utils/src/lib/files/http-file.ts b/modules/loader-utils/src/lib/files/http-file.ts index 8cc046cb8b..f55d577538 100644 --- a/modules/loader-utils/src/lib/files/http-file.ts +++ b/modules/loader-utils/src/lib/files/http-file.ts @@ -2,121 +2,292 @@ // SPDX-License-Identifier: MIT // Copyright (c) vis.gl contributors -import {ReadableFile, Stat} from './file'; +import type {ReadableFile, Stat} from './file'; +import type {RangeRequestTransportResult} from '../request-utils/range-request-scheduler'; +import {RangeRequestScheduler} from '../request-utils/range-request-scheduler'; +import type {HttpFileIdentity, HttpFileOptions, HttpFileTelemetry} from './http-file-types'; +import {HttpFileTransport} from './http-file-transport'; +export type { + HttpFileConsistency, + HttpFileFetch, + HttpFileIdentity, + HttpFileOptions, + HttpFileTelemetry +} from './http-file-types'; + +/** A reusable, validator-aware, byte-range-addressable HTTP file. */ export class HttpFile implements ReadableFile { + /** URL used as the ReadableFile handle. */ readonly handle: string; - readonly size: number = 0; - readonly bigsize: bigint = 0n; + /** Remote object URL. */ readonly url: string; - constructor(url: string) { + /** HTTP transport and response validator for this file. */ + private readonly transport: HttpFileTransport; + /** Scheduler used to coalesce compatible range reads. */ + private readonly rangeScheduler: RangeRequestScheduler; + /** Unique key that prevents coalescing across transport or credential contexts. */ + private readonly schedulerIsolationKey = {}; + /** Caller-supplied length available before identity discovery. */ + private readonly suppliedByteLength?: number; + + /** Pinned remote object identity. */ + private identity: HttpFileIdentity | null; + /** Shared opening probe, while identity discovery is in progress. */ + private openPromise: Promise | null = null; + /** Whether the file has stopped accepting operations. */ + private closed = false; + + /** Creates a lazy HTTP file. Use {@link HttpFile.open} when identity must be pinned immediately. */ + constructor(url: string, options: HttpFileOptions = {}) { this.handle = url; this.url = url; + this.transport = new HttpFileTransport(url, options); + this.rangeScheduler = + options.rangeScheduler || + new RangeRequestScheduler({batchDelayMs: 0, ...options.rangeSchedulerProps}); + this.suppliedByteLength = options.byteLength; + this.identity = this.transport.getInitialIdentity(); } - async close(): Promise {} + /** Opens a file and pins its byte length and available validators. */ + static async open( + url: string, + options: HttpFileOptions = {}, + signal?: AbortSignal + ): Promise { + const file = new HttpFile(url, options); + await file.open(signal); + return file; + } - async stat(): Promise { - const response = await fetch(this.handle, {method: 'HEAD'}); - if (!response.ok) { - throw new Error(`Failed to fetch HEAD ${this.handle}`); + /** Pins this file's byte length and available validators. */ + async open(signal?: AbortSignal): Promise { + this.assertOpen(); + if (signal?.aborted) { + const error = createAbortError(); + this.transport.trackFailure(error); + throw error; } - const size = parseInt(response.headers.get('Content-Length') || '0'); + await this.getIdentity(signal); + return this; + } + + /** Pinned file length, or zero before a lazy file has opened. */ + get size(): number { + return this.identity?.byteLength ?? this.suppliedByteLength ?? 0; + } + + /** Pinned file length as a bigint, or zero before a lazy file has opened. */ + get bigsize(): bigint { + return BigInt(this.size); + } + + /** Returns the pinned identity, or null before a lazy file has opened. */ + getIdentitySnapshot(): HttpFileIdentity | null { + return this.identity; + } + + /** Returns an immutable point-in-time copy of this file's transport counters. */ + getTelemetry(): HttpFileTelemetry { + return this.transport.getTelemetry(); + } + + /** Prevents new operations. Active fetches remain controlled by their read signals. */ + async close(): Promise { + this.closed = true; + } + + /** Returns pinned file information without issuing a HEAD request. */ + async stat(): Promise { + this.assertOpen(); + const identity = await this.getIdentity(); return { - size, - bigsize: BigInt(size), + size: identity.byteLength, + bigsize: BigInt(identity.byteLength), isDirectory: false }; } - async read(offset: number | bigint = 0, length: number = 0): Promise { - const response = await this.fetchRange(offset, length); - const arrayBuffer = await response.arrayBuffer(); - return arrayBuffer; + /** Reads exactly one byte range, with optional cancellation. */ + async read( + offset: number | bigint = 0, + length: number = 0, + signal?: AbortSignal + ): Promise { + this.assertOpen(); + if (signal?.aborted) { + const error = createAbortError(); + this.transport.trackFailure(error); + throw error; + } + + const identity = await this.getIdentity(signal); + const numericOffset = normalizeOffset(offset); + validateRange(numericOffset, length, identity.byteLength); + if (length === 0) { + return new ArrayBuffer(0); + } + + try { + return await this.rangeScheduler.scheduleRequest({ + sourceId: this.url, + isolationKey: this.schedulerIsolationKey, + offset: numericOffset, + length, + signal, + fetchRange: this.fetchScheduledRange + }); + } catch (error) { + this.transport.trackFailure(error); + throw error; + } } - /** - * - * @param offset - * @param length - * @param signal - * @returns - * @see https://github.com/protomaps/PMTiles - */ - // eslint-disable-next-line complexity + /** Reads a range and returns a Response for legacy ReadableFile consumers. */ async fetchRange( offset: number | bigint, length: number, signal?: AbortSignal ): Promise { - const nOffset = Number(offset); - const nLength = Number(length); - - let controller: AbortController | undefined; - if (!signal) { - // ToDO why is it so important to abort in case 200? - // TODO check this works or assert 206 - controller = new AbortController(); - signal = controller.signal; - } + const numericOffset = normalizeOffset(offset); + const arrayBuffer = await this.read(numericOffset, length, signal); + const headers = createIdentityHeaders(this.identity, numericOffset, length); + return new Response(arrayBuffer, {status: 206, headers}); + } - const url = this.handle; - let response = await fetch(url, { - signal, - headers: {Range: `bytes=${nOffset}-${nOffset + nLength - 1}`} - }); + /** Stable scheduler callback used by every read from this file. */ + private readonly fetchScheduledRange = async ( + offset: number, + length: number, + signal?: AbortSignal + ): Promise => { + const identity = await this.getIdentity(signal); + const result = await this.transport.requestRange(offset, length, signal, identity); + return { + arrayBuffer: result.arrayBuffer, + status: result.status, + transportBytes: result.transportBytes, + networkTimeMs: result.networkTimeMs + }; + }; - switch (response.status) { - case 206: // Partial Content success - // This is the expected success code for a range request - break; - - case 200: - // some well-behaved backends, e.g. DigitalOcean CDN, respond with 200 instead of 206 - // but we also need to detect no support for Byte Serving which is returning the whole file - const contentLength = response.headers.get('Content-Length'); - if (!contentLength || Number(contentLength) > length) { - if (controller) { - controller.abort(); - } - throw Error( - 'content-length header missing or exceeding request. Server must support HTTP Byte Serving.' - ); + /** Returns the shared lazy identity promise, retrying after a failed probe. */ + private getIdentity(signal?: AbortSignal): Promise { + if (this.identity) { + return Promise.resolve(this.identity); + } + if (!this.openPromise) { + // The shared probe must not be owned by the first caller's cancellation signal. + // Each caller can independently stop waiting while the probe populates the cache. + const nextOpenPromise = this.probeIdentity(); + const cachedOpenPromise = nextOpenPromise.catch(error => { + if (this.openPromise === cachedOpenPromise) { + this.openPromise = null; } + throw error; + }); + this.openPromise = cachedOpenPromise; + } + return waitForPromiseWithSignal(this.openPromise, signal).catch(error => { + this.transport.trackFailure(error); + throw error; + }); + } - // @eslint-disable-next-line no-fallthrough - case 416: // "Range Not Satisfiable" - // some HTTP servers don't accept ranges beyond the end of the resource. - // Retry with the exact length - // TODO: can return 416 with offset > 0 if content changed, which will have a blank etag. - // See https://github.com/protomaps/PMTiles/issues/90 - if (offset === 0) { - const contentRange = response.headers.get('Content-Range'); - if (!contentRange || !contentRange.startsWith('bytes *')) { - throw Error('Missing content-length on 416 response'); - } - const actualLength = Number(contentRange.substr(8)); - response = await fetch(this.url, { - signal, - headers: {Range: `bytes=0-${actualLength - 1}`} - }); - } - break; + /** Uses a one-byte GET range to discover object length and validators. */ + private async probeIdentity(): Promise { + try { + const result = await this.transport.requestRange(0, 1, undefined, null); + this.identity = result.identity; + return result.identity; + } catch (error) { + this.transport.trackFailure(error); + throw error; + } + } - default: - if (response.status >= 300) { - throw Error(`Bad response code: ${response.status}`); - } + /** Throws when the file has been closed. */ + private assertOpen(): void { + if (this.closed) { + throw new Error('HttpFile is closed'); } + } +} + +/** Creates the identity headers returned by the compatibility fetchRange method. */ +function createIdentityHeaders( + identity: HttpFileIdentity | null, + offset: number, + length: number +): Headers { + const headers = new Headers(); + if (length > 0 && identity) { + headers.set('Content-Range', `bytes ${offset}-${offset + length - 1}/${identity.byteLength}`); + } + if (identity?.etag) { + headers.set('ETag', identity.etag); + } + if (identity?.lastModified) { + headers.set('Last-Modified', identity.lastModified); + } + return headers; +} + +/** Converts and validates a number or bigint offset. */ +function normalizeOffset(offset: number | bigint): number { + const numericOffset = Number(offset); + if (!Number.isSafeInteger(numericOffset) || numericOffset < 0) { + throw new Error('HTTP byte-range offset must be a non-negative safe integer'); + } + return numericOffset; +} - return response; - // const data = await response.arrayBuffer(); - // return { - // data, - // etag: response.headers.get('ETag') || undefined, - // cacheControl: response.headers.get('Cache-Control') || undefined, - // expires: response.headers.get('Expires') || undefined - // }; +/** Validates a requested byte range against the pinned object length. */ +function validateRange(offset: number, length: number, byteLength: number): void { + if (!Number.isSafeInteger(length) || length < 0) { + throw new Error('HTTP byte-range length must be a non-negative safe integer'); + } + if (offset + length > byteLength) { + throw new Error('HTTP byte range extends beyond the end of the file'); } } + +/** Creates a conventional abort error without requiring DOMException in every host. */ +function createAbortError(): Error { + if (typeof DOMException !== 'undefined') { + return new DOMException('Request aborted', 'AbortError'); + } + const error = new Error('Request aborted'); + error.name = 'AbortError'; + return error; +} + +/** Lets one caller cancel its wait without cancelling a shared operation. */ +function waitForPromiseWithSignal(promise: Promise, signal?: AbortSignal): Promise { + if (!signal) { + return promise; + } + if (signal.aborted) { + return Promise.reject(createAbortError()); + } + + return new Promise((resolve, reject) => { + const abortListener = () => { + signal.removeEventListener('abort', abortListener); + reject(createAbortError()); + }; + signal.addEventListener('abort', abortListener, {once: true}); + promise.then( + value => { + signal.removeEventListener('abort', abortListener); + resolve(value); + }, + error => { + signal.removeEventListener('abort', abortListener); + reject(error); + } + ); + }); +} diff --git a/modules/loader-utils/src/lib/request-utils/range-request-scheduler.ts b/modules/loader-utils/src/lib/request-utils/range-request-scheduler.ts index d015817c5b..0f9d809753 100644 --- a/modules/loader-utils/src/lib/request-utils/range-request-scheduler.ts +++ b/modules/loader-utils/src/lib/request-utils/range-request-scheduler.ts @@ -22,6 +22,8 @@ export type RangeStats = { transportBytes: number; /** Number of bytes read from transport responses. */ responseBytes: number; + /** Aggregate time spent awaiting transport requests, in milliseconds. */ + networkTimeMs: number; /** Number of transport-requested bytes that were not part of logical caller requests. */ overfetchBytes: number; /** Number of transport ranges that failed. */ @@ -58,6 +60,8 @@ export type RangeRequest = { length: number; /** Optional caller abort signal. */ signal?: AbortSignal; + /** Requests only coalesce when this key is identical. Defaults to the shared source identifier. */ + isolationKey?: unknown; /** Fetches bytes for this source. */ fetchRange: ( offset: number, @@ -78,6 +82,8 @@ export type RangeFetchRequest = { length: number; /** Optional caller abort signal. */ signal?: AbortSignal; + /** Requests only coalesce when this key is identical. Omit it to isolate this fetch call. */ + isolationKey?: unknown; /** Optional fetch implementation for tests or host environments. */ fetch?: (url: string, options?: RequestInit) => Promise; /** Optional fetch options merged into the transport request. */ @@ -90,8 +96,12 @@ export type RangeRequestTransportResult = { arrayBuffer: ArrayBuffer; /** HTTP status code or transport-specific equivalent. */ status?: number; + /** Total source length, when known. Used to validate a response clamped at end of file. */ + sourceByteLength?: number; /** Number of response bytes read from the transport before slicing. */ transportBytes?: number; + /** Time spent awaiting this transport request, in milliseconds. */ + networkTimeMs?: number; /** True when the transport returned the complete object and the requested range was sliced locally. */ fullResponse?: boolean; }; @@ -116,6 +126,8 @@ export type RangeRequestEvent = { transportBytes?: number; /** Bytes returned by the transport before slicing. */ responseBytes?: number; + /** Time spent awaiting the transport request, in milliseconds. */ + networkTimeMs?: number; /** Bytes fetched only to bridge nearby requested ranges. */ overfetchBytes?: number; /** HTTP status code or transport-specific equivalent. */ @@ -129,10 +141,14 @@ export type RangeRequestEvent = { type PendingRequest = RangeRequest & { resolve: (arrayBuffer: ArrayBuffer) => void; reject: (error: unknown) => void; + settled: boolean; + abortListener?: () => void; + onAbort?: () => void; }; type MergedRequest = { sourceId: string; + isolationKey: unknown; offset: number; endOffset: number; fetchRange: RangeRequest['fetchRange']; @@ -149,6 +165,7 @@ const RANGE_STATS_KEYS: Record = { requestedBytes: 'Logical Range Request Bytes', transportBytes: 'Range Transport Bytes Requested', responseBytes: 'Range Response Bytes', + networkTimeMs: 'Range Network Time', overfetchBytes: 'Range Overfetch Bytes', failedTransportRanges: 'Range Transport Requests Failed', abortedLogicalRanges: 'Aborted Logical Range Requests', @@ -203,11 +220,23 @@ export class RangeRequestScheduler { } if (request.signal?.aborted) { - return Promise.reject(new Error('Request aborted')); + return Promise.reject(createAbortError()); + } + + if (request.length === 0) { + return Promise.resolve(new ArrayBuffer(0)); } const promise = new Promise((resolve, reject) => { - this.pendingRequests.push({...request, resolve, reject}); + const pendingRequest: PendingRequest = {...request, resolve, reject, settled: false}; + if (request.signal) { + pendingRequest.abortListener = () => { + this.abortRequest(pendingRequest); + pendingRequest.onAbort?.(); + }; + request.signal.addEventListener('abort', pendingRequest.abortListener, {once: true}); + } + this.pendingRequests.push(pendingRequest); }); this.trackEvent({ @@ -233,6 +262,9 @@ export class RangeRequestScheduler { offset: request.offset, length: request.length, signal: request.signal, + // Different fetch calls may carry different credentials or transport implementations. + // Sharing is therefore opt-in through an explicit common isolation key. + isolationKey: request.isolationKey ?? {}, fetchRange: (offset, length, signal) => fetchHttpRange({ url: request.url, @@ -256,16 +288,8 @@ export class RangeRequestScheduler { this.pendingRequests = []; const activeRequests = pendingRequests.filter(request => { - if (request.signal?.aborted) { - request.reject(new Error('Request aborted')); - this.trackEvent({ - type: 'abort', - sourceId: request.sourceId, - offset: request.offset, - length: request.length, - logicalRequestCount: 1, - logicalBytes: request.length - }); + if (request.settled || request.signal?.aborted) { + this.abortRequest(request); return false; } return true; @@ -310,6 +334,7 @@ export class RangeRequestScheduler { } else { mergedRequests.push({ sourceId: request.sourceId, + isolationKey: request.isolationKey ?? request.sourceId, offset: request.offset, endOffset: requestEndOffset, fetchRange: request.fetchRange, @@ -325,7 +350,8 @@ export class RangeRequestScheduler { private async fetchMergedRequest(mergedRequest: MergedRequest): Promise { const {offset, endOffset, fetchRange, requests} = mergedRequest; const abortController = new AbortController(); - const abortListener = () => { + const requestStartTime = getTimestamp(); + const abortTransportIfUnused = () => { if (requests.every(request => request.signal?.aborted)) { abortController.abort(); } @@ -333,7 +359,11 @@ export class RangeRequestScheduler { try { for (const request of requests) { - request.signal?.addEventListener('abort', abortListener, {once: true}); + request.onAbort = abortTransportIfUnused; + } + abortTransportIfUnused(); + if (requests.every(request => request.settled)) { + return; } const logicalBytes = getLogicalRequestBytes(requests); @@ -353,6 +383,18 @@ export class RangeRequestScheduler { await fetchRange(offset, length, abortController.signal) ); const {arrayBuffer} = transportResult; + const isClampedFullResponse = + transportResult.fullResponse && offset === 0 && arrayBuffer.byteLength < length; + const isClampedEndOfFile = + arrayBuffer.byteLength < length && + Number.isSafeInteger(transportResult.sourceByteLength) && + offset + arrayBuffer.byteLength === transportResult.sourceByteLength; + if (arrayBuffer.byteLength !== length && !isClampedFullResponse && !isClampedEndOfFile) { + throw new Error( + `Byte-range transport returned ${arrayBuffer.byteLength} bytes; expected ${length}` + ); + } + const networkTimeMs = transportResult.networkTimeMs ?? getTimestamp() - requestStartTime; this.trackEvent({ type: 'response', @@ -363,26 +405,19 @@ export class RangeRequestScheduler { logicalBytes, transportBytes: length, responseBytes: transportResult.transportBytes ?? arrayBuffer.byteLength, + networkTimeMs, overfetchBytes: Math.max(length - logicalBytes, 0), status: transportResult.status, fullResponse: transportResult.fullResponse }); for (const request of requests) { - if (request.signal?.aborted) { - request.reject(new Error('Request aborted')); - this.trackEvent({ - type: 'abort', - sourceId: request.sourceId, - offset: request.offset, - length: request.length, - logicalRequestCount: 1, - logicalBytes: request.length - }); + if (request.settled) { continue; } const start = request.offset - offset; + request.settled = true; request.resolve(arrayBuffer.slice(start, start + request.length)); } } catch (error) { @@ -393,23 +428,56 @@ export class RangeRequestScheduler { length: endOffset - offset, logicalRequestCount: requests.length, logicalBytes: getLogicalRequestBytes(requests), + networkTimeMs: getTimestamp() - requestStartTime, error }); for (const request of requests) { - request.reject(error); + if (request.settled) { + continue; + } + if (request.signal?.aborted) { + this.abortRequest(request); + } else { + request.settled = true; + request.reject(error); + } } } finally { for (const request of requests) { - request.signal?.removeEventListener('abort', abortListener); + if (request.abortListener) { + request.signal?.removeEventListener('abort', request.abortListener); + } + request.onAbort = undefined; } } } + /** Rejects and tracks one logical request exactly once when its caller aborts. */ + private abortRequest(request: PendingRequest): void { + if (request.settled) { + return; + } + request.settled = true; + request.reject(createAbortError()); + this.trackEvent({ + type: 'abort', + sourceId: request.sourceId, + offset: request.offset, + length: request.length, + logicalRequestCount: 1, + logicalBytes: request.length + }); + } + /** Emits one event to Stats and to the optional callback. */ private trackEvent(event: RangeRequestEvent): void { trackStatsEvent(this.stats, event); - this.onEvent?.(event); + try { + this.onEvent?.(event); + } catch { + // Diagnostics must not change transport behavior or orphan queued requests. + } } /** Emits one batch event after logical requests have been merged. */ @@ -434,6 +502,10 @@ export class RangeRequestScheduler { * Tracks range batching events in a probe.gl Stats object. */ export function trackStatsEvent(stats: Stats, event: RangeRequestEvent): void { + if ((event.type === 'response' || event.type === 'error') && event.networkTimeMs !== undefined) { + stats.get(RANGE_STATS_KEYS.networkTimeMs, 'time').addTime(event.networkTimeMs); + } + switch (event.type) { case 'queued': stats.get(RANGE_STATS_KEYS.logicalRanges, 'count').incrementCount(); @@ -488,6 +560,7 @@ export function getRangeStats(stats: Stats): RangeStats { requestedBytes: stats.get(RANGE_STATS_KEYS.requestedBytes).count, transportBytes: stats.get(RANGE_STATS_KEYS.transportBytes).count, responseBytes: stats.get(RANGE_STATS_KEYS.responseBytes).count, + networkTimeMs: stats.get(RANGE_STATS_KEYS.networkTimeMs).time, overfetchBytes: stats.get(RANGE_STATS_KEYS.overfetchBytes).count, failedTransportRanges: stats.get(RANGE_STATS_KEYS.failedTransportRanges).count, abortedLogicalRanges: stats.get(RANGE_STATS_KEYS.abortedLogicalRanges).count, @@ -501,8 +574,10 @@ export async function fetchHttpRange( Pick ): Promise { const abortContext = createAbortableFetchContext(request.signal); + const requestStartTime = getTimestamp(); try { + let fullResponse = false; let response = await request.fetch( request.url, createRangeFetchOptions( @@ -522,6 +597,7 @@ export async function fetchHttpRange( request.url, createRangeFetchOptions(request.fetchOptions, 0, actualLength, abortContext.signal) ); + fullResponse = true; } if (response.status === 200) { @@ -535,10 +611,31 @@ export async function fetchHttpRange( } const arrayBuffer = await response.arrayBuffer(); + const contentRangeHeader = response.headers.get('Content-Range'); + const contentRange = parseSatisfiedContentRange(contentRangeHeader); + if (contentRangeHeader && !contentRange) { + throw new Error(`Invalid Content-Range header: ${contentRangeHeader}`); + } + if (contentRange) { + const responseLength = contentRange.endOffset - contentRange.offset + 1; + const requestedEndOffset = request.offset + request.length - 1; + if ( + contentRange.offset !== request.offset || + contentRange.endOffset > requestedEndOffset || + responseLength !== arrayBuffer.byteLength || + (contentRange.sourceByteLength !== undefined && + contentRange.endOffset >= contentRange.sourceByteLength) + ) { + throw new Error(`Content-Range does not match requested range: ${contentRangeHeader}`); + } + } return { arrayBuffer, status: response.status, - transportBytes: arrayBuffer.byteLength + sourceByteLength: contentRange?.sourceByteLength, + transportBytes: arrayBuffer.byteLength, + networkTimeMs: getTimestamp() - requestStartTime, + fullResponse }; } finally { abortContext.removeAbortListener(); @@ -554,6 +651,9 @@ function canMergeRequests( if (mergedRequest.sourceId !== request.sourceId) { return false; } + if (mergedRequest.isolationKey !== (request.isolationKey ?? request.sourceId)) { + return false; + } const requestEndOffset = request.offset + request.length; const gapBytes = request.offset - mergedRequest.endOffset; @@ -573,6 +673,7 @@ function initializeStats(stats: Stats): void { stats.get(RANGE_STATS_KEYS.transportBytes, 'count'); stats.get(RANGE_STATS_KEYS.completedTransportRanges, 'count'); stats.get(RANGE_STATS_KEYS.responseBytes, 'count'); + stats.get(RANGE_STATS_KEYS.networkTimeMs, 'time'); stats.get(RANGE_STATS_KEYS.overfetchBytes, 'count'); stats.get(RANGE_STATS_KEYS.fullResponseFallbacks, 'count'); stats.get(RANGE_STATS_KEYS.failedTransportRanges, 'count'); @@ -628,6 +729,30 @@ function parseUnsatisfiedContentRange(contentRange: string | null): number | nul return match ? Number(match[1]) : null; } +/** Parses a `Content-Range: bytes -/` response header. */ +function parseSatisfiedContentRange(contentRange: string | null): { + offset: number; + endOffset: number; + sourceByteLength?: number; +} | null { + const match = contentRange?.match(/^bytes (\d+)-(\d+)\/(\d+|\*)$/); + if (!match) { + return null; + } + const offset = Number(match[1]); + const endOffset = Number(match[2]); + const sourceByteLength = match[3] === '*' ? undefined : Number(match[3]); + if ( + !Number.isSafeInteger(offset) || + !Number.isSafeInteger(endOffset) || + offset > endOffset || + (sourceByteLength !== undefined && !Number.isSafeInteger(sourceByteLength)) + ) { + return null; + } + return {offset, endOffset, sourceByteLength}; +} + /** Creates an inner abort controller that follows an optional parent signal and can cancel one fetch. */ function createAbortableFetchContext(parentSignal?: AbortSignal): { signal: AbortSignal; @@ -652,3 +777,18 @@ function createAbortableFetchContext(parentSignal?: AbortSignal): { async function cancelIgnoredRangeResponse(response: Response): Promise { await response.body?.cancel().catch(() => {}); } + +/** Returns a high-resolution timestamp when available. */ +function getTimestamp(): number { + return globalThis.performance?.now() ?? Date.now(); +} + +/** Creates a conventional abort error without requiring DOMException in every host. */ +function createAbortError(): Error { + if (typeof DOMException !== 'undefined') { + return new DOMException('Request aborted', 'AbortError'); + } + const error = new Error('Request aborted'); + error.name = 'AbortError'; + return error; +} diff --git a/modules/loader-utils/test/index.ts b/modules/loader-utils/test/index.ts index dfa452e357..3d5f2dc078 100644 --- a/modules/loader-utils/test/index.ts +++ b/modules/loader-utils/test/index.ts @@ -26,5 +26,6 @@ import './lib/sources/data-source-manager.spec'; // import './lib/filesystems/node-filesystem-facade.spec'; import './lib/readable-file/readable-file.spec'; +import './lib/readable-file/http-file.spec'; import './lib/worker-loader-utils/parse-with-worker.browser.spec'; diff --git a/modules/loader-utils/test/lib/readable-file/http-file.spec.ts b/modules/loader-utils/test/lib/readable-file/http-file.spec.ts new file mode 100644 index 0000000000..25beeaf0e8 --- /dev/null +++ b/modules/loader-utils/test/lib/readable-file/http-file.spec.ts @@ -0,0 +1,491 @@ +// loaders.gl +// SPDX-License-Identifier: MIT +// Copyright (c) vis.gl contributors + +import {expect, test} from 'vitest'; +import { + HttpFile, + RangeRequestScheduler, + getRangeStats, + type HttpFileFetch +} from '@loaders.gl/loader-utils'; + +const DATA = Uint8Array.from({length: 32}, (_, index) => index); +const URL = 'https://example.com/data.parquet'; + +test('HttpFile#open recognizes an empty 416 identity probe', async () => { + let requestedRange: string | null = null; + const file = await HttpFile.open(URL, { + fetch: async (_url, options) => { + requestedRange = new Headers(options?.headers).get('Range'); + return new Response(null, { + status: 416, + headers: { + 'Content-Range': 'bytes */0', + ETag: '"empty-version"' + } + }); + } + }); + + expect(requestedRange, 'probes the first byte').toBe('bytes=0-0'); + expect(file.getIdentitySnapshot(), 'pins the empty object identity').toEqual({ + byteLength: 0, + etag: '"empty-version"', + lastModified: undefined + }); + await expect(file.stat(), 'reports an empty file').resolves.toMatchObject({size: 0, bigsize: 0n}); + await expect(file.read(0, 0), 'serves an empty read without another request').resolves.toEqual( + new ArrayBuffer(0) + ); + expect(file.getTelemetry(), 'counts only the identity probe').toMatchObject({ + requestedBytes: 1, + downloadedBytes: 0, + requestCount: 1, + errorCount: 0 + }); +}); + +test('HttpFile#open pins identity and reports frozen transport telemetry', async () => { + const requests: RequestInit[] = []; + const fetch: HttpFileFetch = async (_url, options) => { + requests.push(options || {}); + return createRangeResponse(options, { + etag: '"version-1"', + lastModified: 'Mon, 01 Jan 2024 00:00:00 GMT' + }); + }; + + const file = await HttpFile.open(URL, { + fetch, + fetchOptions: {headers: {Authorization: 'Bearer token'}} + }); + expect(file.getIdentitySnapshot(), 'pins response identity').toEqual({ + byteLength: DATA.byteLength, + etag: '"version-1"', + lastModified: 'Mon, 01 Jan 2024 00:00:00 GMT' + }); + expect(Object.isFrozen(file.getIdentitySnapshot()), 'freezes response identity').toBeTruthy(); + expect(file.size, 'sets the synchronous byte length').toBe(DATA.byteLength); + + const bytes = await file.read(4, 3); + expect(Array.from(new Uint8Array(bytes)), 'returns the exact requested bytes').toEqual([4, 5, 6]); + expect( + requests.map(request => new Headers(request.headers).get('Range')), + 'uses one-byte discovery and exact data ranges' + ).toEqual(['bytes=0-0', 'bytes=4-6']); + expect( + requests.every(request => new Headers(request.headers).get('Authorization') === 'Bearer token'), + 'preserves caller headers' + ).toBeTruthy(); + + const telemetry = file.getTelemetry(); + expect(telemetry, 'reports transport counters').toMatchObject({ + requestedBytes: 4, + downloadedBytes: 4, + requestCount: 2, + abortCount: 0, + errorCount: 0 + }); + expect(telemetry.networkTimeMs, 'reports network time').toBeGreaterThanOrEqual(0); + expect(Object.isFrozen(telemetry), 'freezes telemetry snapshots').toBeTruthy(); +}); + +test('HttpFile#known identity avoids an opening probe and enforces strict validators', async () => { + let requestCount = 0; + const fetch: HttpFileFetch = async (_url, options) => { + requestCount++; + return createRangeResponse(options, { + etag: '"manifest-version"', + lastModified: 'Tue, 02 Jan 2024 00:00:00 GMT' + }); + }; + const file = await HttpFile.open(URL, { + fetch, + byteLength: DATA.byteLength, + etag: '"manifest-version"', + lastModified: 'Tue, 02 Jan 2024 00:00:00 GMT', + consistency: 'strict' + }); + + expect(requestCount, 'does not probe a complete supplied identity').toBe(0); + await file.read(8, 4); + expect(requestCount, 'performs only the requested data fetch').toBe(1); +}); + +test('HttpFile#strict consistency rejects missing validators before consuming the body', async () => { + let cancelCount = 0; + const filePromise = HttpFile.open(URL, { + consistency: 'strict', + fetch: async (_url, options) => + createCancelableResponse(options, {}, () => { + cancelCount++; + }) + }); + + await expect(filePromise, 'rejects a response without a validator').rejects.toThrow( + /does not provide an ETag or Last-Modified/ + ); + expect(cancelCount, 'cancels the untrusted response body').toBe(1); +}); + +test('HttpFile#best-effort consistency accepts a stable length without validators', async () => { + const file = await HttpFile.open(URL, { + consistency: 'best-effort', + fetch: async (_url, options) => createRangeResponse(options) + }); + + expect(file.getIdentitySnapshot(), 'pins the available identity').toEqual({ + byteLength: DATA.byteLength, + etag: undefined, + lastModified: undefined + }); + expect(Array.from(new Uint8Array(await file.read(1, 2))), 'supports later reads').toEqual([1, 2]); +}); + +test('HttpFile#rejects changed validators before consuming the response body', async () => { + let etag = '"version-1"'; + let cancelCount = 0; + let requestCount = 0; + const file = await HttpFile.open(URL, { + fetch: async (_url, options) => { + requestCount++; + if (requestCount === 1) { + return createRangeResponse(options, {etag}); + } + return createCancelableResponse(options, {etag}, () => { + cancelCount++; + }); + } + }); + etag = '"version-2"'; + + await expect(file.read(4, 2), 'rejects the changed object').rejects.toThrow(/ETag changed/); + expect(cancelCount, 'cancels the changed response body').toBe(1); + expect(file.getTelemetry(), 'does not count an unconsumed response body').toMatchObject({ + requestedBytes: 3, + downloadedBytes: 1, + requestCount: 2, + errorCount: 1 + }); +}); + +test('HttpFile#uses Last-Modified when ETag is unavailable', async () => { + let lastModified = 'Mon, 01 Jan 2024 00:00:00 GMT'; + const file = await HttpFile.open(URL, { + consistency: 'strict', + fetch: async (_url, options) => createRangeResponse(options, {lastModified}) + }); + lastModified = 'Tue, 02 Jan 2024 00:00:00 GMT'; + + await expect(file.read(2, 2), 'rejects a changed Last-Modified value').rejects.toThrow( + /Last-Modified value changed/ + ); +}); + +test('HttpFile#checks Last-Modified when a best-effort response omits the pinned ETag', async () => { + let requestCount = 0; + const file = await HttpFile.open(URL, { + consistency: 'best-effort', + fetch: async (_url, options) => { + requestCount++; + return createRangeResponse( + options, + requestCount === 1 + ? { + etag: '"version-1"', + lastModified: 'Mon, 01 Jan 2024 00:00:00 GMT' + } + : {lastModified: 'Tue, 02 Jan 2024 00:00:00 GMT'} + ); + } + }); + + await expect(file.read(2, 2), 'rejects the changed fallback validator').rejects.toThrow( + /Last-Modified value changed/ + ); +}); + +test('HttpFile#rejects ignored and malformed range responses', async () => { + let ignoredBodyCancelCount = 0; + const ignoredRange = HttpFile.open(URL, { + fetch: async () => + createCancelableRawResponse(200, {}, () => { + ignoredBodyCancelCount++; + }) + }); + await expect(ignoredRange, 'rejects an ignored Range header').rejects.toThrow( + /expected 206, received 200/ + ); + expect(ignoredBodyCancelCount, 'cancels the ignored full response').toBe(1); + + let malformedBodyCancelCount = 0; + const malformedRange = HttpFile.open(URL, { + fetch: async () => + createCancelableRawResponse(206, {'Content-Range': 'invalid'}, () => { + malformedBodyCancelCount++; + }) + }); + await expect(malformedRange, 'rejects malformed Content-Range').rejects.toThrow( + /invalid Content-Range/ + ); + expect(malformedBodyCancelCount, 'cancels the malformed response').toBe(1); + + let mismatchedBodyCancelCount = 0; + const mismatchedFile = await HttpFile.open(URL, { + byteLength: DATA.byteLength, + etag: '"version-1"', + fetch: async () => + createCancelableRawResponse( + 206, + { + 'Content-Range': `bytes 5-6/${DATA.byteLength}`, + ETag: '"version-1"' + }, + () => { + mismatchedBodyCancelCount++; + } + ) + }); + await expect(mismatchedFile.read(4, 2), 'rejects a mismatched Content-Range').rejects.toThrow( + /does not match the requested range/ + ); + expect(mismatchedBodyCancelCount, 'cancels the mismatched response').toBe(1); +}); + +test('HttpFile#rejects short response bodies', async () => { + const file = await HttpFile.open(URL, { + byteLength: DATA.byteLength, + etag: '"version-1"', + fetch: async (_url, options) => { + const {offset, length} = getRequestedRange(options); + return new Response(DATA.slice(offset, offset + length - 1), { + status: 206, + headers: { + 'Content-Range': `bytes ${offset}-${offset + length - 1}/${DATA.byteLength}`, + ETag: '"version-1"' + } + }); + } + }); + + await expect(file.read(4, 3), 'rejects a short body').rejects.toThrow( + /contained 2 bytes; expected 3/ + ); + expect(file.getTelemetry(), 'counts downloaded bytes and the error').toMatchObject({ + requestedBytes: 3, + downloadedBytes: 2, + requestCount: 1, + errorCount: 1 + }); +}); + +test('HttpFile#read aborts an in-flight request and remains observable', async () => { + let markStarted: () => void = () => {}; + const started = new Promise(resolve => { + markStarted = resolve; + }); + const file = await HttpFile.open(URL, { + byteLength: DATA.byteLength, + etag: '"version-1"', + fetch: async (_url, options) => { + markStarted(); + return await new Promise((_resolve, reject) => { + options?.signal?.addEventListener( + 'abort', + () => reject(new DOMException('Request aborted', 'AbortError')), + {once: true} + ); + }); + } + }); + const abortController = new AbortController(); + const readPromise = file.read(0, 4, abortController.signal); + await started; + abortController.abort(); + + await expect(readPromise, 'rejects the cancelled read').rejects.toThrow(/aborted/i); + expect(file.getTelemetry(), 'counts the transport and cancellation once').toMatchObject({ + requestedBytes: 4, + downloadedBytes: 0, + requestCount: 1, + abortCount: 1, + errorCount: 0 + }); +}); + +test('HttpFile#read honors the persistent fetch-options signal after opening', async () => { + let markStarted: () => void = () => {}; + const started = new Promise(resolve => { + markStarted = resolve; + }); + const abortController = new AbortController(); + const file = await HttpFile.open(URL, { + byteLength: DATA.byteLength, + etag: '"version-1"', + fetchOptions: {signal: abortController.signal}, + fetch: async (_url, options) => { + markStarted(); + return await new Promise((_resolve, reject) => { + options?.signal?.addEventListener( + 'abort', + () => reject(new DOMException('Request aborted', 'AbortError')), + {once: true} + ); + }); + } + }); + const readPromise = file.read(0, 4); + await started; + abortController.abort(); + + await expect(readPromise, 'rejects when the persistent signal aborts').rejects.toThrow( + /aborted/i + ); + expect(file.getTelemetry(), 'counts the persistent cancellation').toMatchObject({ + requestedBytes: 4, + downloadedBytes: 0, + requestCount: 1, + abortCount: 1, + errorCount: 0 + }); +}); + +test('HttpFile#concurrent callers cancel identity waits independently', async () => { + let markProbeStarted: () => void = () => {}; + let probeOptions: RequestInit | undefined; + const probeStarted = new Promise(resolve => { + markProbeStarted = resolve; + }); + let releaseProbe: (response: Response) => void = () => {}; + const fetch: HttpFileFetch = async (_url, options) => { + probeOptions = options; + markProbeStarted(); + return await new Promise(resolve => { + releaseProbe = resolve; + }); + }; + const file = new HttpFile(URL, {fetch}); + const firstAbortController = new AbortController(); + const secondAbortController = new AbortController(); + const firstOpen = file.open(firstAbortController.signal); + const secondOpen = file.open(secondAbortController.signal); + + await probeStarted; + firstAbortController.abort(); + await expect(firstOpen, 'cancels only the first caller').rejects.toThrow(/aborted/i); + expect(probeOptions?.signal?.aborted, 'keeps the shared probe alive').toBeFalsy(); + + releaseProbe( + createRangeResponse(probeOptions, { + etag: '"version-1"' + }) + ); + await expect(secondOpen, 'allows the second caller to finish').resolves.toBe(file); + expect(file.getIdentitySnapshot(), 'caches the completed shared probe').toEqual({ + byteLength: DATA.byteLength, + etag: '"version-1"', + lastModified: undefined + }); + expect(file.getTelemetry(), 'counts one probe and one cancelled waiter').toMatchObject({ + requestedBytes: 1, + downloadedBytes: 1, + requestCount: 1, + abortCount: 1, + errorCount: 0 + }); +}); + +test('HttpFile#shared scheduler isolates request contexts', async () => { + const scheduler = new RangeRequestScheduler({batchDelayMs: 0, rangeExpansionBytes: 32}); + const authorizations: (string | null)[] = []; + const fetch: HttpFileFetch = async (_url, options) => { + authorizations.push(new Headers(options?.headers).get('Authorization')); + return createRangeResponse(options, {etag: '"version-1"'}); + }; + const firstFile = await HttpFile.open(URL, { + fetch, + fetchOptions: {headers: {Authorization: 'Bearer first'}}, + byteLength: DATA.byteLength, + etag: '"version-1"', + rangeScheduler: scheduler + }); + const secondFile = await HttpFile.open(URL, { + fetch, + fetchOptions: {headers: {Authorization: 'Bearer second'}}, + byteLength: DATA.byteLength, + etag: '"version-1"', + rangeScheduler: scheduler + }); + + await Promise.all([firstFile.read(0, 2), secondFile.read(4, 2)]); + expect(authorizations.sort(), 'preserves both authentication contexts').toEqual([ + 'Bearer first', + 'Bearer second' + ]); + expect( + getRangeStats(scheduler.stats).transportRanges, + 'does not coalesce requests across isolated files' + ).toBe(2); +}); + +function createRangeResponse( + options?: RequestInit, + identity: {etag?: string; lastModified?: string} = {} +): Response { + const {offset, length} = getRequestedRange(options); + const headers = new Headers({ + 'Content-Range': `bytes ${offset}-${offset + length - 1}/${DATA.byteLength}` + }); + if (identity.etag) { + headers.set('ETag', identity.etag); + } + if (identity.lastModified) { + headers.set('Last-Modified', identity.lastModified); + } + return new Response(DATA.slice(offset, offset + length), {status: 206, headers}); +} + +function createCancelableResponse( + options: RequestInit | undefined, + identity: {etag?: string; lastModified?: string}, + onCancel: () => void +): Response { + const {offset, length} = getRequestedRange(options); + const headers: Record = { + 'Content-Range': `bytes ${offset}-${offset + length - 1}/${DATA.byteLength}` + }; + if (identity.etag) { + headers.ETag = identity.etag; + } + if (identity.lastModified) { + headers['Last-Modified'] = identity.lastModified; + } + return createCancelableRawResponse(206, headers, onCancel); +} + +function createCancelableRawResponse( + status: number, + headers: HeadersInit, + onCancel: () => void +): Response { + const body = new ReadableStream({ + start(controller) { + controller.enqueue(Uint8Array.of(0)); + }, + cancel() { + onCancel(); + } + }); + return new Response(body, {status, headers}); +} + +function getRequestedRange(options?: RequestInit): {offset: number; length: number} { + const range = new Headers(options?.headers).get('Range'); + const match = range?.match(/^bytes=(\d+)-(\d+)$/); + if (!match) { + throw new Error(`Invalid test Range header: ${range}`); + } + const offset = Number(match[1]); + return {offset, length: Number(match[2]) - offset + 1}; +} diff --git a/modules/loader-utils/test/lib/request-utils/range-request-scheduler.node.spec.ts b/modules/loader-utils/test/lib/request-utils/range-request-scheduler.node.spec.ts index 6b2eaafe97..fc4b494034 100644 --- a/modules/loader-utils/test/lib/request-utils/range-request-scheduler.node.spec.ts +++ b/modules/loader-utils/test/lib/request-utils/range-request-scheduler.node.spec.ts @@ -80,7 +80,8 @@ test('RangeRequestScheduler#stats and events describe coalesced ranges', async ( 'counts completed transport requests' ).toBe(1); expect(stats.get('Range Overfetch Bytes').count, 'counts over-fetched gap bytes').toBe(2); - expect(getRangeStats(stats), 'reads typed RangeStats from probe.gl Stats').toEqual({ + const rangeStats = getRangeStats(stats); + expect(rangeStats, 'reads typed RangeStats from probe.gl Stats').toEqual({ logicalRanges: 2, rangeBatches: 1, transportRanges: 1, @@ -89,11 +90,13 @@ test('RangeRequestScheduler#stats and events describe coalesced ranges', async ( requestedBytes: 8, transportBytes: 10, responseBytes: 10, + networkTimeMs: rangeStats.networkTimeMs, overfetchBytes: 2, failedTransportRanges: 0, abortedLogicalRanges: 0, fullResponseFallbacks: 0 }); + expect(rangeStats.networkTimeMs, 'records aggregate transport time').toBeGreaterThanOrEqual(0); expect( events.some(event => event.type === 'batch' && event.logicalRequestCount === 2), 'emits batch event' @@ -142,6 +145,7 @@ test('RangeRequestScheduler#accepts maxGapBytes as rangeExpansionBytes alias', a test('RangeRequestScheduler#fetch sends merged HTTP range and preserves headers', async () => { await withFakeTimers(async () => { const scheduler = new RangeRequestScheduler({batchDelayMs: 0, rangeExpansionBytes: 8}); + const isolationKey = {}; const fetches: { url: string; authorization: string | null; @@ -163,6 +167,7 @@ test('RangeRequestScheduler#fetch sends merged HTTP range and preserves headers' url: 'https://example.com/archive.pmtiles', offset: 10, length: 4, + isolationKey, fetch: fetchRange, fetchOptions: {headers: {Authorization: 'Bearer token'}} }); @@ -170,6 +175,7 @@ test('RangeRequestScheduler#fetch sends merged HTTP range and preserves headers' url: 'https://example.com/archive.pmtiles', offset: 16, length: 4, + isolationKey, fetch: fetchRange, fetchOptions: {headers: {Authorization: 'Bearer token'}} }); @@ -190,6 +196,116 @@ test('RangeRequestScheduler#fetch sends merged HTTP range and preserves headers' ]); }); }); +test('RangeRequestScheduler#fetch isolates request contexts by default', async () => { + await withFakeTimers(async () => { + const scheduler = new RangeRequestScheduler({batchDelayMs: 0, rangeExpansionBytes: 8}); + const authorizations: (string | null)[] = []; + const fetch = async (_url: string, options?: RequestInit) => { + authorizations.push(new Headers(options?.headers).get('Authorization')); + const range = new Headers(options?.headers).get('Range'); + const match = range?.match(/^bytes=(\d+)-(\d+)$/); + if (!match) { + throw new Error('Missing test range'); + } + const offset = Number(match[1]); + const endOffset = Number(match[2]); + return new Response(BYTES.buffer.slice(offset, endOffset + 1), {status: 206}); + }; + const firstRequest = scheduler.fetch({ + url: 'https://example.com/archive.pmtiles', + offset: 10, + length: 4, + fetch, + fetchOptions: {headers: {Authorization: 'Bearer first'}} + }); + const secondRequest = scheduler.fetch({ + url: 'https://example.com/archive.pmtiles', + offset: 16, + length: 4, + fetch, + fetchOptions: {headers: {Authorization: 'Bearer second'}} + }); + + await advanceTimersAndFlush(); + await Promise.all([firstRequest, secondRequest]); + + expect(authorizations.sort(), 'does not share transport contexts without opt-in').toEqual([ + 'Bearer first', + 'Bearer second' + ]); + expect(getRangeStats(scheduler.stats).transportRanges, 'creates two transport requests').toBe( + 2 + ); + }); +}); +test('RangeRequestScheduler#fetch preserves the offset-zero 416 clamp fallback', async () => { + await withFakeTimers(async () => { + const scheduler = new RangeRequestScheduler({batchDelayMs: 0}); + let fetchCount = 0; + const request = scheduler.fetch({ + url: 'https://example.com/archive.pmtiles', + offset: 0, + length: 64, + fetch: async () => { + fetchCount++; + if (fetchCount === 1) { + return new Response(null, { + status: 416, + headers: {'Content-Range': 'bytes */32'} + }); + } + return new Response(BYTES.buffer.slice(0, 32), {status: 206}); + } + }); + + await advanceTimersAndFlush(); + const result = await request; + + expect(result.byteLength, 'returns the server-reported shorter object').toBe(32); + expect(fetchCount, 'retries using the object length').toBe(2); + expect(getRangeStats(scheduler.stats).fullResponseFallbacks, 'records the fallback').toBe(1); + }); +}); +test('RangeRequestScheduler#fetch accepts an HTTP range clamped at end of file', async () => { + await withFakeTimers(async () => { + const scheduler = new RangeRequestScheduler({batchDelayMs: 0}); + const request = scheduler.fetch({ + url: 'https://example.com/archive.pmtiles', + offset: 250, + length: 16, + fetch: async () => + new Response(BYTES.buffer.slice(250), { + status: 206, + headers: {'Content-Range': 'bytes 250-255/256'} + }) + }); + + await advanceTimersAndFlush(); + const result = await request; + expect(result.byteLength, 'returns the bytes available before EOF').toBe(6); + }); +}); +test('RangeRequestScheduler#fetch validates Content-Range against the request and body', async () => { + await withFakeTimers(async () => { + const scheduler = new RangeRequestScheduler({batchDelayMs: 0}); + const request = scheduler.fetch({ + url: 'https://example.com/archive.pmtiles', + offset: 10, + length: 4, + fetch: async () => + new Response(BYTES.buffer.slice(10, 14), { + status: 206, + headers: {'Content-Range': 'bytes 11-14/256'} + }) + }); + const rejection = expect(request, 'rejects a mismatched Content-Range').rejects.toThrow( + /does not match requested range/ + ); + + await advanceTimersAndFlush(); + await rejection; + }); +}); test('RangeRequestScheduler#fetch rejects ignored range responses', async () => { await withFakeTimers(async () => { const scheduler = new RangeRequestScheduler({batchDelayMs: 0}); @@ -289,3 +405,202 @@ test('RangeRequestScheduler#abort before flush rejects one child request', async await rejection; }); }); +test('RangeRequestScheduler#zero-length ranges do not call the transport', async () => { + const scheduler = new RangeRequestScheduler({batchDelayMs: 0}); + let fetchCount = 0; + const result = await scheduler.scheduleRequest({ + sourceId: 'source', + offset: 10, + length: 0, + fetchRange: async () => { + fetchCount++; + return new ArrayBuffer(0); + } + }); + + expect(result.byteLength, 'returns an empty buffer').toBe(0); + expect(fetchCount, 'does not call the transport').toBe(0); + expect(getRangeStats(scheduler.stats).logicalRanges, 'does not queue a logical range').toBe(0); +}); +test('RangeRequestScheduler#rejects a short merged response', async () => { + await withFakeTimers(async () => { + const scheduler = new RangeRequestScheduler({batchDelayMs: 0}); + const request = scheduler.scheduleRequest({ + sourceId: 'source', + offset: 10, + length: 4, + fetchRange: async () => new ArrayBuffer(3) + }); + const rejection = expect(request, 'rejects the short response').rejects.toThrow( + /returned 3 bytes; expected 4/ + ); + + await advanceTimersAndFlush(); + await rejection; + expect(getRangeStats(scheduler.stats).failedTransportRanges, 'counts the failure').toBe(1); + }); +}); +test('RangeRequestScheduler#accepts a short response clamped at a declared end of file', async () => { + await withFakeTimers(async () => { + const scheduler = new RangeRequestScheduler({batchDelayMs: 0}); + const request = scheduler.scheduleRequest({ + sourceId: 'source', + offset: 250, + length: 16, + fetchRange: async () => ({ + arrayBuffer: BYTES.buffer.slice(250), + sourceByteLength: BYTES.byteLength + }) + }); + + await advanceTimersAndFlush(); + const result = await request; + expect(result.byteLength, 'returns the bytes available before EOF').toBe(6); + }); +}); +test('RangeRequestScheduler#rejects a short response not ending at the declared source length', async () => { + await withFakeTimers(async () => { + const scheduler = new RangeRequestScheduler({batchDelayMs: 0}); + const request = scheduler.scheduleRequest({ + sourceId: 'source', + offset: 10, + length: 4, + fetchRange: async () => ({ + arrayBuffer: new ArrayBuffer(3), + sourceByteLength: 256 + }) + }); + const rejection = expect( + request, + 'rejects an incorrectly marked short response' + ).rejects.toThrow(/returned 3 bytes; expected 4/); + + await advanceTimersAndFlush(); + await rejection; + }); +}); +test('RangeRequestScheduler#rejects an oversized nonzero full-response result', async () => { + await withFakeTimers(async () => { + const scheduler = new RangeRequestScheduler({batchDelayMs: 0}); + const request = scheduler.scheduleRequest({ + sourceId: 'source', + offset: 10, + length: 4, + fetchRange: async () => ({ + arrayBuffer: BYTES.buffer.slice(0, 20), + fullResponse: true + }) + }); + const rejection = expect(request, 'rejects an ambiguous full-object response').rejects.toThrow( + /returned 20 bytes; expected 4/ + ); + + await advanceTimersAndFlush(); + await rejection; + }); +}); +test('RangeRequestScheduler#diagnostic callback errors do not affect reads or telemetry', async () => { + await withFakeTimers(async () => { + const scheduler = new RangeRequestScheduler({ + batchDelayMs: 0, + onEvent: () => { + throw new Error('diagnostic failure'); + } + }); + const request = scheduler.scheduleRequest({ + sourceId: 'source', + offset: 10, + length: 4, + fetchRange: async () => BYTES.buffer.slice(10, 14) + }); + + await advanceTimersAndFlush(); + const result = await request; + expect(result.byteLength, 'still resolves the transport result').toBe(4); + + const rangeStats = getRangeStats(scheduler.stats); + expect(rangeStats.completedTransportRanges, 'records one successful transport').toBe(1); + expect(rangeStats.failedTransportRanges, 'does not record a diagnostic failure').toBe(0); + }); +}); +test('RangeRequestScheduler#counts an in-flight abort', async () => { + await withFakeTimers(async () => { + const scheduler = new RangeRequestScheduler({batchDelayMs: 0}); + const abortController = new AbortController(); + const request = scheduler.scheduleRequest({ + sourceId: 'source', + offset: 0, + length: 4, + signal: abortController.signal, + fetchRange: async (_offset, _length, signal) => + await new Promise((_resolve, reject) => { + signal?.addEventListener( + 'abort', + () => reject(new DOMException('Request aborted', 'AbortError')), + {once: true} + ); + }) + }); + + await advanceTimersAndFlush(); + abortController.abort(); + await expect(request, 'rejects the in-flight request').rejects.toThrow(/aborted/i); + + const rangeStats = getRangeStats(scheduler.stats); + expect(rangeStats.abortedLogicalRanges, 'counts the aborted logical request').toBe(1); + expect(rangeStats.failedTransportRanges, 'counts the failed transport').toBe(1); + }); +}); +test('RangeRequestScheduler#settles coalesced aborts without poisoning active siblings', async () => { + await withFakeTimers(async () => { + const scheduler = new RangeRequestScheduler({batchDelayMs: 0}); + const firstAbortController = new AbortController(); + const secondAbortController = new AbortController(); + let transportAbortCount = 0; + const fetchRange = async (_offset: number, _length: number, signal?: AbortSignal) => + await new Promise((_resolve, reject) => { + signal?.addEventListener( + 'abort', + () => { + transportAbortCount++; + reject(new DOMException('Request aborted', 'AbortError')); + }, + {once: true} + ); + }); + const firstRequest = scheduler.scheduleRequest({ + sourceId: 'source', + offset: 0, + length: 4, + signal: firstAbortController.signal, + fetchRange + }); + const secondRequest = scheduler.scheduleRequest({ + sourceId: 'source', + offset: 4, + length: 4, + signal: secondAbortController.signal, + fetchRange + }); + + await advanceTimersAndFlush(); + const firstRejection = expect(firstRequest, 'rejects the first child promptly').rejects.toThrow( + /aborted/i + ); + firstAbortController.abort(); + await firstRejection; + expect(transportAbortCount, 'keeps transport alive for the active sibling').toBe(0); + expect(getRangeStats(scheduler.stats).abortedLogicalRanges, 'tracks the first abort once').toBe( + 1 + ); + + const secondRejection = expect( + secondRequest, + 'rejects the final child promptly' + ).rejects.toThrow(/aborted/i); + secondAbortController.abort(); + await secondRejection; + expect(transportAbortCount, 'aborts transport after all children cancel').toBe(1); + expect(getRangeStats(scheduler.stats).abortedLogicalRanges, 'tracks each abort once').toBe(2); + }); +}); diff --git a/modules/parquet/package.json b/modules/parquet/package.json index 57f9ae2c13..8c8bdef12b 100644 --- a/modules/parquet/package.json +++ b/modules/parquet/package.json @@ -50,7 +50,12 @@ "./parquet-worker-node.cjs": { "import": "./dist/parquet-worker-node.cjs", "require": "./dist/parquet-worker-node.cjs" - } + }, + "./wasm": { + "types": "./dist/wasm.d.ts", + "import": "./dist/wasm.js" + }, + "./parquet_wasm_bg.wasm": "./dist/parquet_wasm_bg.wasm" }, "sideEffects": false, "files": [ @@ -92,7 +97,7 @@ "lz4js": "^0.2.0", "node-int64": "^0.4.0", "object-stream": "0.0.1", - "parquet-wasm": "0.7.1", + "parquet-wasm": "0.7.2", "snappyjs": "^0.6.0", "thrift": "^0.24.0", "util": "^0.12.5", diff --git a/modules/parquet/src/bundled.ts b/modules/parquet/src/bundled.ts index b4bf1ad994..396005eff7 100644 --- a/modules/parquet/src/bundled.ts +++ b/modules/parquet/src/bundled.ts @@ -18,3 +18,7 @@ export { type ParquetBatchMetadata, type ParquetSourceBatch } from './parquet-source-loader'; +export { + PARQUET_SOURCE_CAPABILITIES, + type ParquetSourceCapabilities +} from './parquet-source-capabilities'; diff --git a/modules/parquet/src/index.ts b/modules/parquet/src/index.ts index 02e1825d89..5c74b02cb6 100644 --- a/modules/parquet/src/index.ts +++ b/modules/parquet/src/index.ts @@ -30,6 +30,10 @@ export type { ParquetTelemetry, ParquetTelemetryEvent } from './parquet-source-types'; +export { + PARQUET_SOURCE_CAPABILITIES, + type ParquetSourceCapabilities +} from './parquet-source-capabilities'; export {ParquetWriter} from './parquet-writer'; export type {ParquetJSWriterOptions} from './parquet-js-writer'; diff --git a/modules/parquet/src/lib/utils/make-stream-iterator.ts b/modules/parquet/src/lib/utils/make-stream-iterator.ts index 4c123567f0..8dde0d36ac 100644 --- a/modules/parquet/src/lib/utils/make-stream-iterator.ts +++ b/modules/parquet/src/lib/utils/make-stream-iterator.ts @@ -7,6 +7,8 @@ import {isBrowser} from '@loaders.gl/loader-utils'; export type StreamIteratorOptions = { _streamReadAhead?: boolean; + /** Cancels the active stream read and rejects iteration with the signal reason. */ + signal?: AbortSignal; }; /** @@ -17,11 +19,16 @@ export function makeStreamIterator( stream: ReadableStream | Readable, options?: StreamIteratorOptions ): AsyncIterable { - return isBrowser + return isBrowser || isReadableStream(stream) ? makeBrowserStreamIterator(stream as ReadableStream, options) : makeNodeStreamIterator(stream as Readable, options); } +/** Detects WHATWG streams in Node.js as well as in browsers. */ +function isReadableStream(stream: ReadableStream | Readable): stream is ReadableStream { + return typeof (stream as ReadableStream).getReader === 'function'; +} + /** * Returns an async iterable that reads from a DOM (browser) stream * @param stream stream to iterate from @@ -41,6 +48,11 @@ async function* makeBrowserStreamIterator( // In the browser, we first need to get a lock on the stream const reader = stream.getReader(); + const signal = options?.signal; + const abortListener = () => { + // Canceling the reader settles a pending read. The loop then throws the signal reason. + void reader.cancel(signal?.reason).catch(() => {}); + }; let nextBatchPromise: Promise<{done?: boolean; value?: T}> | undefined; let streamFinished = false; @@ -48,6 +60,9 @@ async function* makeBrowserStreamIterator( let iteratorFailed = false; try { + throwIfAborted(signal); + signal?.addEventListener('abort', abortListener, {once: true}); + // eslint-disable-next-line no-constant-condition while (true) { const currentBatchPromise = nextBatchPromise || reader.read(); @@ -60,6 +75,7 @@ async function* makeBrowserStreamIterator( streamReadFailed = true; throw error; } + throwIfAborted(signal); const {done, value} = batch; // Exit if we're done if (done) { @@ -81,6 +97,7 @@ async function* makeBrowserStreamIterator( iteratorFailed = true; throw error; } finally { + signal?.removeEventListener('abort', abortListener); let cleanupError: unknown; if (!streamFinished && !streamReadFailed) { @@ -113,7 +130,70 @@ async function* makeNodeStreamIterator( stream: Readable, options?: StreamIteratorOptions ): AsyncIterable { - // Hacky test for node version to ensure we don't call bad polyfills - // NODE 10+: stream is an asyncIterator - yield* stream; + // Node streams and modern Node.js Web streams both expose an async iterator. + const iterator = stream[Symbol.asyncIterator](); + const signal = options?.signal; + let abortListener: (() => void) | undefined; + const abortPromise = signal + ? new Promise((_resolve, reject) => { + abortListener = () => reject(getAbortReason(signal)); + signal.addEventListener('abort', abortListener, {once: true}); + }) + : null; + let iteratorFailed = false; + let streamFinished = false; + + try { + throwIfAborted(signal); + // eslint-disable-next-line no-constant-condition + while (true) { + const nextPromise = iterator.next(); + if (abortPromise) { + // Observe a late failure if cancellation wins the race. + void nextPromise.catch(() => {}); + } + const result = abortPromise + ? await Promise.race([nextPromise, abortPromise]) + : await nextPromise; + throwIfAborted(signal); + if (result.done) { + streamFinished = true; + return; + } + yield result.value; + } + } catch (error) { + iteratorFailed = true; + throw error; + } finally { + if (abortListener) { + signal?.removeEventListener('abort', abortListener); + } + if (!streamFinished) { + try { + await iterator.return?.(); + } catch (error) { + if (!iteratorFailed) { + throw error; + } + } + } + } +} + +/** Returns the standard AbortSignal reason, including a fallback for older runtimes. */ +function getAbortReason(signal: AbortSignal): unknown { + if (signal.reason !== undefined) { + return signal.reason; + } + const error = new Error('The operation was aborted'); + error.name = 'AbortError'; + return error; +} + +/** Throws at a cancellation checkpoint. */ +function throwIfAborted(signal?: AbortSignal): void { + if (signal?.aborted) { + throw getAbortReason(signal); + } } diff --git a/modules/parquet/src/parquet-source-capabilities.ts b/modules/parquet/src/parquet-source-capabilities.ts new file mode 100644 index 0000000000..cee90e915d --- /dev/null +++ b/modules/parquet/src/parquet-source-capabilities.ts @@ -0,0 +1,47 @@ +// loaders.gl +// SPDX-License-Identifier: MIT +// Copyright (c) vis.gl contributors + +/** Feature support advertised by a {@link ParquetSource}. */ +export type ParquetSourceCapabilities = Readonly<{ + /** Schema and footer metadata are cached for the lifetime of the source. */ + supportsCachedMetadata: boolean; + /** Reads can select explicit Parquet row groups. */ + supportsRowGroupSelection: boolean; + /** Reads can project explicit Parquet column paths. */ + supportsColumnProjection: boolean; + /** Every emitted batch identifies its source row group and row offsets. */ + supportsBatchProvenance: boolean; + /** Read cancellation is propagated cooperatively through the active stream. */ + supportsCooperativeReadCancellation: boolean; + /** The package includes a local parquet-wasm binary for self-hosted delivery. */ + supportsLocalWasmAsset: boolean; + /** Column min/max/null statistics are exposed in public metadata. */ + supportsColumnStatistics: boolean; + /** Callers can supply the random-access transport used by the source. */ + supportsCustomRangeTransport: boolean; + /** Range requests validate that the source object version remains unchanged. */ + supportsObjectVersionValidation: boolean; + /** Requested bytes, downloaded bytes, request counts, and network time are reported. */ + supportsNetworkTelemetry: boolean; + /** Decoder time is reported separately from network and Arrow conversion time. */ + supportsDecodeTelemetry: boolean; + /** Parquet decoding can run in a worker and transfer Arrow output. */ + supportsWorkerDecoding: boolean; +}>; + +/** Capabilities of the current range-backed TypeScript {@link ParquetSource}. */ +export const PARQUET_SOURCE_CAPABILITIES: ParquetSourceCapabilities = Object.freeze({ + supportsCachedMetadata: true, + supportsRowGroupSelection: true, + supportsColumnProjection: true, + supportsBatchProvenance: true, + supportsCooperativeReadCancellation: true, + supportsLocalWasmAsset: true, + supportsColumnStatistics: true, + supportsCustomRangeTransport: true, + supportsObjectVersionValidation: true, + supportsNetworkTelemetry: true, + supportsDecodeTelemetry: true, + supportsWorkerDecoding: false +}); diff --git a/modules/parquet/src/parquet-source-loader.ts b/modules/parquet/src/parquet-source-loader.ts index 0dd0b3a318..72a3dcb3de 100644 --- a/modules/parquet/src/parquet-source-loader.ts +++ b/modules/parquet/src/parquet-source-loader.ts @@ -9,6 +9,10 @@ import {convertTable} from '@loaders.gl/schema-utils'; import {getSchemaFromParquetReader} from './lib/parsers/get-parquet-schema'; import {ParquetRangeFile} from './lib/sources/parquet-range-file'; +import { + PARQUET_SOURCE_CAPABILITIES, + type ParquetSourceCapabilities +} from './parquet-source-capabilities'; import {ParquetSourceLoader as ParquetSourceLoaderMetadata} from './parquet-source-loader-types'; import type { ParquetBatch, @@ -24,6 +28,10 @@ import type { ParquetTelemetry, ParquetTelemetryEvent } from './parquet-source-types'; +export { + PARQUET_SOURCE_CAPABILITIES, + type ParquetSourceCapabilities +} from './parquet-source-capabilities'; import {preloadCompressions} from './parquetjs/compression'; import { CompressionCodec, @@ -122,6 +130,8 @@ export {ParquetSourceLoaderWithParser as ParquetSourceLoader}; /** Reusable Parquet source that caches footer/schema state and selectively reads byte ranges. */ export class ParquetSource extends DataSource { + /** Immutable feature support for the current range-backed source implementation. */ + readonly capabilities: ParquetSourceCapabilities = PARQUET_SOURCE_CAPABILITIES; /** Shared initialization for this source instance. */ private initializationPromise: Promise | null = null; /** File allocated during source initialization, including while it is opening. */ @@ -164,7 +174,7 @@ export class ParquetSource extends DataSource void; } { const abortController = new AbortController(); - const abortRead = (): void => abortController.abort(); + const abortRead = (): void => abortController.abort(signal?.reason); if (signal?.aborted) { abortRead(); } else { @@ -793,11 +803,25 @@ function throwIfAborted(signal: AbortSignal): void { if (!signal.aborted) { return; } + if (signal.reason !== undefined) { + throw signal.reason; + } const error = new Error('Parquet read aborted'); error.name = 'AbortError'; throw error; } +/** Recursively freezes the plain schema tree cached by a source. */ +function deepFreeze(value: T): T { + if (!value || typeof value !== 'object' || Object.isFrozen(value)) { + return value; + } + for (const nestedValue of Object.values(value)) { + deepFreeze(nestedValue); + } + return Object.freeze(value); +} + /** Converts footer key/value pairs into an object. */ function getKeyValueMetadata(fileMetadata: FileMetaData): Record { const result: Record = {}; diff --git a/modules/parquet/src/unbundled.ts b/modules/parquet/src/unbundled.ts index bb98e580a6..773f30e244 100644 --- a/modules/parquet/src/unbundled.ts +++ b/modules/parquet/src/unbundled.ts @@ -18,3 +18,7 @@ export { type ParquetBatchMetadata, type ParquetSourceBatch } from './parquet-source-loader'; +export { + PARQUET_SOURCE_CAPABILITIES, + type ParquetSourceCapabilities +} from './parquet-source-capabilities'; diff --git a/modules/parquet/src/wasm.ts b/modules/parquet/src/wasm.ts new file mode 100644 index 0000000000..708a772fdc --- /dev/null +++ b/modules/parquet/src/wasm.ts @@ -0,0 +1,6 @@ +// loaders.gl +// SPDX-License-Identifier: MIT +// Copyright (c) vis.gl contributors + +/** Package-local parquet-wasm binary URL for ESM applications and bundlers. */ +export const PARQUET_WASM_URL = new URL('./parquet_wasm_bg.wasm', import.meta.url); diff --git a/modules/parquet/test/index.ts b/modules/parquet/test/index.ts index fa4022e9e2..40ee596829 100644 --- a/modules/parquet/test/index.ts +++ b/modules/parquet/test/index.ts @@ -21,6 +21,7 @@ import './parquetjs/reader.spec'; import './parquet-arrow-loader.spec'; import './parquet-arrow-writer.spec'; import './parquet-source-loader.spec'; +import './parquet-source-capabilities.spec'; import './parquet-loader.spec'; import './geoparquet-loader.spec'; diff --git a/modules/parquet/test/make-stream-iterator.spec.ts b/modules/parquet/test/make-stream-iterator.spec.ts index 5e660d4472..4e347f2222 100644 --- a/modules/parquet/test/make-stream-iterator.spec.ts +++ b/modules/parquet/test/make-stream-iterator.spec.ts @@ -71,6 +71,28 @@ test('Parquet makeStreamIterator#unlocks after natural completion', async t => { t.end(); }); +test('Parquet makeStreamIterator#aborts a pending read and releases the stream', async t => { + const abortController = new AbortController(); + const abortReason = new Error('Stop reading Parquet data'); + let cancellationCount = 0; + const stream = new ReadableStream({ + cancel() { + cancellationCount++; + } + }); + + const valuesPromise = collectValues( + makeStreamIterator(stream, {signal: abortController.signal}) + ); + await Promise.resolve(); + abortController.abort(abortReason); + + await t.rejects(valuesPromise, abortReason, 'rejects with the AbortSignal reason'); + t.equal(cancellationCount, 1, 'cancels the pending stream'); + t.notOk(stream.locked, 'releases the reader or async-iterator lock'); + t.end(); +}); + async function collectValues(values: AsyncIterable): Promise { const collectedValues: T[] = []; for await (const value of values) { diff --git a/modules/parquet/test/parquet-arrow-writer.spec.ts b/modules/parquet/test/parquet-arrow-writer.spec.ts index 8c17783a49..3f646e3419 100644 --- a/modules/parquet/test/parquet-arrow-writer.spec.ts +++ b/modules/parquet/test/parquet-arrow-writer.spec.ts @@ -23,12 +23,16 @@ test('ParquetWriter#writer objects', (t) => { test('ParquetSource#public exports', (t) => { t.ok(parquet.ParquetSourceLoader, 'root exports lightweight ParquetSourceLoader metadata'); t.notOk('ParquetSource' in parquet, 'root does not export runtime ParquetSource'); + t.ok(parquet.PARQUET_SOURCE_CAPABILITIES, 'root exports source capabilities'); t.ok(parquetSource.ParquetSourceLoader, 'source subpath exports runtime loader'); t.ok(parquetSource.ParquetSource, 'source subpath exports runtime source'); + t.ok(parquetSource.PARQUET_SOURCE_CAPABILITIES, 'source subpath exports capabilities'); t.ok(bundledParquet.ParquetSourceLoader, 'bundled entry point exports ParquetSourceLoader'); t.ok(bundledParquet.ParquetSource, 'bundled entry point exports ParquetSource'); + t.ok(bundledParquet.PARQUET_SOURCE_CAPABILITIES, 'bundled exports source capabilities'); t.ok(unbundledParquet.ParquetSourceLoader, 'unbundled entry point exports ParquetSourceLoader'); t.ok(unbundledParquet.ParquetSource, 'unbundled entry point exports ParquetSource'); + t.ok(unbundledParquet.PARQUET_SOURCE_CAPABILITIES, 'unbundled exports source capabilities'); t.end(); }); diff --git a/modules/parquet/test/parquet-source-capabilities.spec.ts b/modules/parquet/test/parquet-source-capabilities.spec.ts new file mode 100644 index 0000000000..c7f59c4267 --- /dev/null +++ b/modules/parquet/test/parquet-source-capabilities.spec.ts @@ -0,0 +1,35 @@ +// loaders.gl +// SPDX-License-Identifier: MIT +// Copyright (c) vis.gl contributors + +import test from 'tape-promise/tape'; + +import { + PARQUET_SOURCE_CAPABILITIES, + type ParquetSourceCapabilities +} from '../src/parquet-source-capabilities'; + +test('ParquetSourceCapabilities#advertises implemented and deferred features', t => { + const expectedCapabilities: ParquetSourceCapabilities = { + supportsCachedMetadata: true, + supportsRowGroupSelection: true, + supportsColumnProjection: true, + supportsBatchProvenance: true, + supportsCooperativeReadCancellation: true, + supportsLocalWasmAsset: true, + supportsColumnStatistics: true, + supportsCustomRangeTransport: true, + supportsObjectVersionValidation: true, + supportsNetworkTelemetry: true, + supportsDecodeTelemetry: true, + supportsWorkerDecoding: false + }; + + t.ok(Object.isFrozen(PARQUET_SOURCE_CAPABILITIES), 'freezes the shared capability descriptor'); + t.deepEqual( + PARQUET_SOURCE_CAPABILITIES, + expectedCapabilities, + 'distinguishes implemented source features from deferred backend features' + ); + t.end(); +}); diff --git a/modules/parquet/test/parquet-source-loader.spec.ts b/modules/parquet/test/parquet-source-loader.spec.ts index 0ffea4c2d9..ec7896d2d8 100644 --- a/modules/parquet/test/parquet-source-loader.spec.ts +++ b/modules/parquet/test/parquet-source-loader.spec.ts @@ -7,6 +7,7 @@ import test from 'tape-promise/tape'; import {createDataSource, encode, fetchFile, load} from '@loaders.gl/core'; import type {ObjectRowTable} from '@loaders.gl/schema'; import { + PARQUET_SOURCE_CAPABILITIES, type ParquetBatch, ParquetJSWriter, ParquetSourceLoader, @@ -59,6 +60,8 @@ test('ParquetSourceLoader#Blob metadata and schema are cached', async (t) => { const source = (await load(new Blob([fixture]), ParquetSourceLoader)) as ParquetSource; t.ok(source instanceof ParquetSource, 'root metadata loader preloads the runtime source'); + t.equal(source.capabilities, PARQUET_SOURCE_CAPABILITIES, 'advertises immutable capabilities'); + t.ok(Object.isFrozen(source.getTelemetry()), 'returns frozen telemetry snapshots'); const metadata = await source.getMetadata(); const schema = await source.getSchema(); @@ -79,6 +82,9 @@ test('ParquetSourceLoader#Blob metadata and schema are cached', async (t) => { t.equal(await source.getMetadata(), metadata, 'returns cached metadata object'); t.equal(await source.getSchema(), schema, 'returns cached schema object'); t.ok(Object.isFrozen(metadata), 'freezes the cached metadata object'); + t.ok(Object.isFrozen(metadata.schema), 'freezes the cached schema'); + t.ok(Object.isFrozen(metadata.schema.fields), 'freezes cached schema fields'); + t.ok(Object.isFrozen(metadata.schema.fields[0]), 'freezes each cached schema field'); t.ok(Object.isFrozen(metadata.rowGroups), 'freezes the cached row-group list'); t.ok(Object.isFrozen(metadata.rowGroups[0].columns), 'freezes cached column metadata'); @@ -206,6 +212,7 @@ test('ParquetSource#read selects row groups and columns with exact provenance', 'projects the batch schema' ); t.notOk(batches[0].data.getChild('ignored_payload'), 'does not materialize ignored columns'); + t.ok(Object.isFrozen(batches[0].metadata), 'freezes batch provenance'); const selectedRanges = getColumnRanges(metadata, 1, ['x', 'source_id']); const dataRequests = requests.slice(metadataRequestCount); @@ -220,6 +227,22 @@ test('ParquetSource#read selects row groups and columns with exact provenance', t.end(); }); +test('ParquetSource#read preserves the caller AbortSignal reason', async (t) => { + const fixture = await createSelectiveFixture(); + const source = new ParquetSource(new Blob([fixture]), {}); + const abortController = new AbortController(); + const abortReason = new Error('Query superseded'); + const iterator = source.read({batchSize: 1, signal: abortController.signal})[Symbol.asyncIterator](); + + const firstResult = await iterator.next(); + t.notOk(firstResult.done, 'emits a batch before cancellation'); + abortController.abort(abortReason); + await t.rejects(iterator.next(), abortReason, 'rejects with the caller AbortSignal reason'); + + await source.close(); + t.end(); +}); + test('ParquetSource#read cancels outstanding ranges when iteration ends early', async (t) => { const fixture = await createSelectiveFixture(); let markBlockedRequestStarted: () => void = () => {}; diff --git a/modules/parquet/test/parquet-wasm-assets.node.spec.ts b/modules/parquet/test/parquet-wasm-assets.node.spec.ts new file mode 100644 index 0000000000..aade985c2f --- /dev/null +++ b/modules/parquet/test/parquet-wasm-assets.node.spec.ts @@ -0,0 +1,43 @@ +// loaders.gl +// SPDX-License-Identifier: MIT +// Copyright (c) vis.gl contributors + +import {readFileSync} from 'node:fs'; +import test from 'tape-promise/tape'; + +import {PARQUET_WASM_URL} from '@loaders.gl/parquet/wasm'; + +type ParquetPackage = { + exports: Record>; +}; + +test('Parquet WASM assets#exports a package-local URL', t => { + t.ok(PARQUET_WASM_URL instanceof URL, 'exports a URL'); + t.ok( + PARQUET_WASM_URL.pathname.endsWith('/parquet_wasm_bg.wasm'), + 'resolves the packaged WASM filename next to the ESM module' + ); + t.end(); +}); + +test('Parquet WASM assets#package export map', t => { + const packageJson = JSON.parse( + readFileSync(new URL('../package.json', import.meta.url), 'utf8') + ) as ParquetPackage; + const wasmModuleExport = packageJson.exports['./wasm']; + + t.deepEqual( + wasmModuleExport, + { + types: './dist/wasm.d.ts', + import: './dist/wasm.js' + }, + 'exposes the URL helper as an import-only subpath' + ); + t.equal( + packageJson.exports['./parquet_wasm_bg.wasm'], + './dist/parquet_wasm_bg.wasm', + 'exposes the raw packaged WASM asset' + ); + t.end(); +}); diff --git a/modules/pmtiles/src/lib/range-request-source.ts b/modules/pmtiles/src/lib/range-request-source.ts index b4e1049f31..ea0a84ee28 100644 --- a/modules/pmtiles/src/lib/range-request-source.ts +++ b/modules/pmtiles/src/lib/range-request-source.ts @@ -19,6 +19,7 @@ export class RangeRequestSource implements Source { readonly url: string; private readonly fetch: (url: string, options?: RequestInit) => Promise; private readonly scheduler: RangeRequestScheduler; + private readonly requestContext = {}; /** Creates a PMTiles package Source backed by scheduled HTTP range requests. */ constructor(url: string, options: RangeRequestSourceOptions = {}) { @@ -47,6 +48,7 @@ export class RangeRequestSource implements Source { offset, length, signal, + isolationKey: this.requestContext, fetch: this.fetch }); diff --git a/modules/pmtiles/test/index.ts b/modules/pmtiles/test/index.ts index 10fc82575b..b493275eb0 100644 --- a/modules/pmtiles/test/index.ts +++ b/modules/pmtiles/test/index.ts @@ -3,5 +3,6 @@ // Copyright (c) vis.gl contributors import './pmtiles-source.spec'; +import './range-request-source.spec'; import './pmtiles-loader.spec'; diff --git a/modules/pmtiles/test/range-request-source.spec.ts b/modules/pmtiles/test/range-request-source.spec.ts new file mode 100644 index 0000000000..c62e73662b --- /dev/null +++ b/modules/pmtiles/test/range-request-source.spec.ts @@ -0,0 +1,74 @@ +// loaders.gl +// SPDX-License-Identifier: MIT +// Copyright (c) vis.gl contributors + +import test from 'tape-promise/tape'; +import {RangeRequestSource} from '../src/lib/range-request-source'; + +const BYTES = Uint8Array.from({length: 64}, (_, index) => index); +const URL = 'https://example.com/archive.pmtiles'; + +test('RangeRequestSource coalesces sibling reads from one source', async t => { + const requestedRanges: string[] = []; + const source = new RangeRequestSource(URL, { + batchDelayMs: 0, + rangeExpansionBytes: 8, + fetch: async (_url, options) => { + const range = new Headers(options?.headers).get('Range'); + requestedRanges.push(range || ''); + return createRangeResponse(range); + } + }); + + const [first, second] = await Promise.all([source.getBytes(10, 4), source.getBytes(16, 4)]); + + t.deepEqual(requestedRanges, ['bytes=10-19'], 'uses one merged HTTP request'); + t.deepEqual(Array.from(new Uint8Array(first.data)), [10, 11, 12, 13], 'returns first slice'); + t.deepEqual(Array.from(new Uint8Array(second.data)), [16, 17, 18, 19], 'returns second slice'); + t.end(); +}); + +test('RangeRequestSource keeps distinct source contexts isolated', async t => { + const firstRanges: string[] = []; + const secondRanges: string[] = []; + const firstSource = new RangeRequestSource(URL, { + batchDelayMs: 0, + fetch: makeFetch(firstRanges) + }); + const secondSource = new RangeRequestSource(URL, { + batchDelayMs: 0, + fetch: makeFetch(secondRanges) + }); + + const [first, second] = await Promise.all([ + firstSource.getBytes(2, 2), + secondSource.getBytes(6, 2) + ]); + + t.deepEqual(firstRanges, ['bytes=2-3'], 'uses the first source transport'); + t.deepEqual(secondRanges, ['bytes=6-7'], 'uses the second source transport'); + t.deepEqual(Array.from(new Uint8Array(first.data)), [2, 3], 'returns first source bytes'); + t.deepEqual(Array.from(new Uint8Array(second.data)), [6, 7], 'returns second source bytes'); + t.end(); +}); + +function makeFetch(requestedRanges: string[]) { + return async (_url: string, options?: RequestInit): Promise => { + const range = new Headers(options?.headers).get('Range'); + requestedRanges.push(range || ''); + return createRangeResponse(range); + }; +} + +function createRangeResponse(range: string | null): Response { + const match = range?.match(/^bytes=(\d+)-(\d+)$/); + if (!match) { + throw new Error(`Invalid test Range header: ${range}`); + } + const offset = Number(match[1]); + const endOffset = Number(match[2]); + return new Response(BYTES.slice(offset, endOffset + 1), { + status: 206, + headers: {'Content-Range': `bytes ${offset}-${endOffset}/${BYTES.byteLength}`} + }); +} diff --git a/yarn.lock b/yarn.lock index e6ad034558..2a7df88543 100644 --- a/yarn.lock +++ b/yarn.lock @@ -5715,7 +5715,7 @@ __metadata: lz4js: "npm:^0.2.0" node-int64: "npm:^0.4.0" object-stream: "npm:0.0.1" - parquet-wasm: "npm:0.7.1" + parquet-wasm: "npm:0.7.2" snappyjs: "npm:^0.6.0" thrift: "npm:^0.24.0" util: "npm:^0.12.5" @@ -27729,10 +27729,10 @@ __metadata: languageName: node linkType: hard -"parquet-wasm@npm:0.7.1": - version: 0.7.1 - resolution: "parquet-wasm@npm:0.7.1" - checksum: 10c0/05bf2450455bc1a43acfbe0195f5e4e9ca4569af2ec5a2edf301e41f2c63b9551e1ae25ad2e89e7a12ddd1984a7c771f1d3609a6d01989103e795f44fd10d5af +"parquet-wasm@npm:0.7.2": + version: 0.7.2 + resolution: "parquet-wasm@npm:0.7.2" + checksum: 10c0/cfdf72c542b7e6768e6b38e386c0ab7d13bd92c9d8c45da5619409846654107c279343177bb3dba8c7fcb40c3f7be09dc154b7d175479a669b599b8bce3210b7 languageName: node linkType: hard