diff --git a/fizarrita/README.md b/fizarrita/README.md index cc02b1f..36b3328 100644 --- a/fizarrita/README.md +++ b/fizarrita/README.md @@ -141,12 +141,13 @@ const read = getWorker(arr, [zarr.slice(0, 256), zarr.slice(0, 256)], { controller.abort() ``` -When the signal fires, the signal is forwarded to every `store.get` call, so -stores that honour it (e.g. `FetchStore`, whose options are a `RequestInit`) -cancel their network requests; chunk tasks still queued on the pool are -dropped rather than started; and the returned promise rejects with the +The signal is passed to each chunk `store.get` call the read makes, so stores +that honour it (e.g. `FetchStore`, whose options are a `RequestInit`) cancel +their network requests when it fires. Chunk tasks still queued on the pool are +dropped rather than started, and the returned promise rejects with the signal's reason. A decode already running on a worker is not interrupted — -its result is discarded. +its result is discarded. The shared metadata read and chunk-shape probe are +the one exception — they run without it; see [Chunk caching](#chunk-caching). If `opts` carries its own store-level `signal`, the two are combined: when either fires, fetches abort, still-queued tasks are dropped, and the promise @@ -178,6 +179,16 @@ Cache keys use the format `store_N:/array/path:c/0/1/2`. A `WeakMap`-based store ID ensures keys are unique across store instances, so a single cache can safely be shared across multiple arrays and stores. +The array metadata read and the chunk-shape probe are memoised per +(store, array path) — both are immutable for the lifetime of an array — so +only the first `getWorker` call on an array touches the store for them. A +repeat read served entirely from a warm cache performs zero store requests. +Concurrent calls on a cold array share one resolution. Store options in `opts` +(headers, credentials, …) reach those reads too, with one exception: an +`AbortSignal` governs only the calling read's wait, never the shared +resolution — aborting one caller rejects it promptly without failing the others +that joined it, and the result still lands for the next read. + ### LRU / bounded caches For bounded memory, pass any LRU cache that implements the same `get`/`set` diff --git a/fizarrita/src/get-worker.ts b/fizarrita/src/get-worker.ts index ca6042f..1762ec3 100644 --- a/fizarrita/src/get-worker.ts +++ b/fizarrita/src/get-worker.ts @@ -151,6 +151,15 @@ export interface ArrayMetadata { fillValue: Scalar | null } +/** + * Read a zarr array's metadata, trying v3 (`zarr.json`) then v2 (`.zarray`). + * + * `storeOpts` is forwarded to every `store.get` this makes, so an AbortSignal, + * auth header, or any other per-request option governs the metadata reads on + * the same terms as the chunk reads that follow them — a signal that aborts + * the chunk fetches but silently leaves the `zarr.json` read running would be + * a surprising asymmetry. + */ export async function readArrayMetadata< D extends DataType, Store extends Readable, @@ -575,8 +584,10 @@ export function inferChunkShape( * candidate's chunks are too large (the real grid has more chunks in that * dimension) and should be rejected. * - * Returns true if the candidate is valid (probe returned 404/empty), - * false if invalid (probe returned data, meaning chunks are too coarse). + * `valid` is true if the candidate holds (probe returned 404/empty), false if + * invalid (probe returned data, meaning chunks are too coarse). `conclusive` + * is false when the answer came from a swallowed fetch error rather than from + * a completed probe — see {@link ChunkShapeProbe}. */ async function validateCandidateChunkShape< D extends DataType, @@ -587,7 +598,7 @@ async function validateCandidateChunkShape< candidate: number[], storeOpts?: Parameters[1], signal?: AbortSignal, -): Promise { +): Promise<{ valid: boolean; conclusive: boolean }> { const ndim = candidate.length // Compute grid dimensions and find the dimension with the smallest extent > 1 @@ -607,8 +618,10 @@ async function validateCandidateChunkShape< } if (probeDim === -1) { - // All dimensions have only 1 chunk — can't validate, assume correct - return true + // All dimensions have only 1 chunk — can't validate, assume correct. + // Conclusive: this answer is a property of the grid, not of a failed + // request, so it will be the same on every retry. + return { valid: true, conclusive: true } } // Probe one-past-the-end: if the store has a chunk at this coordinate, @@ -621,12 +634,14 @@ async function validateCandidateChunkShape< try { const probeBytes = await arr.store.get(probePath, storeOpts) // If data returned, there's a chunk beyond our expected grid → reject - return !probeBytes + return { valid: !probeBytes, conclusive: true } } catch (error) { // A caller abort is not a probe outcome — the whole read is over. if (signal?.aborted) throw error - // Fetch error (404, network error) → no chunk there → accept - return true + // Fetch error (404, network error) → no chunk there → accept. The two are + // indistinguishable here, so the acceptance is a guess made under an + // error and must not be memoised as settled. + return { valid: true, conclusive: false } } } @@ -658,6 +673,52 @@ export async function probeActualChunkShape< storeOpts?: Parameters[1], signal?: AbortSignal, ): Promise { + const { shape } = await probeChunkShape( + arr, + encodeChunkKey, + codecMeta, + bytesPerElement, + storeOpts, + signal, + ) + return shape +} + +/** + * A probed chunk shape, plus whether the probe actually concluded it. + * + * `conclusive` is false when the shape is what the probe fell back to after + * swallowing a store failure — a fetch that threw — rather than what it read + * from the data. (An abort of the `signal` the probe was given to watch is not + * swallowed at all: it propagates and ends the read.) + * + * The distinction exists because {@link resolveArrayInfo} memoises the result. + * Swallowing the failure is right for a single read: the probe is a heuristic + * correction, and failing a whole read because a *heuristic* could not fetch + * `c/0/0` would break reads of arrays whose first chunk merely happens to be + * unreachable. But an inconclusive answer must not outlive the call that made + * it. Before memoisation each read re-probed, so a transient blip cost one + * uncorrected read and healed itself; cached forever, that same blip leaves a + * mis-declared array decoding at the wrong shape for the lifetime of the + * store. So the fallback still returns — and is then refused a cache entry. + */ +interface ChunkShapeProbe { + shape: number[] + conclusive: boolean +} + +/** + * {@link probeActualChunkShape}, reporting whether the answer was concluded + * from data or fallen back to after a store failure. + */ +async function probeChunkShape( + arr: ZarrArray, + encodeChunkKey: (chunk_coords: number[]) => string, + codecMeta: CodecChunkMeta, + bytesPerElement: number, + storeOpts?: Parameters[1], + signal?: AbortSignal, +): Promise { const metadataChunkShape = codecMeta.chunk_shape const metaElements = metadataChunkShape.reduce((a, b) => a * b, 1) @@ -666,9 +727,11 @@ export async function probeActualChunkShape< const chunkKey = encodeChunkKey(zeroCoords) const chunkPath = arr.resolve(chunkKey).path + // Every early return below is conclusive: each is a determination made from + // bytes actually read, so re-probing would reach the same answer. try { const rawBytes = await arr.store.get(chunkPath, storeOpts) - if (!rawBytes) return metadataChunkShape + if (!rawBytes) return { shape: metadataChunkShape, conclusive: true } // Determine decompressed size via hybrid strategy const decompressedBytes = await probeDecompressedSize( @@ -676,10 +739,14 @@ export async function probeActualChunkShape< codecMeta, bytesPerElement, ) - if (decompressedBytes == null) return metadataChunkShape + if (decompressedBytes == null) { + return { shape: metadataChunkShape, conclusive: true } + } const actualElements = decompressedBytes / bytesPerElement - if (actualElements === metaElements) return metadataChunkShape + if (actualElements === metaElements) { + return { shape: metadataChunkShape, conclusive: true } + } // Mismatch detected — infer chunk shape from element count + heuristics const candidates = inferChunkShape( @@ -687,28 +754,35 @@ export async function probeActualChunkShape< metadataChunkShape, arr.shape, ) - if (candidates.length === 0) return metadataChunkShape + if (candidates.length === 0) { + return { shape: metadataChunkShape, conclusive: true } + } // Validate candidates by probing one-past-the-end. // The first candidate that passes validation wins. // Limit validation attempts to avoid excessive network requests. + // A candidate accepted because its validation probe *failed* rather than + // came back empty taints the result: the choice was a guess, so it is + // returned but not treated as settled. + let conclusive = true const maxValidationAttempts = Math.min(candidates.length, 5) for (let i = 0; i < maxValidationAttempts; i++) { const candidate = candidates[i] - const isValid = await validateCandidateChunkShape( + const validation = await validateCandidateChunkShape( arr, encodeChunkKey, candidate, storeOpts, signal, ) - if (isValid) { + if (!validation.conclusive) conclusive = false + if (validation.valid) { console.warn( `[fizarrita] Metadata chunk_shape ${JSON.stringify(metadataChunkShape)} ` + `does not match actual chunk data (${actualElements} elements). ` + `Using inferred chunk_shape: ${JSON.stringify(candidate)}`, ) - return candidate + return { shape: candidate, conclusive } } } @@ -719,13 +793,239 @@ export async function probeActualChunkShape< `does not match actual chunk data (${actualElements} elements). ` + `Using inferred chunk_shape: ${JSON.stringify(fallback)} (unvalidated)`, ) - return fallback + return { shape: fallback, conclusive } } catch (error) { // The catch-all exists to degrade gracefully when the probe fetch fails; // a caller abort is not that — it has to stop the whole read. if (signal?.aborted) throw error - return metadataChunkShape + // A store failure, not a determination — see ChunkShapeProbe. + return { shape: metadataChunkShape, conclusive: false } + } +} + +// --------------------------------------------------------------------------- +// Per-array resolution — metadata read + chunk-shape probe, memoised +// --------------------------------------------------------------------------- + +/** + * Resolved array info per store, keyed by array path. + * + * Both the array metadata and the probed chunk shape are immutable for the + * lifetime of an array, but resolving them costs store round-trips: one read + * of `zarr.json` (two, when falling back to v2), one chunk read for the probe, + * and up to five one-past-the-end probes on a mismatch. Without memoisation + * every `getWorker` call pays them *before* the chunk cache is consulted, so a + * fully populated cache cannot eliminate them — for a tiled viewer that is + * per-tile overhead scaling with pan/zoom activity rather than with cache + * misses. + * + * Keyed on the store instance (a WeakMap, so entries die with the store) plus + * the array path, mirroring the chunk-cache key of {@link createCacheKey}, so + * distinct `zarr.open` handles onto the same array share one entry. + * + * The promise is memoised, not the value, so concurrent `getWorker` calls on a + * cold array share one resolution instead of racing store reads. + */ +const resolvedArrayInfo = new WeakMap< + object, + Map> +>() + +function isAbortSignal(value: unknown): value is AbortSignal { + return ( + typeof value === "object" && + value !== null && + typeof (value as AbortSignal).aborted === "boolean" && + typeof (value as AbortSignal).addEventListener === "function" + ) +} + +/** + * Split a caller's store options into what a *shared* store request may carry + * and the caller's own `AbortSignal`, if the options hold one. + * + * Everything else — headers, credentials, cache mode — describes how to talk + * to the store and is the same for every caller of the same store, so it can + * safely govern a request made on behalf of all of them. A signal is the one + * option that belongs to a single caller: it says "I no longer want this", + * which is not a statement the others have made. + * + * Only a real signal is separated. Options with no `signal`, or one that is + * not an `AbortSignal`, are passed through untouched — same object, no copy. + */ +function separateSignal(storeOpts: Opts): { + shared: Opts + signal: AbortSignal | undefined +} { + if (storeOpts && typeof storeOpts === "object" && "signal" in storeOpts) { + const { signal, ...shared } = storeOpts as Opts & { signal?: unknown } + if (isAbortSignal(signal)) { + return { shared: shared as Opts, signal } + } + } + return { shared: storeOpts, signal: undefined } +} + +/** + * Settle as `promise` does, unless `signal` aborts first — then reject with + * the abort reason, exactly as a fetch given that signal would. `promise` + * itself is untouched and keeps running for whoever else awaits it. + * + * Whatever the outcome, `promise` is observed here: once the caller has been + * rejected on the signal's account, this is the only place still watching + * the promise it was handed, and a promise that later rejects with nobody + * watching is an unhandled rejection — fatal under Node's default. + */ +function untilAborted( + promise: Promise, + signal: AbortSignal | undefined, +): Promise { + if (!signal) return promise + const reason = () => + signal.reason ?? new DOMException("The operation was aborted.", "AbortError") + if (signal.aborted) { + // The caller never sees `promise` — a fresh derived promise at both call + // sites — so absorb its outcome rather than leave a rejection unhandled. + // Other observers of the same chain are unaffected by this. + promise.catch(() => {}) + return Promise.reject(reason()) + } + return new Promise((resolve, reject) => { + const onAbort = () => reject(reason()) + signal.addEventListener("abort", onAbort, { once: true }) + const settled = () => signal.removeEventListener("abort", onAbort) + promise.then( + (value) => { + settled() + resolve(value) + }, + (error) => { + settled() + reject(error) + }, + ) + }) +} + +/** + * A private copy of a memoised entry for one caller. `codecMeta` is plain + * JSON data (that is what {@link getMetaId} relies on), so a structured clone + * is a faithful deep copy; the key encoder is a stateless closure and is + * shared. Nothing the caller does to the copy can reach the memo, or the next + * caller — and nothing the memo holds is anyone else's object either: the + * entry itself is built from a clone (see {@link resolveArrayInfo}), because + * the v2 metadata path hands back zarrita's own `arr.chunks` array by + * reference. + */ +function detach(info: ArrayMetadata): ArrayMetadata { + return { + codecMeta: structuredClone(info.codecMeta), + encodeChunkKey: info.encodeChunkKey, + fillValue: structuredClone(info.fillValue), + } +} + +/** + * Read array metadata and probe the actual chunk shape, once per + * (store, array path) — repeat calls return the memoised promise without + * touching the store. + * + * The returned metadata's `codecMeta.chunk_shape` already carries the probe's + * correction, so it describes the chunks as stored, not as the metadata + * claimed. Each caller receives its own copy; the memoised entry is private + * to this module and cannot be reached, or altered, through a returned value. + * + * `storeOpts` is forwarded to every store read the resolution makes — both the + * metadata reads and the shape probe — with one exception: an `AbortSignal` in + * the options is not. The resolution is a shared, memoised resource whose + * result outlives every caller that wanted it, so it runs on the options that + * are the same for all of them (headers, credentials, and so on) and cannot be + * cancelled by any one of them. A caller's signal governs *its own wait* + * instead: aborting rejects that caller promptly with the signal's reason, + * while the callers sharing the resolution — and the memo — still get their + * result. Binding the shared request to whichever caller happened to start it + * would let one aborted tile fail every other tile that joined it, or hand + * them the fallback chunk shape for a probe *they* never aborted. The cost is + * that a resolution nobody wants any more still completes; since the next + * caller on the array is served from it, that is rarely wasted. + * + * Two outcomes are deliberately *not* kept. A rejected resolution is evicted, + * so a transient store failure is retried by the next call instead of becoming + * permanent. So is a resolution whose chunk-shape probe was inconclusive — one + * that swallowed a store failure and fell back to the declared shape rather + * than reading the real one (see {@link ChunkShapeProbe}). Both still serve the + * call that produced them, and every caller already waiting on them; they just + * do not outlive it. Caching a guess made under an error is how a one-off blip + * would otherwise turn into an array that decodes at the wrong shape forever. + */ +export function resolveArrayInfo( + arr: ZarrArray, + storeOpts?: Parameters[1], +): Promise { + const { shared, signal } = separateSignal(storeOpts) + + let infoByPath = resolvedArrayInfo.get(arr.store) + if (!infoByPath) { + infoByPath = new Map() + resolvedArrayInfo.set(arr.store, infoByPath) + } + const memoised = infoByPath.get(arr.path) + if (memoised) { + return untilAborted( + memoised.then(({ info }) => detach(info)), + signal, + ) + } + + const promise = (async () => { + const { codecMeta, encodeChunkKey, fillValue } = await readArrayMetadata( + arr, + shared, + ) + const Ctr = get_ctr(arr.dtype) + const bytesPerElement = (Ctr as unknown as { BYTES_PER_ELEMENT: number }) + .BYTES_PER_ELEMENT + // No signal for the probe to watch: its fetches run under `shared`, which + // carries none — the caller's signal governs the caller's wait (below), + // never the shared resolution. + const { shape, conclusive } = await probeChunkShape( + arr, + encodeChunkKey, + codecMeta, + bytesPerElement, + shared, + ) + return { + // Cloned so the memo owns its data outright — see detach. + info: detach({ + codecMeta: + shape !== codecMeta.chunk_shape + ? { ...codecMeta, chunk_shape: shape } + : codecMeta, + encodeChunkKey, + fillValue, + }), + conclusive, + } + })() + + const paths = infoByPath + paths.set(arr.path, promise) + // Conditional on the entry still being ours, so a later attempt that already + // replaced it is not dropped by our own settlement. + const forget = () => { + if (paths.get(arr.path) === promise) { + paths.delete(arr.path) + } } + promise.then(({ conclusive }) => { + if (!conclusive) forget() + }, forget) + + return untilAborted( + promise.then(({ info }) => detach(info)), + signal, + ) } // --------------------------------------------------------------------------- @@ -830,8 +1130,13 @@ export async function getWorker< assertSharedArrayBufferAvailable() } - // Read metadata from store — single read, single parse - const { codecMeta, encodeChunkKey, fillValue } = await readArrayMetadata( + // Metadata read + chunk-shape probe, memoised per (store, array path): + // only the first call on an array pays the store round-trips, so repeat + // reads served from a warm chunk cache never touch the store at all. + // codecMeta.chunk_shape is already the probed (possibly corrected) shape. + // Given `storeOpts`, so the combined signal governs *this call's wait* on + // the resolution — the resolution itself is shared and runs signal-free. + const { codecMeta, encodeChunkKey, fillValue } = await resolveArrayInfo( arr, storeOpts, ) @@ -840,42 +1145,25 @@ export async function getWorker< const bytesPerElement = (Ctr as unknown as { BYTES_PER_ELEMENT: number }) .BYTES_PER_ELEMENT - // Probe actual chunk shape — detects metadata vs data mismatch - const actualChunkShape = await probeActualChunkShape( - arr, - encodeChunkKey, - codecMeta, - bytesPerElement, - storeOpts, - // The probe's fetches run under the combined signal, so its abort - // detection has to watch the same one — with only `signal`, a store-level - // abort would be swallowed by the probe's catch-alls. - fetchSignal, - ) - // Checkpoint for stores that ignore the signal: their metadata and probe - // reads complete instead of rejecting, and this is the last await before - // the pool (whose own signal handling covers the rest) — without it, a - // fully-cached read would return data after its caller already walked away. - // Watches the combined signal so a store-level abort is caught too. + // reads complete instead of rejecting — and a memoised resolution never + // touches the store at all — and this is the last await before the pool + // (whose own signal handling covers the rest). Without it, a fully-cached + // read would return data after its caller already walked away. Watches the + // combined signal so a store-level abort is caught too. if (fetchSignal?.aborted) { throw fetchSignal.reason } - // Update codecMeta to use the actual chunk shape for codec pipeline - const correctedCodecMeta = - actualChunkShape !== codecMeta.chunk_shape - ? { ...codecMeta, chunk_shape: actualChunkShape } - : codecMeta - // Get stable metaId for the codec metadata (used by worker-rpc meta-init) - const metaId = getMetaId(correctedCodecMeta) + const metaId = getMetaId(codecMeta) // Set up the indexer with the actual (possibly corrected) chunk shape + const chunkShape = codecMeta.chunk_shape const indexer = new BasicIndexer({ selection, shape: arr.shape, - chunk_shape: actualChunkShape, + chunk_shape: chunkShape, }) // Allocate output — backed by SharedArrayBuffer when requested @@ -885,9 +1173,6 @@ export async function getWorker< const outStride = get_strides(indexer.shape) const out = setter.prepare(data, indexer.shape, outStride) as Chunk - // Pre-compute chunk invariants (hoisted out of loop) - const chunkShape = actualChunkShape - // Build tasks — one per chunk const tasks: WorkerPoolTask[] = [] @@ -959,7 +1244,7 @@ export async function getWorker< worker, rawBytes, metaId, - correctedCodecMeta, + codecMeta, buffer as SharedArrayBuffer, size * bytesPerElement, outStride, @@ -999,7 +1284,7 @@ export async function getWorker< worker, rawBytes, metaId, - correctedCodecMeta, + codecMeta, isEdgeChunk ? edgeChunkShape : undefined, ) } catch (error) { diff --git a/fizarrita/src/index.ts b/fizarrita/src/index.ts index 0b044f6..eba1f43 100644 --- a/fizarrita/src/index.ts +++ b/fizarrita/src/index.ts @@ -23,6 +23,7 @@ export { readArrayMetadata, readBloscFrameContentSize, readZstdFrameContentSize, + resolveArrayInfo, } from "./get-worker.js" // Internals — exported for building custom workers that extend the codec worker export { create_codec_pipeline } from "./internals/codec-pipeline.js" diff --git a/test/browser/zarrita-worker.spec.ts b/test/browser/zarrita-worker.spec.ts index 5b2d174..6caf1ec 100644 --- a/test/browser/zarrita-worker.spec.ts +++ b/test/browser/zarrita-worker.spec.ts @@ -2256,8 +2256,8 @@ test.describe('@fideus-labs/fizarrita — getWorker / setWorker', () => { // Hold reads of c/1 open until every caller has asked for it. Only c/1 is // gated: the shape probe reads c/0 before any task runs, so gating that - // would stall all three callers before they reached the task phase and - // nothing would ever overlap. + // would stall the shared metadata resolution and nothing would ever + // overlap. const originalGet = arr.store.get.bind(arr.store) const chunkPaths: string[] = [] let gatedRequests = 0 @@ -2276,6 +2276,21 @@ test.describe('@fideus-labs/fizarrita — getWorker / setWorker', () => { return originalGet(path, ...rest) } + // Metadata and the shape probe are memoised per array, so callers two + // and three perform no store I/O of their own before their task phase — + // store reads can no longer signal that every caller is under way. + // Submitting tasks is the observable that remains: with free slots in + // the pool, runTasks invokes each task function synchronously, and a + // c/1 task's first act is to register in the in-flight map, joining the + // gated fetch rather than re-issuing it. Three runTasks calls therefore + // mean every caller's c/1 task has already joined. + let runTasksCalls = 0 + const originalRunTasks = pool.runTasks.bind(pool) + ;(pool as any).runTasks = (...args: any[]) => { + runTasksCalls += 1 + return originalRunTasks(...args) + } + const waitFor = async (ready: () => boolean, label: string) => { const deadline = Date.now() + 5000 while (!ready()) { @@ -2290,19 +2305,13 @@ test.describe('@fideus-labs/fizarrita — getWorker / setWorker', () => { getWorker(arr, null, { pool }), ] - // Release only once every caller is demonstrably at the point of wanting - // c/1, rather than after a fixed sleep. - // - // The gated-request count cannot be the signal — dedup working means only - // one request ever reaches the gate, so waiting for three would hang on a - // passing run. The shape probe is the observable that survives dedup: it - // reads c/0 once per call, outside the task path, before that call builds - // any tasks. Three c/0 reads therefore means all three callers are past - // probing and into their task phase, and a task's first act is the - // store.get for its chunk — no I/O in between. + // Release only once every caller is demonstrably past the point of + // wanting c/1, rather than after a fixed sleep. The gated-request count + // alone cannot be the signal — dedup working means only one request ever + // reaches the gate, so waiting for three would hang on a passing run. await waitFor( - () => chunkPaths.filter((path) => path.endsWith('/c/0')).length >= 3, - 'all three callers to finish probing', + () => runTasksCalls >= 3, + 'all three callers to submit their tasks', ) await waitFor(() => gatedRequests >= 1, 'the shared c/1 fetch to begin') release() diff --git a/test/node/fizarrita.test.js b/test/node/fizarrita.test.js index 7108036..52c9056 100644 --- a/test/node/fizarrita.test.js +++ b/test/node/fizarrita.test.js @@ -15,6 +15,7 @@ import { createDefaultWorker, DEFAULT_WORKER_URL, getWorker, + resolveArrayInfo, setWorker, } from '../../fizarrita/dist/index.js' @@ -60,6 +61,31 @@ async function withPool(size, fn) { } } +/** A Map store that records every `get`, so tests can count store round-trips. */ +class CountingStore extends Map { + reads = [] + get(key) { + this.reads?.push(key) + return super.get(key) + } +} + +/** An 8x8 int32 array over 4x4 chunks on a CountingStore, fully populated. */ +async function makeCountedArray() { + const store = new CountingStore() + const arr = await zarr.create(zarr.root(store).resolve('/data'), { + shape: [8, 8], + chunk_shape: [4, 4], + data_type: 'int32', + }) + await zarr.set(arr, null, { + data: Int32Array.from({ length: 64 }, (_, i) => i), + shape: [8, 8], + stride: [8, 1], + }) + return { store, arr } +} + test('setWorker/getWorker round-trip a scalar fill', async () => { const arr = await makeArray({ shape: [8, 8], chunk_shape: [4, 4] }) @@ -432,9 +458,11 @@ test('a concurrent read survives another read aborting their shared chunks', { t const readB = getWorker(arr, null, { pool: poolB }) // B re-reads the unparked first chunk itself (A's share of it has long - // settled), then joins A's parked in-flight fetches as a follower. + // settled), then joins A's parked in-flight fetches as a follower. Three + // reads of that chunk by then: A's shape probe, A's fetch, B's fetch — + // B does not probe, the array info is memoised from A's resolution. await waitFor( - () => store.chunkGets.filter((k) => k === '/data/c/0/0').length >= 4, + () => store.chunkGets.filter((k) => k === '/data/c/0/0').length >= 3, 'read B should have read the first chunk', ) // Wait for B to have progressed into fetching additional chunks. Since @@ -481,3 +509,395 @@ test('a decode failure rejects instead of hanging', { timeout: 15_000 }, async ( await assert.rejects(getWorker(arr, null, { pool })) }) }) + +// Issue #6 — the metadata read and chunk-shape probe used to run on every +// getWorker call, ahead of the chunk cache, so a fully populated cache could +// never eliminate them. Both are now memoised per (store, array path). + +test('a warm chunk cache serves a repeat read with zero store round-trips', async () => { + const { store, arr } = await makeCountedArray() + + await withPool(2, async (pool) => { + const cache = new Map() + store.reads.length = 0 + const first = await getWorker(arr, null, { pool, cache }) + assert.ok( + store.reads.includes('/data/zarr.json'), + 'the first read resolves metadata from the store', + ) + + store.reads.length = 0 + const second = await getWorker(arr, null, { pool, cache }) + assert.deepEqual(Array.from(second.data), Array.from(first.data)) + assert.deepEqual(store.reads, [], 'the repeat read never touches the store') + }) +}) + +test('repeat reads without a cache pay only the chunk fetches', async () => { + const { store, arr } = await makeCountedArray() + + await withPool(2, async (pool) => { + await getWorker(arr, null, { pool }) + + store.reads.length = 0 + await getWorker(arr, null, { pool }) + assert.deepEqual( + [...store.reads].sort(), + ['/data/c/0/0', '/data/c/0/1', '/data/c/1/0', '/data/c/1/1'], + 'no metadata read, no probe — one fetch per chunk', + ) + }) +}) + +test('concurrent reads on a cold array share one metadata read and one probe', async () => { + const { store, arr } = await makeCountedArray() + + await withPool(2, async (pool) => { + const cache = new Map() + store.reads.length = 0 + const [a, b] = await Promise.all([ + getWorker(arr, null, { pool, cache }), + getWorker(arr, null, { pool, cache }), + ]) + assert.deepEqual(Array.from(a.data), Array.from(b.data)) + + const metadataReads = store.reads.filter((k) => k === '/data/zarr.json') + assert.equal(metadataReads.length, 1, 'one zarr.json read for both calls') + // 1 metadata read + 1 probe of c/0/0 + 4 chunk fetches: the probe and the + // chunk fetch of c/0/0 both count, everything else exactly once. + assert.equal(store.reads.length, 6, `reads: ${store.reads.join(', ')}`) + }) +}) + +test('store options reach the metadata reads, not just the probe', async () => { + const store = new CountingStore() + const arr = await zarr.create(zarr.root(store).resolve('/data'), { + shape: [8, 8], + chunk_shape: [4, 4], + data_type: 'int32', + }) + await zarr.set(arr, null, { + data: Int32Array.from({ length: 64 }, (_, i) => i), + shape: [8, 8], + stride: [8, 1], + }) + + // Record the options every read is given, keyed by path. + const seenOpts = new Map() + const realGet = CountingStore.prototype.get.bind(store) + store.get = (key, opts) => { + seenOpts.set(key, opts) + return realGet(key) + } + + const marker = { headers: { authorization: 'sentinel' } } + await withPool(2, async (pool) => { + await getWorker(arr, null, { pool, opts: marker }) + }) + + // The metadata read used to be the one store request that silently dropped + // the caller's options while the probe and chunk fetches honoured them. + // Same object, not a copy: options without a signal are passed through as-is. + assert.equal(seenOpts.get('/data/zarr.json'), marker) + assert.equal(seenOpts.get('/data/c/0/0'), marker) + assert.equal(seenOpts.get('/data/c/1/1'), marker) +}) + +/** + * A store whose reads honour `opts.signal` the way fetch does — rejecting + * with the signal's reason, including mid-flight — and which, once `hold(re)` + * is called, does not complete reads of paths matching `re` until `release()`. + * Lets a test hold a resolution open, act while it is in flight, then let it + * finish. Armed after setup so zarrita's own reads are never parked. + * + * `entered` resolves when the first gated read starts, so a test can wait for + * "the resolution is in flight" instead of guessing at it with a sleep. + */ +class GatedStore extends CountingStore { + #gate = null + #open + #release + #entered + entered + constructor() { + super() + this.#open = new Promise((resolve) => { + this.#release = resolve + }) + this.entered = new Promise((resolve) => { + this.#entered = resolve + }) + } + hold(gate) { + this.#gate = gate + } + release() { + this.#release() + } + get(key, opts) { + const signal = opts?.signal + const bytes = super.get(key) + return new Promise((resolve, reject) => { + if (signal?.aborted) return reject(signal.reason) + signal?.addEventListener('abort', () => reject(signal.reason), { + once: true, + }) + const gated = this.#gate?.test(key) ?? false + if (gated) this.#entered() + const wait = gated ? this.#open : Promise.resolve() + wait.then(() => resolve(bytes)) + }) + } +} + +/** Populate 8x8 int32 0..63 at `/data` on `store`, chunked `chunk_shape`. */ +async function populate(store, chunk_shape) { + const arr = await zarr.create(zarr.root(store).resolve('/data'), { + shape: [8, 8], + chunk_shape, + data_type: 'int32', + }) + const expected = Int32Array.from({ length: 64 }, (_, i) => i) + await zarr.set(arr, null, { data: expected, shape: [8, 8], stride: [8, 1] }) + return { arr, expected } +} + +test('aborting one caller does not fail the callers sharing its metadata resolution', async () => { + // Hold the zarr.json read open so both callers are in flight on one + // resolution when the first one aborts. + const store = new GatedStore() + const { arr, expected } = await populate(store, [4, 4]) + store.hold(/zarr\.json$/) + + await withPool(2, async (pool) => { + const first = new AbortController() + const second = new AbortController() + const a = getWorker(arr, null, { pool, opts: { signal: first.signal } }) + const b = getWorker(arr, null, { pool, opts: { signal: second.signal } }) + // The shared read is parked at the gate — both callers are on it, the + // second having joined the memoised resolution without touching the store. + await store.entered + first.abort() + + // The aborter is rejected promptly with its own reason — the shared read + // is still parked behind the gate at this point. + await assert.rejects(a, (err) => err?.name === 'AbortError') + + store.release() + // The other caller neither aborted nor failed: it gets its data. + const result = await b + assert.deepEqual(Array.from(result.data), Array.from(expected)) + // And the resolution really was shared — one metadata read for both. + const metadataReads = store.reads.filter((k) => k === '/data/zarr.json') + assert.equal(metadataReads.length, 1) + }) +}) + +test('aborting one caller does not hand the callers sharing its probe an unprobed chunk shape', async () => { + // Chunks are really 4x8; the metadata is rewritten to claim 4x4, so the + // probe's correction is load-bearing. Hold the probe's chunk read open. + const store = new GatedStore() + const { expected } = await populate(store, [4, 8]) + const meta = JSON.parse( + new TextDecoder().decode(await store.get('/data/zarr.json')), + ) + meta.chunk_grid.configuration.chunk_shape = [4, 4] + store.set('/data/zarr.json', new TextEncoder().encode(JSON.stringify(meta))) + const misdeclared = await zarr.open(zarr.root(store).resolve('/data'), { + kind: 'array', + }) + store.hold(/\/c\/0\/0$/) + + await withPool(2, async (pool) => { + const first = new AbortController() + const a = getWorker(misdeclared, null, { pool, opts: { signal: first.signal } }) + const b = getWorker(misdeclared, null, { pool }) + await store.entered + first.abort() + await assert.rejects(a, (err) => err?.name === 'AbortError') + + store.release() + // Had the aborter's signal governed the shared probe, the probe would have + // swallowed the abort and fallen back to the declared 4x4 for everyone — + // and this read would come back at the wrong shape. + const result = await b + assert.deepEqual(result.shape, [8, 8]) + assert.deepEqual(Array.from(result.data), Array.from(expected)) + }) +}) + +test('a lone caller that aborts is rejected promptly, and the resolution still lands for the next one', async () => { + const store = new GatedStore() + const { arr, expected } = await populate(store, [4, 4]) + store.hold(/zarr\.json$/) + + await withPool(2, async (pool) => { + const controller = new AbortController() + const a = getWorker(arr, null, { pool, opts: { signal: controller.signal } }) + await store.entered + controller.abort() + // Rejected while the store is still gated: the abort did not wait for the + // shared read to finish. + await assert.rejects(a, (err) => err?.name === 'AbortError') + + store.release() + // The resolution completed on its own and was memoised: joining it here + // waits for exactly that, and the next caller pays no metadata read at all. + await resolveArrayInfo(arr) + store.reads.length = 0 + const result = await getWorker(arr, null, { pool }) + assert.deepEqual(Array.from(result.data), Array.from(expected)) + assert.ok(!store.reads.includes('/data/zarr.json'), store.reads.join(', ')) + }) +}) + +test('a caller that had already aborted leaves no unhandled rejection behind when the resolution it declined then fails', async () => { + const store = new GatedStore() + const { arr, expected } = await populate(store, [4, 4]) + store.hold(/zarr\.json$/) + // Once released, the metadata read fails — once. The resolution the aborted + // caller declined is the one that fails; the read after it succeeds. + let failures = 1 + const gatedGet = store.get.bind(store) + store.get = (key, opts) => + gatedGet(key, opts).then((bytes) => { + if (key.endsWith('zarr.json') && failures > 0) { + failures-- + throw new Error('transient store failure') + } + return bytes + }) + + const unhandled = [] + const onUnhandled = (reason) => unhandled.push(reason) + process.on('unhandledRejection', onUnhandled) + try { + await withPool(1, async (pool) => { + // A store-level signal is not pre-checked by getWorker the way + // `opts.signal` is: it reaches resolveArrayInfo already aborted, and the + // caller is rejected on the spot — while the shared read is still parked. + const controller = new AbortController() + controller.abort(new Error('gone before it began')) + await assert.rejects( + getWorker(arr, null, { pool, opts: { signal: controller.signal } }), + /gone before it began/, + ) + await store.entered + + // Nobody is waiting on that resolution any more. Let it fail now, and + // give the runtime a turn to report a rejection nobody handled. + store.release() + await new Promise((r) => setImmediate(r)) + await new Promise((r) => setImmediate(r)) + assert.deepEqual(unhandled, [], 'a declined resolution failed unobserved') + + // The failure was not memoised either: the next read resolves afresh. + const result = await getWorker(arr, null, { pool }) + assert.deepEqual(Array.from(result.data), Array.from(expected)) + }) + } finally { + process.off('unhandledRejection', onUnhandled) + } +}) + +test('resolveArrayInfo hands out copies — mutating one cannot reach the memo or other callers', async () => { + const { store, arr } = await makeCountedArray() + + const first = await resolveArrayInfo(arr) + // Vandalise everything a caller could reach. + first.codecMeta.chunk_shape[0] = 999 + first.codecMeta.chunk_shape.push(1) + first.codecMeta.codecs.push({ name: 'bogus', configuration: {} }) + first.codecMeta.data_type = 'float64' + + const second = await resolveArrayInfo(arr) + assert.notEqual(second.codecMeta, first.codecMeta) + assert.deepEqual(second.codecMeta.chunk_shape, [4, 4]) + assert.ok( + !second.codecMeta.codecs.some((c) => c.name === 'bogus'), + 'the injected codec did not reach the memo', + ) + assert.equal(second.codecMeta.data_type, 'int32') + + // zarrita's own metadata is untouched too — the memo aliases nobody's data. + assert.deepEqual(arr.chunks, [4, 4]) + + // And a read after the vandalism decodes as it should. + await withPool(2, async (pool) => { + store.reads.length = 0 + const result = await getWorker(arr, null, { pool }) + assert.deepEqual(result.shape, [8, 8]) + assert.equal(result.data[63], 63) + assert.ok(!store.reads.includes('/data/zarr.json'), 'still memoised') + }) +}) + +test('a probe that fails transiently is retried, not memoised as a missed correction', async () => { + const store = new CountingStore() + // Chunks are really 4x8. The metadata is rewritten below to claim 4x4, so + // the shape probe has real work to do — exactly the case where silently + // memoising "no correction needed" would corrupt every later read. + const arr = await zarr.create(zarr.root(store).resolve('/data'), { + shape: [8, 8], + chunk_shape: [4, 8], + data_type: 'int32', + }) + const expected = Int32Array.from({ length: 64 }, (_, i) => i) + await zarr.set(arr, null, { data: expected, shape: [8, 8], stride: [8, 1] }) + + const meta = JSON.parse(new TextDecoder().decode(store.get('/data/zarr.json'))) + meta.chunk_grid.configuration.chunk_shape = [4, 4] + store.set( + '/data/zarr.json', + new TextEncoder().encode(JSON.stringify(meta)), + ) + const misdeclared = await zarr.open(zarr.root(store).resolve('/data'), { + kind: 'array', + }) + + // Fail the probe's chunk fetch exactly once. The metadata read must still + // succeed, or the resolution would reject and be evicted by the other path. + let probeFailures = 1 + const realGet = CountingStore.prototype.get.bind(store) + store.get = (key) => { + if (key.includes('/c/') && probeFailures > 0) { + probeFailures-- + throw new Error('transient probe failure') + } + return realGet(key) + } + + await withPool(2, async (pool) => { + // The first read loses the probe and falls back to the declared 4x4. + // Whether it then throws or returns misshapen data is not the point — + // what matters is that the miss is not remembered. + await getWorker(misdeclared, null, { pool }).catch(() => {}) + + // The second read probes again and finds the real 4x8 chunking. + const result = await getWorker(misdeclared, null, { pool }) + assert.deepEqual(result.shape, [8, 8]) + assert.deepEqual(Array.from(result.data), Array.from(expected)) + }) +}) + +test('a failed metadata read is retried, not memoised', async () => { + const { store, arr } = await makeCountedArray() + + let failures = 1 + const realGet = CountingStore.prototype.get.bind(store) + store.get = (key) => { + if (failures > 0) { + failures-- + throw new Error('transient store failure') + } + return realGet(key) + } + + await withPool(2, async (pool) => { + await assert.rejects(getWorker(arr, null, { pool }), /transient/) + + const result = await getWorker(arr, null, { pool }) + assert.deepEqual(result.shape, [8, 8]) + assert.equal(result.data[63], 63) + }) +})