Drop Options generic and thread signal directly

`AsyncReadable<Options>` existed so stores could receive arbitrary
per-call state. Auth headers, presigning context, cancellation signals,
even chunk-layer concerns like caching and prefetch priority (#296,
vole-core's `wrapArray`).

It was a catch-all for extensions that had nowhere else to live, and the
cost was a pile of type magic: higher-kinded-type encoding in the store
middleware system (#384), threading generic types that didn't actually
provide that much type safety. (TypeScript could often bail out to `any`
when inference broke.)

Those extensions now have proper homes. Store middleware (#384) gives
transport-layer concerns (auth, presigning, request transformation) a
proper extension point, and the custom `fetch` option on `FetchStore`
(#388) handles the per-store cases at the callsite. Chunk-layer concerns
will move to `zarr.extendArray` in a follow-up.

What's left is `signal`, which now lives properly on the (non-generic)
`AsyncReadable` interface and can be passed directly in `zarr.get` and
`zarr.set` from the caller:

    // Before
    interface AsyncReadable<Options = unknown> {
      get(key: AbsolutePath, opts?: Options): Promise<Uint8Array | undefined>;
    }
    await zarr.get(arr, null, { opts: { signal: ctl.signal } });

    // After
    interface AsyncReadable {
      get(key: AbsolutePath, opts?: { signal?: AbortSignal }): Promise<Uint8Array | undefined>;
    }
    await zarr.get(arr, null, { signal: ctl.signal });

Batched caller signals in `withRangeBatching` are now merged with
`AbortSignal.any` instead of a user-supplied `mergeOptions` reducer. The
deprecated `opts?: { signal? }` shape still works for one major version
and is folded into the new `signal` via `AbortSignal.any`.

Author: manzt

packages/@zarrita-ndarray/src/index.ts

@@ -49,7 +49,7 @@ export async function get<
 >(
 	arr: zarr.Array<D, Store>,
 	selection: Sel | null = null,
-	opts: zarr.GetOptions<Parameters<Store["get"]>[1]> = {},
+	opts: zarr.GetOptions = {},
 ): Promise<
 	null extends Sel[number]
 		? ndarray.NdArray<zarr.TypedArray<D>>

packages/@zarrita-storage/src/fetch.ts

@@ -131,7 +131,7 @@ interface FetchStoreOptions {
  * });
  * ```
  */
-class FetchStore implements AsyncReadable<RequestInit> {
+class FetchStore implements AsyncReadable {
 	#fetch: (request: Request) => Promise<Response>;
 	#overrides: RequestInit;
 	#useSuffixRequest: boolean;

packages/@zarrita-storage/src/fs.ts

@@ -2,7 +2,12 @@ import { Buffer } from "node:buffer";
 import * as fs from "node:fs";
 import * as path from "node:path";
 
-import type { AbsolutePath, AsyncMutable, RangeQuery } from "./types.js";
+import type {
+	AbsolutePath,
+	AsyncMutable,
+	GetOptions,
+	RangeQuery,
+} from "./types.js";
 import { stripPrefix } from "./util.js";
 
 function isErrorNoEntry(err: unknown): err is { code: "ENOENT" } {
@@ -13,18 +18,26 @@ function isErrorNoEntry(err: unknown): err is { code: "ENOENT" } {
 class FileSystemStore implements AsyncMutable {
 	constructor(public root: string) {}
 
-	async get(key: AbsolutePath): Promise<Uint8Array | undefined> {
+	async get(
+		key: AbsolutePath,
+		opts: GetOptions = {},
+	): Promise<Uint8Array | undefined> {
+		opts.signal?.throwIfAborted();
 		let fp = path.join(this.root, stripPrefix(key));
-		return fs.promises.readFile(fp).catch((err: NodeJS.ErrnoException) => {
-			if (err.code === "ENOENT") return undefined;
-			throw err;
-		});
+		return fs.promises
+			.readFile(fp, { signal: opts.signal })
+			.catch((err: NodeJS.ErrnoException) => {
+				if (err.code === "ENOENT") return undefined;
+				throw err;
+			});
 	}
 
 	async getRange(
 		key: AbsolutePath,
 		range: RangeQuery,
+		opts: GetOptions = {},
 	): Promise<Uint8Array | undefined> {
+		opts.signal?.throwIfAborted();
 		let fp = path.join(this.root, stripPrefix(key));
 		let filehandle: fs.promises.FileHandle | undefined;
 		try {
@@ -38,10 +51,12 @@ class FileSystemStore implements AsyncMutable {
 					range.suffixLength,
 					stats.size - range.suffixLength,
 				);
+				opts.signal?.throwIfAborted();
 				return data;
 			}
 			let data = Buffer.alloc(range.length);
 			await filehandle.read(data, 0, range.length, range.offset);
+			opts.signal?.throwIfAborted();
 			return data;
 		} catch (err: unknown) {
 			// return undefined is no file or directory

packages/@zarrita-storage/src/ref.ts

@@ -163,7 +163,7 @@ function parseReferencesJson(refsJson: unknown): Map<string, ResolvedEntry> {
  *
  * @experimental
  */
-class ReferenceStore implements AsyncReadable<RequestInit> {
+class ReferenceStore implements AsyncReadable {
 	#inner: FetchStore;
 
 	constructor(

packages/@zarrita-storage/src/types.ts

@@ -9,23 +9,25 @@ export type RangeQuery =
 			suffixLength: number;
 	  };
 
-export type Readable<GetOptions = unknown> =
-	| AsyncReadable<GetOptions>
-	| SyncReadable<GetOptions>;
-export interface AsyncReadable<Options = unknown> {
-	get(key: AbsolutePath, opts?: Options): Promise<Uint8Array | undefined>;
+export interface GetOptions {
+	signal?: AbortSignal;
+}
+
+export type Readable = AsyncReadable | SyncReadable;
+export interface AsyncReadable {
+	get(key: AbsolutePath, opts?: GetOptions): Promise<Uint8Array | undefined>;
 	getRange?(
 		key: AbsolutePath,
 		range: RangeQuery,
-		opts?: Options,
+		opts?: GetOptions,
 	): Promise<Uint8Array | undefined>;
 }
-export interface SyncReadable<Options = unknown> {
-	get(key: AbsolutePath, opts?: Options): Uint8Array | undefined;
+export interface SyncReadable {
+	get(key: AbsolutePath, opts?: GetOptions): Uint8Array | undefined;
 	getRange?(
 		key: AbsolutePath,
 		range: RangeQuery,
-		opts?: Options,
+		opts?: GetOptions,
 	): Uint8Array | undefined;
 }
 
@@ -37,10 +39,6 @@ export interface SyncWritable {
 	set(key: AbsolutePath, value: Uint8Array): void;
 }
 
-export type AsyncMutable<GetOptions = unknown> = AsyncReadable<GetOptions> &
-	AsyncWritable;
-export type SyncMutable<GetOptions = unknown> = SyncReadable<GetOptions> &
-	SyncWritable;
-export type Mutable<GetOptions = unknown> =
-	| AsyncMutable<GetOptions>
-	| SyncMutable<GetOptions>;
+export type AsyncMutable = AsyncReadable & AsyncWritable;
+export type SyncMutable = SyncReadable & SyncWritable;
+export type Mutable = AsyncMutable | SyncMutable;

packages/zarrita/__tests__/batched-fetch.test.ts

@@ -306,97 +306,44 @@ describe("withRangeBatching", () => {
 		});
 	});
 
-	describe("mergeOptions", () => {
-		it("uses first caller's options by default", async () => {
+	describe("signal merging", () => {
+		it("merges caller signals via AbortSignal.any across a batch", async () => {
 			let inner = fakeStore();
 			let store = withRangeBatching(inner);
 
+			let a = new AbortController();
+			let b = new AbortController();
 			await Promise.all([
 				store.getRange(
 					"/data/chunk",
 					{ offset: 0, length: 100 },
-					{ headers: { "x-req": "first" } },
+					{ signal: a.signal },
 				),
 				store.getRange(
 					"/data/chunk",
 					{ offset: 100, length: 100 },
-					{ headers: { "x-req": "second" } },
+					{ signal: b.signal },
 				),
 			]);
 
-			expect(inner.getRange.mock.calls[0][2]?.headers).toEqual({
-				"x-req": "first",
-			});
+			let passedSignal = inner.getRange.mock.calls[0][2]?.signal;
+			expect(passedSignal).toBeInstanceOf(AbortSignal);
+			// Aborting either upstream signal should propagate to the merged one.
+			a.abort(new Error("a aborted"));
+			expect(passedSignal?.aborted).toBe(true);
 		});
 
-		it("rejects pending requests and recovers when mergeOptions throws", async () => {
+		it("passes a single signal through unchanged", async () => {
 			let inner = fakeStore();
-			let shouldThrow = true;
-			let store = withRangeBatching(inner, {
-				mergeOptions: () => {
-					if (shouldThrow) throw new Error("bad merge");
-					return undefined;
-				},
-			});
+			let store = withRangeBatching(inner);
+			let ctl = new AbortController();
 
-			// First batch: mergeOptions throws, all requests should reject
-			let results = await Promise.allSettled([
-				store.getRange("/data/chunk", { offset: 0, length: 100 }),
-				store.getRange("/data/chunk", { offset: 100, length: 100 }),
-			]);
-			expect(results[0].status).toBe("rejected");
-			expect(results[1].status).toBe("rejected");
-			expect((results[0] as PromiseRejectedResult).reason.message).toBe(
-				"bad merge",
+			await store.getRange(
+				"/data/chunk",
+				{ offset: 0, length: 100 },
+				{ signal: ctl.signal },
 			);
-
-			// Second batch: store should not be deadlocked
-			shouldThrow = false;
-			let r = await store.getRange("/data/chunk", { offset: 0, length: 100 });
-			expect(r?.length).toBe(100);
-		});
-
-		it("applies mergeOptions reducer across batched callers", async () => {
-			interface TaggedOptions {
-				tags: string[];
-			}
-			let inner = {
-				get: vi.fn((_key: AbsolutePath, _opts?: TaggedOptions) =>
-					Promise.resolve<Uint8Array | undefined>(new Uint8Array(0)),
-				),
-				getRange: vi.fn(
-					(
-						_key: AbsolutePath,
-						range: RangeQuery,
-						_options?: TaggedOptions,
-					): Promise<Uint8Array | undefined> => {
-						if ("suffixLength" in range) {
-							return Promise.resolve(new Uint8Array(range.suffixLength));
-						}
-						return Promise.resolve(new Uint8Array(range.length));
-					},
-				),
-			};
-			let store = withRangeBatching(inner, {
-				mergeOptions: (batch) => ({
-					tags: batch.flatMap((o) => o?.tags ?? []),
-				}),
-			});
-
-			await Promise.all([
-				store.getRange(
-					"/data/chunk",
-					{ offset: 0, length: 100 },
-					{ tags: ["a"] },
-				),
-				store.getRange(
-					"/data/chunk",
-					{ offset: 100, length: 100 },
-					{ tags: ["b"] },
-				),
-			]);
-
-			expect(inner.getRange.mock.calls[0][2]?.tags).toEqual(["a", "b"]);
+			expect(inner.getRange.mock.calls[0][2]?.signal).toBe(ctl.signal);
 		});
 	});
 

packages/zarrita/__tests__/get-types.test.ts

@@ -162,7 +162,6 @@ describe("Chunk type per union dtype", () => {
 		expectType(chunk.data).toMatchInlineSnapshot(
 			`
 			| string[]
-				| unknown[]
 				| Int8Array<ArrayBufferLike>
 				| Int16Array<ArrayBufferLike>
 				| Int32Array<ArrayBufferLike>
@@ -177,6 +176,7 @@ describe("Chunk type per union dtype", () => {
 				| zarr.BoolArray<ArrayBufferLike>
 				| zarr.UnicodeStringArray<ArrayBufferLike>
 				| zarr.ByteStringArray<ArrayBufferLike>
+				| unknown[]
 		`,
 		);
 	});

packages/zarrita/__tests__/middleware-types.test.ts

@@ -10,20 +10,13 @@ describe("extendStore", () => {
 		expectType(store).toMatchInlineSnapshot(`Promise<zarr.FetchStore>`);
 	});
 
-	test("direct form in pipeline infers store options", () => {
+	test("direct form in pipeline", () => {
 		let store = zarr.extendStore(new zarr.FetchStore(""), (s) =>
-			zarr.withRangeBatching(s, {
-				mergeOptions: (batch) => {
-					expectType(batch).toMatchInlineSnapshot(
-						`ReadonlyArray<RequestInit | undefined>`,
-					);
-					return batch[0];
-				},
-			}),
+			zarr.withRangeBatching(s),
 		);
 		expectType(store).toMatchInlineSnapshot(`
 			Promise<
-				Required<AsyncReadable<RequestInit>> & {
+				Required<AsyncReadable> & {
 					stats: Readonly<zarr.RangeBatchingStats>;
 					url: string | URL;
 				}
@@ -36,12 +29,12 @@ describe("extendStore", () => {
 			return zarr.extendStore(
 				new zarr.FetchStore(""),
 				zarr.withConsolidation,
-				(s) => zarr.withRangeBatching(s, { mergeOptions: (batch) => batch[0] }),
+				(s) => zarr.withRangeBatching(s),
 			);
 		}
 		expectType(check).toMatchInlineSnapshot(`
 			() => Promise<
-				Required<AsyncReadable<RequestInit>> & {
+				Required<AsyncReadable> & {
 					stats: Readonly<zarr.RangeBatchingStats>;
 					url: string | URL;
 					contents: () => { path: AbsolutePath; kind: "array" | "group" }[];
@@ -68,46 +61,15 @@ describe("defineStoreMiddleware", () => {
 		let store = withCustom(new zarr.FetchStore(""), { flag: true });
 		expectType(store).toMatchInlineSnapshot(
 			`
-			Required<AsyncReadable<RequestInit>> & {
+			Required<AsyncReadable> & {
 				url: string | URL;
 				hello: () => string;
 			}
 		`,
 		);
 	});
 
-	test("generic: store Options flows into opts parameter", () => {
-		interface ThingOptions<O> {
-			storeOptions?: O;
-			retries?: number;
-		}
-
-		let withThing = defineStoreMiddleware(
-			<O>(store: AsyncReadable<O>, opts: ThingOptions<O>) => {
-				return {
-					async get(key: AbsolutePath, options?: O) {
-						return store.get(key, options ?? opts.storeOptions);
-					},
-					retries: opts.retries ?? 3,
-				};
-			},
-		);
-
-		let store = withThing(new zarr.FetchStore(""), {
-			storeOptions: { signal: AbortSignal.timeout(1000) },
-			retries: 5,
-		});
-		expectType(store).toMatchInlineSnapshot(
-			`
-			Required<AsyncReadable<RequestInit>> & {
-				url: string | URL;
-				retries: number;
-			}
-		`,
-		);
-	});
-
-	test("chaining preserves Options through wrappers", () => {
+	test("chaining preserves store through wrappers", () => {
 		let withA = defineStoreMiddleware(
 			(store: AsyncReadable, _opts: { a: number }) => {
 				return {
@@ -134,7 +96,7 @@ describe("defineStoreMiddleware", () => {
 		);
 		let store = withB(withA(new zarr.FetchStore(""), { a: 1 }), { b: "x" });
 		expectType(store).toMatchInlineSnapshot(`
-			Required<AsyncReadable<RequestInit>> & {
+			Required<AsyncReadable> & {
 				url: string | URL;
 				methodB: () => string;
 				methodA: () => number;

packages/zarrita/src/codecs/sharding.ts

@@ -1,13 +1,13 @@
-import type { Readable } from "@zarrita/storage";
+import type { GetOptions, Readable } from "@zarrita/storage";
 import { createCodecPipeline } from "../codecs.js";
 import type { Location } from "../hierarchy.js";
 import type { Chunk } from "../metadata.js";
 import { assert, type ShardingCodecMetadata } from "../util.js";
 
 const MAX_BIG_UINT = 18446744073709551615n;
 
-export function createShardedChunkGetter<Store extends Readable>(
-	location: Location<Store>,
+export function createShardedChunkGetter(
+	location: Location<Readable>,
 	shardShape: number[],
 	encodeShardKey: (coord: number[]) => string,
 	shardingConfig: ShardingCodecMetadata["configuration"],
@@ -24,10 +24,7 @@ export function createShardedChunkGetter<Store extends Readable>(
 	let checksumSize = 4;
 	let indexSize = 16 * indexShape.reduce((a, b) => a * b, 1);
 	let cache: Record<string, Promise<Chunk<"uint64"> | null>> = {};
-	return async (
-		chunkCoord: number[],
-		options?: Parameters<Store["get"]>[1],
-	) => {
+	return async (chunkCoord: number[], options?: GetOptions) => {
 		let shardCoord = chunkCoord.map((d, i) => Math.floor(d / indexShape[i]));
 		let shardPath = location.resolve(encodeShardKey(shardCoord)).path;