From fe9bc6a145087e1b2a6d0a26a68336558242cf9c Mon Sep 17 00:00:00 2001 From: Matt McCormick Date: Thu, 13 Aug 2026 15:07:33 -0400 Subject: [PATCH 1/7] perf: memoise metadata read and chunk-shape probe per (store, path) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit getWorker paid two store round-trips of its own on every call — the zarr.json read in readArrayMetadata and the chunk fetch in probeActualChunkShape — both ahead of the per-chunk cache lookup, so a fully populated ChunkCache could never eliminate them. For a tiled viewer that is per-tile overhead scaling with pan/zoom activity rather than with cache misses. Both results are immutable for the lifetime of an array, so resolveArrayInfo now memoises them per (store, array path): a WeakMap keyed on the store instance (entries die with the store), holding a Map keyed by array path, mirroring the chunk-cache key. The promise is memoised rather than the value, so concurrent calls on a cold array share one resolution; a rejected resolution is evicted so a transient store failure is retried instead of becoming permanent. The memoised codecMeta.chunk_shape already carries the probe's correction, which also drops the per-call correctedCodecMeta spread. A repeat read served from a warm cache now performs zero store requests, covered by new node tests counting store.get calls. The browser test for concurrent chunk dedup used "three c/0 probe reads" as its readiness signal — exactly the redundancy removed here — and now counts runTasks submissions instead, which is sound because runTasks invokes task functions synchronously while the pool has free slots. Fixes #6 Co-Authored-By: Claude Fable 5 --- fizarrita/README.md | 5 + fizarrita/src/get-worker.ts | 140 +++++++++++++++++++++------- fizarrita/src/index.ts | 1 + test/browser/zarrita-worker.spec.ts | 37 +++++--- test/node/fizarrita.test.js | 112 +++++++++++++++++++++- 5 files changed, 247 insertions(+), 48 deletions(-) diff --git a/fizarrita/README.md b/fizarrita/README.md index cc02b1f..fa5b180 100644 --- a/fizarrita/README.md +++ b/fizarrita/README.md @@ -178,6 +178,11 @@ 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. + ### 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..9141641 100644 --- a/fizarrita/src/get-worker.ts +++ b/fizarrita/src/get-worker.ts @@ -728,6 +728,97 @@ export async function probeActualChunkShape< } } +// --------------------------------------------------------------------------- +// 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> +>() + +/** + * 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. + * + * `storeOpts` only reaches the store on the call that performs the resolution; + * memoised results are shared across callers regardless of their options. A + * rejected resolution is evicted so a transient store failure is retried by + * the next call instead of becoming permanent. + */ +export function resolveArrayInfo( + arr: ZarrArray, + storeOpts?: Parameters[1], +): Promise { + 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 memoised + + const promise = (async (): Promise => { + const { codecMeta, encodeChunkKey, fillValue } = await readArrayMetadata(arr) + const Ctr = get_ctr(arr.dtype) + const bytesPerElement = (Ctr as unknown as { BYTES_PER_ELEMENT: number }) + .BYTES_PER_ELEMENT + // The probe's fetches run under `storeOpts`, so its abort detection has + // to watch the same signal — otherwise its catch-alls would swallow an + // abort as a store failure and hand back the fallback shape. + const signal = (storeOpts as { signal?: AbortSignal } | undefined)?.signal + const chunkShape = await probeActualChunkShape( + arr, + encodeChunkKey, + codecMeta, + bytesPerElement, + storeOpts, + signal, + ) + return { + codecMeta: + chunkShape !== codecMeta.chunk_shape + ? { ...codecMeta, chunk_shape: chunkShape } + : codecMeta, + encodeChunkKey, + fillValue, + } + })() + + const paths = infoByPath + paths.set(arr.path, promise) + promise.catch(() => { + if (paths.get(arr.path) === promise) { + paths.delete(arr.path) + } + }) + return promise +} + // --------------------------------------------------------------------------- // getWorker // --------------------------------------------------------------------------- @@ -830,8 +921,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. + // Runs under `storeOpts`, so the store reads it makes carry the combined + // signal. + const { codecMeta, encodeChunkKey, fillValue } = await resolveArrayInfo( arr, storeOpts, ) @@ -840,42 +936,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 +964,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 +1035,7 @@ export async function getWorker< worker, rawBytes, metaId, - correctedCodecMeta, + codecMeta, buffer as SharedArrayBuffer, size * bytesPerElement, outStride, @@ -999,7 +1075,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..a295fff 100644 --- a/test/node/fizarrita.test.js +++ b/test/node/fizarrita.test.js @@ -60,6 +60,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 +457,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 +508,84 @@ 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('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) + }) +}) From d624380e7943f52d2c1c46bc9cd4869128140384 Mon Sep 17 00:00:00 2001 From: Matt McCormick Date: Thu, 13 Aug 2026 15:51:37 -0400 Subject: [PATCH 2/7] fix: forward store options to the array metadata reads MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit readArrayMetadata took no store options, so its zarr.json and .zarray reads went out bare while probeActualChunkShape and every chunk fetch honoured the caller's opts. An AbortSignal would abort the chunk reads and leave the metadata read running; auth headers reached every request but that one. resolveArrayInfo made the split conspicuous by passing opts to the probe and not to the metadata read beside it, and its docstring claimed the options reached the store, unqualified. readArrayMetadata now takes storeOpts as an optional second parameter — backward compatible for a function exported from the package index — and forwards it to both reads. The resolveArrayInfo docstring now says what is actually true: options are forwarded to every read the resolution makes, but only on the call that performs it, so a caller served by the memoised promise contributes no request for its own signal to abort. Raised by Copilot in review of #14. Co-Authored-By: Claude Fable 5 --- fizarrita/src/get-worker.ts | 24 ++++++++++++++++++++---- test/node/fizarrita.test.js | 33 +++++++++++++++++++++++++++++++++ 2 files changed, 53 insertions(+), 4 deletions(-) diff --git a/fizarrita/src/get-worker.ts b/fizarrita/src/get-worker.ts index 9141641..789dfd1 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, @@ -765,9 +774,13 @@ const resolvedArrayInfo = new WeakMap< * correction, so it describes the chunks as stored, not as the metadata * claimed. * - * `storeOpts` only reaches the store on the call that performs the resolution; - * memoised results are shared across callers regardless of their options. A - * rejected resolution is evicted so a transient store failure is retried by + * `storeOpts` is forwarded to every store read the resolution makes — both the + * metadata reads and the shape probe — but only on the call that actually + * performs it: a later caller hitting the memoised promise contributes no + * store request for its own options to govern. An AbortSignal therefore aborts + * the resolution it started, not one already in flight for someone else. + * + * A rejected resolution is evicted so a transient store failure is retried by * the next call instead of becoming permanent. */ export function resolveArrayInfo( @@ -783,7 +796,10 @@ export function resolveArrayInfo( if (memoised) return memoised const promise = (async (): Promise => { - const { codecMeta, encodeChunkKey, fillValue } = await readArrayMetadata(arr) + const { codecMeta, encodeChunkKey, fillValue } = await readArrayMetadata( + arr, + storeOpts, + ) const Ctr = get_ctr(arr.dtype) const bytesPerElement = (Ctr as unknown as { BYTES_PER_ELEMENT: number }) .BYTES_PER_ELEMENT diff --git a/test/node/fizarrita.test.js b/test/node/fizarrita.test.js index a295fff..f544c3a 100644 --- a/test/node/fizarrita.test.js +++ b/test/node/fizarrita.test.js @@ -568,6 +568,39 @@ test('concurrent reads on a cold array share one metadata read and one probe', a }) }) +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. + 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) +}) + test('a failed metadata read is retried, not memoised', async () => { const { store, arr } = await makeCountedArray() From 1d8aaccd266b86d37cb6db28ceba8efce9085f17 Mon Sep 17 00:00:00 2001 From: Matt McCormick Date: Thu, 13 Aug 2026 16:00:43 -0400 Subject: [PATCH 3/7] fix: do not memoise a chunk-shape probe that swallowed a store failure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit probeActualChunkShape catches store failures and falls back to the declared chunk shape. For a single read that is right: 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. Each read re-probed, so a transient blip cost one uncorrected read and healed itself. Memoising the result removed the healing. A mis-declared array that hits one blip during its first probe now decodes at the wrong shape for the lifetime of the store — a regression this branch introduced. The fallback still returns; it is now refused a cache entry. probeChunkShape reports whether the shape was concluded from bytes it read or fallen back to after an error, and resolveArrayInfo evicts an inconclusive resolution once it settles, on the same conditional-identity path already used for rejections. Callers already awaiting it are still served — only the memoisation is withheld. validateCandidateChunkShape reports the same way, since accepting a candidate because its probe threw is equally a guess made under an error; it cannot tell a 404 from a network failure. Errors are not propagated instead, as review suggested: reads that work today would start throwing, and 404-throwing stores make the errors unclassifiable at that layer. Covered by a test that mis-declares 4x8 chunks as 4x4, fails the first probe fetch, and asserts the second read finds the real chunking. Verified to fail without the eviction. Raised by CodeRabbit in review of #14. Co-Authored-By: Claude Fable 5 --- fizarrita/src/get-worker.ts | 145 ++++++++++++++++++++++++++++-------- test/node/fizarrita.test.js | 48 ++++++++++++ 2 files changed, 161 insertions(+), 32 deletions(-) diff --git a/fizarrita/src/get-worker.ts b/fizarrita/src/get-worker.ts index 789dfd1..9df4186 100644 --- a/fizarrita/src/get-worker.ts +++ b/fizarrita/src/get-worker.ts @@ -584,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, @@ -596,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 @@ -616,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, @@ -630,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 } } } @@ -667,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) @@ -675,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( @@ -685,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( @@ -696,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 } } } @@ -728,12 +793,13 @@ 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 } } } @@ -762,7 +828,7 @@ export async function probeActualChunkShape< */ const resolvedArrayInfo = new WeakMap< object, - Map> + Map> >() /** @@ -780,8 +846,14 @@ const resolvedArrayInfo = new WeakMap< * store request for its own options to govern. An AbortSignal therefore aborts * the resolution it started, not one already in flight for someone else. * - * A rejected resolution is evicted so a transient store failure is retried by - * the next call instead of becoming permanent. + * 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, @@ -793,9 +865,9 @@ export function resolveArrayInfo( resolvedArrayInfo.set(arr.store, infoByPath) } const memoised = infoByPath.get(arr.path) - if (memoised) return memoised + if (memoised) return memoised.then(({ info }) => info) - const promise = (async (): Promise => { + const promise = (async () => { const { codecMeta, encodeChunkKey, fillValue } = await readArrayMetadata( arr, storeOpts, @@ -807,7 +879,7 @@ export function resolveArrayInfo( // to watch the same signal — otherwise its catch-alls would swallow an // abort as a store failure and hand back the fallback shape. const signal = (storeOpts as { signal?: AbortSignal } | undefined)?.signal - const chunkShape = await probeActualChunkShape( + const { shape, conclusive } = await probeChunkShape( arr, encodeChunkKey, codecMeta, @@ -816,23 +888,32 @@ export function resolveArrayInfo( signal, ) return { - codecMeta: - chunkShape !== codecMeta.chunk_shape - ? { ...codecMeta, chunk_shape: chunkShape } - : codecMeta, - encodeChunkKey, - fillValue, + info: { + codecMeta: + shape !== codecMeta.chunk_shape + ? { ...codecMeta, chunk_shape: shape } + : codecMeta, + encodeChunkKey, + fillValue, + }, + conclusive, } })() const paths = infoByPath paths.set(arr.path, promise) - promise.catch(() => { + // 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) } - }) - return promise + } + promise.then(({ conclusive }) => { + if (!conclusive) forget() + }, forget) + + return promise.then(({ info }) => info) } // --------------------------------------------------------------------------- diff --git a/test/node/fizarrita.test.js b/test/node/fizarrita.test.js index f544c3a..23e10e2 100644 --- a/test/node/fizarrita.test.js +++ b/test/node/fizarrita.test.js @@ -601,6 +601,54 @@ test('store options reach the metadata reads, not just the probe', async () => { assert.equal(seenOpts.get('/data/c/1/1'), marker) }) +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() From 2dde285ec93015f4f264d7fafcd90254ac467bc3 Mon Sep 17 00:00:00 2001 From: Matt McCormick Date: Tue, 18 Aug 2026 18:18:45 -0400 Subject: [PATCH 4/7] fix: isolate the shared array resolution from any one caller, and hand out copies MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit resolveArrayInfo memoises one resolution per (store, path) that concurrent callers join, but the shared store requests ran on whichever caller's storeOpts happened to start them. A signal belongs to one caller: if the initiator aborted mid-metadata-read every joined caller rejected with an AbortError they never asked for, and if it aborted mid-probe the probe swallowed the abort and every joined caller got the fallback chunk shape for a probe *they* did not abort. Now the shared resolution runs on the options common to all callers (headers, credentials, ...) with the signal separated out; each caller's signal governs its own wait — aborting rejects that caller promptly with the signal's reason, while the resolution runs on for the others and for the memo. The memoised entry was also returned by reference — and, via the v2 metadata path, aliased zarrita's own arr.chunks — so a caller could rewrite codecMeta.chunk_shape or codecs for every later read on the array. The memo is now built from a clone and each caller receives its own structured clone (getMetaId keys on JSON.stringify, so clones share a metaId). Regression tests: an aborting caller neither fails the callers sharing its metadata read nor hands them an unprobed shape; a lone aborter is rejected promptly and its resolution still lands for the next call; mutating a returned copy reaches neither the memo nor the next caller. Co-Authored-By: Claude Fable 5 --- fizarrita/README.md | 5 ++ fizarrita/src/get-worker.ts | 137 +++++++++++++++++++++++++---- test/node/fizarrita.test.js | 168 ++++++++++++++++++++++++++++++++++++ 3 files changed, 292 insertions(+), 18 deletions(-) diff --git a/fizarrita/README.md b/fizarrita/README.md index fa5b180..be6bbf1 100644 --- a/fizarrita/README.md +++ b/fizarrita/README.md @@ -182,6 +182,11 @@ 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 diff --git a/fizarrita/src/get-worker.ts b/fizarrita/src/get-worker.ts index 9df4186..da10f8e 100644 --- a/fizarrita/src/get-worker.ts +++ b/fizarrita/src/get-worker.ts @@ -831,6 +831,89 @@ const resolvedArrayInfo = new WeakMap< 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. + */ +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) 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 @@ -838,13 +921,22 @@ const resolvedArrayInfo = new WeakMap< * * 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. + * 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 — but only on the call that actually - * performs it: a later caller hitting the memoised promise contributes no - * store request for its own options to govern. An AbortSignal therefore aborts - * the resolution it started, not one already in flight for someone else. + * 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 @@ -859,43 +951,49 @@ 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 memoised.then(({ info }) => info) + if (memoised) { + return untilAborted( + memoised.then(({ info }) => detach(info)), + signal, + ) + } const promise = (async () => { const { codecMeta, encodeChunkKey, fillValue } = await readArrayMetadata( arr, - storeOpts, + shared, ) const Ctr = get_ctr(arr.dtype) const bytesPerElement = (Ctr as unknown as { BYTES_PER_ELEMENT: number }) .BYTES_PER_ELEMENT - // The probe's fetches run under `storeOpts`, so its abort detection has - // to watch the same signal — otherwise its catch-alls would swallow an - // abort as a store failure and hand back the fallback shape. - const signal = (storeOpts as { signal?: AbortSignal } | undefined)?.signal + // 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, - storeOpts, - signal, + shared, ) return { - info: { + // 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, } })() @@ -913,7 +1011,10 @@ export function resolveArrayInfo( if (!conclusive) forget() }, forget) - return promise.then(({ info }) => info) + return untilAborted( + promise.then(({ info }) => detach(info)), + signal, + ) } // --------------------------------------------------------------------------- @@ -1022,8 +1123,8 @@ export async function getWorker< // 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. - // Runs under `storeOpts`, so the store reads it makes carry the combined - // signal. + // 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, diff --git a/test/node/fizarrita.test.js b/test/node/fizarrita.test.js index 23e10e2..820c760 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' @@ -596,11 +597,178 @@ test('store options reach the metadata reads, not just the probe', async () => { // 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. + */ +class GatedStore extends CountingStore { + #gate = null + #open + #release + constructor() { + super() + this.#open = new Promise((resolve) => { + this.#release = 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 wait = this.#gate?.test(key) ? 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 } }) + // Let both reach the store before pulling the plug on the first. + await new Promise((r) => setTimeout(r, 20)) + 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 new Promise((r) => setTimeout(r, 20)) + 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 new Promise((r) => setTimeout(r, 20)) + 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: the next caller + // pays no metadata read at all. + await new Promise((r) => setTimeout(r, 20)) + 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('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.deepEqual(second.codecMeta.codecs, []) + 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 From ab225eca27172a6e81717e15b8f30ab7e113e174 Mon Sep 17 00:00:00 2001 From: Matt McCormick Date: Tue, 18 Aug 2026 21:41:31 -0400 Subject: [PATCH 5/7] docs(fizarrita): say where a read's signal does and does not go MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The cancellation section said the signal is forwarded to every `store.get` call when it fires. Neither half held: the signal is placed in the store options when the read starts, before any request goes out, and it reaches only the chunk fetches — the shared metadata read and chunk-shape probe run without it, as the caching section already said two paragraphs down. Reworded to match, with a pointer to the exception. Raised by CodeRabbit in review of #14. Co-Authored-By: Claude Fable 5 --- fizarrita/README.md | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/fizarrita/README.md b/fizarrita/README.md index be6bbf1..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 From fee2137cd4a1666bdc1b6d30e21f38a1b218c3d7 Mon Sep 17 00:00:00 2001 From: Matt McCormick Date: Tue, 18 Aug 2026 21:41:31 -0400 Subject: [PATCH 6/7] test(fizarrita): wait on the gated store instead of sleeping MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The abort-isolation tests slept 20 ms to guess when the shared read had reached the store, and again after release() to guess when the resolution had landed. GatedStore now exposes `entered`, resolved when the first gated read starts, and the tests await that; the post-release wait joins the memoised resolution through resolveArrayInfo instead, which returns exactly when it has settled. No timing left to get wrong. The copies test asserted the memo's codec list is empty, which is really an assertion about what zarrita writes for an uncompressed array; it now asserts what it means — that the injected codec did not reach the memo. Both raised by CodeRabbit in review of #14. Co-Authored-By: Claude Fable 5 --- test/node/fizarrita.test.js | 32 +++++++++++++++++++++++--------- 1 file changed, 23 insertions(+), 9 deletions(-) diff --git a/test/node/fizarrita.test.js b/test/node/fizarrita.test.js index 820c760..3a15cac 100644 --- a/test/node/fizarrita.test.js +++ b/test/node/fizarrita.test.js @@ -609,16 +609,24 @@ test('store options reach the metadata reads, not just the probe', async () => { * 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 @@ -634,7 +642,9 @@ class GatedStore extends CountingStore { signal?.addEventListener('abort', () => reject(signal.reason), { once: true, }) - const wait = this.#gate?.test(key) ? this.#open : Promise.resolve() + const gated = this.#gate?.test(key) ?? false + if (gated) this.#entered() + const wait = gated ? this.#open : Promise.resolve() wait.then(() => resolve(bytes)) }) } @@ -664,8 +674,9 @@ test('aborting one caller does not fail the callers sharing its metadata resolut const second = new AbortController() const a = getWorker(arr, null, { pool, opts: { signal: first.signal } }) const b = getWorker(arr, null, { pool, opts: { signal: second.signal } }) - // Let both reach the store before pulling the plug on the first. - await new Promise((r) => setTimeout(r, 20)) + // 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 @@ -701,7 +712,7 @@ test('aborting one caller does not hand the callers sharing its probe an unprobe const first = new AbortController() const a = getWorker(misdeclared, null, { pool, opts: { signal: first.signal } }) const b = getWorker(misdeclared, null, { pool }) - await new Promise((r) => setTimeout(r, 20)) + await store.entered first.abort() await assert.rejects(a, (err) => err?.name === 'AbortError') @@ -723,16 +734,16 @@ test('a lone caller that aborts is rejected promptly, and the resolution still l await withPool(2, async (pool) => { const controller = new AbortController() const a = getWorker(arr, null, { pool, opts: { signal: controller.signal } }) - await new Promise((r) => setTimeout(r, 20)) + 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: the next caller - // pays no metadata read at all. - await new Promise((r) => setTimeout(r, 20)) + // 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)) @@ -753,7 +764,10 @@ test('resolveArrayInfo hands out copies — mutating one cannot reach the memo o const second = await resolveArrayInfo(arr) assert.notEqual(second.codecMeta, first.codecMeta) assert.deepEqual(second.codecMeta.chunk_shape, [4, 4]) - assert.deepEqual(second.codecMeta.codecs, []) + 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. From b67f15bc2af5c8dcb685860cfe1d310fd9470fa3 Mon Sep 17 00:00:00 2001 From: Matt McCormick Date: Tue, 18 Aug 2026 21:41:31 -0400 Subject: [PATCH 7/7] fix(fizarrita): observe the resolution a caller had already aborted out of MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit untilAborted rejects an already-aborted caller on the spot — right, but it returned without attaching anything to the promise it was handed. Both call sites in resolveArrayInfo hand it a fresh derived promise (`promise.then(({ info }) => detach(info))`), so nothing else observed it, and if the shared resolution then failed, that derived promise rejected with no handler: an unhandled rejection, fatal under Node's default. Reachable from getWorker: only `opts.signal` is pre-checked, so a store-level signal in `opts.opts` that is already aborted reaches resolveArrayInfo aborted, and a transient metadata or probe failure after that would have taken the process down. The early branch now absorbs the promise's outcome before rejecting the caller. Other observers of the chain are unaffected — a `.catch` on one derived promise does not swallow the rejection for anyone else — and the declined resolution still runs to completion for the memo, as documented. Regression test: an already-aborted store-level signal, a gated metadata read that fails once it is released, and an `unhandledRejection` listener that must stay empty; then a fresh read that succeeds, since the failure was not memoised. Fails against the previous build with the raw store error surfacing as an unhandled rejection. Raised by CodeRabbit in review of #14. Co-Authored-By: Claude Fable 5 --- fizarrita/src/get-worker.ts | 13 +++++++++- test/node/fizarrita.test.js | 49 +++++++++++++++++++++++++++++++++++++ 2 files changed, 61 insertions(+), 1 deletion(-) diff --git a/fizarrita/src/get-worker.ts b/fizarrita/src/get-worker.ts index da10f8e..1762ec3 100644 --- a/fizarrita/src/get-worker.ts +++ b/fizarrita/src/get-worker.ts @@ -870,6 +870,11 @@ function separateSignal(storeOpts: Opts): { * 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, @@ -878,7 +883,13 @@ function untilAborted( if (!signal) return promise const reason = () => signal.reason ?? new DOMException("The operation was aborted.", "AbortError") - if (signal.aborted) return Promise.reject(reason()) + 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 }) diff --git a/test/node/fizarrita.test.js b/test/node/fizarrita.test.js index 3a15cac..52c9056 100644 --- a/test/node/fizarrita.test.js +++ b/test/node/fizarrita.test.js @@ -751,6 +751,55 @@ test('a lone caller that aborts is rejected promptly, and the resolution still l }) }) +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()