From f9c700cccd0eb9f47c8c2857b826545762ddfc09 Mon Sep 17 00:00:00 2001 From: Okiki Ojo Date: Sun, 16 Aug 2026 05:29:47 -0400 Subject: [PATCH] refactor(adapter): streamline adapter validation and method checks Signed-off-by: Okiki Ojo --- src/adapter/definition.ts | 51 ++++++++++++++++++--------------------- src/adapter/deno-kv.ts | 20 +++++++-------- src/adapter/record.ts | 11 +++------ src/azure.ts | 15 ++++++------ src/filesystem.ts | 18 +++++--------- src/metrics.ts | 8 ++++++ src/request.ts | 19 +++++++++------ src/s3.ts | 25 ++++++++----------- 8 files changed, 80 insertions(+), 87 deletions(-) diff --git a/src/adapter/definition.ts b/src/adapter/definition.ts index 109e026..8626d6e 100644 --- a/src/adapter/definition.ts +++ b/src/adapter/definition.ts @@ -238,38 +238,33 @@ export interface FileSystemOptionsType { * ``` */ export function defineAdapter(adapter: T): T { - try { - AdapterNameSchema.parse(adapter.name); - AdapterCapabilitiesSchema.parse(adapter.capabilities); - if (adapter.limits !== undefined) AdapterLimitsSchema.parse(adapter.limits); - if (adapter.partition !== undefined) AdapterPartitionSchema.parse(adapter.partition); + AdapterNameSchema.parse(adapter.name); + AdapterCapabilitiesSchema.parse(adapter.capabilities); + if (adapter.limits !== undefined) AdapterLimitsSchema.parse(adapter.limits); + if (adapter.partition !== undefined) AdapterPartitionSchema.parse(adapter.partition); - for (const name of ["stat", "readFile", "writeFile", "readDir", "createDir", "remove"] as const) { - if (typeof adapter[name] !== "function") { - throw new TypeError(`Adapter '${adapter.name}' is missing required method '${name}'.`); - } + for (const name of ["stat", "readFile", "writeFile", "readDir", "createDir", "remove"] as const) { + if (typeof adapter[name] !== "function") { + throw new TypeError(`Adapter '${adapter.name}' is missing required method '${name}'.`); } + } - const pairs = [ - ["streamRead", adapter.capabilities.streamRead, adapter.openReadStream !== undefined], - ["nativeCopy", adapter.capabilities.nativeCopy, adapter.copy !== undefined], - ["nativeMove", adapter.capabilities.nativeMove, adapter.move !== undefined], - ["positionalWrite", adapter.capabilities.positionalWrite, adapter.openWritableFile !== undefined], - ["syncAccess", adapter.capabilities.syncAccess, adapter.openSyncFile !== undefined], - ] as const; - for (const [name, capability, method] of pairs) { - if (capability && !method) { - throw new TypeError(`Adapter '${adapter.name}' capability '${name}' does not match its implementation method.`); - } - } - if (adapter.capabilities.streamWriteModes.length > 0 && adapter.writeStream === undefined) { - throw new TypeError( - `Adapter '${adapter.name}' streamWriteModes do not match its writeStream implementation.`, - ); + const pairs = [ + ["streamRead", adapter.capabilities.streamRead, adapter.openReadStream !== undefined], + ["nativeCopy", adapter.capabilities.nativeCopy, adapter.copy !== undefined], + ["nativeMove", adapter.capabilities.nativeMove, adapter.move !== undefined], + ["positionalWrite", adapter.capabilities.positionalWrite, adapter.openWritableFile !== undefined], + ["syncAccess", adapter.capabilities.syncAccess, adapter.openSyncFile !== undefined], + ] as const; + for (const [name, capability, method] of pairs) { + if (capability && !method) { + throw new TypeError(`Adapter '${adapter.name}' capability '${name}' does not match its implementation method.`); } - } catch (error) { - if (error instanceof TypeError) throw error; - throw new TypeError(error instanceof Error ? error.message : String(error)); + } + if (adapter.capabilities.streamWriteModes.length > 0 && adapter.writeStream === undefined) { + throw new TypeError( + `Adapter '${adapter.name}' streamWriteModes do not match its writeStream implementation.`, + ); } return adapter; } diff --git a/src/adapter/deno-kv.ts b/src/adapter/deno-kv.ts index cfbd384..c04ba28 100644 --- a/src/adapter/deno-kv.ts +++ b/src/adapter/deno-kv.ts @@ -1,5 +1,3 @@ -/// - import { pooledMap } from "@std/async/pool"; import { concat } from "@std/bytes"; import { decodeBase64, encodeBase64 } from "@std/encoding/base64"; @@ -44,13 +42,13 @@ export interface DenoKvEntryType { /** Structural Deno KV subset required by this adapter. */ export interface DenoKvType { /** Reads one exact key. */ - get(key: Deno.KvKey): Promise>; + get(key: readonly unknown[]): Promise>; /** Replaces one key. */ - set(key: Deno.KvKey, value: unknown): Promise; + set(key: readonly unknown[], value: unknown): Promise; /** Removes one key. */ - delete(key: Deno.KvKey): Promise; - /** Streams keys with one prefix through the native Deno KV iterator surface. */ - list(selector: Deno.KvListSelector, options?: Deno.KvListOptions): AsyncIterable>; + delete(key: readonly unknown[]): Promise; + /** Streams keys with one prefix. */ + list(selector: { prefix: readonly unknown[] }, options?: unknown): AsyncIterable>; /** Closes the database when the caller transfers ownership. */ close?(): void; } @@ -96,21 +94,23 @@ const DenoKvManifestSchema = z.object({ file: DenoKvFileSchema, }).strict(); +/** Validated private manifest that publishes one complete partition generation. */ type DenoKvManifestType = z.output; +/** Physical value stored at one logical entry key: inline record or partition manifest. */ type DenoKvStoredType = RecordType | DenoKvManifestType; /** Maps one exact virtual path to a Deno KV entry key derived from its parent and name. */ -function key(prefix: string, path: string): Deno.KvKey { +function key(prefix: string, path: string): readonly unknown[] { return [prefix, "entry", dirname(path), basename(path)]; } /** Prefix whose entries are exactly the direct children of one canonical parent path. */ -function listKey(prefix: string, parent: string): Deno.KvKey { +function listKey(prefix: string, parent: string): readonly unknown[] { return [prefix, "entry", parent]; } /** Maps one logical file generation and part number to a separate raw binary key. */ -function partKey(prefix: string, path: string, generation: string, index: number): Deno.KvKey { +function partKey(prefix: string, path: string, generation: string, index: number): readonly unknown[] { return [prefix, "part", path, generation, index]; } diff --git a/src/adapter/record.ts b/src/adapter/record.ts index 2aa100a..ed110c5 100644 --- a/src/adapter/record.ts +++ b/src/adapter/record.ts @@ -112,11 +112,6 @@ function applyWrite( return output; } -/** Narrows a mixed record-store result to a file record that still carries bytes. */ -function isFileRecord(record: RecordListType | RecordType | null): record is FileRecordType { - return record?.kind === "file" && "data" in record; -} - /** Fails before touching a record store when the adapter was intentionally opened read-only. */ function assertWritable(readOnly: boolean, operation: string, path: string): void { if (readOnly) { @@ -236,9 +231,9 @@ class RecordAdapter implements AdapterType { const existing = options.mode === "replace" ? new Uint8Array() - : isFileRecord(previous) - ? decodeBase64(previous.data) - : new Uint8Array(); + : previous?.kind === "file" && "data" in previous && typeof previous.data === "string" + ? decodeBase64(previous.data) + : new Uint8Array(); const bytes = applyWrite(existing, data, options.mode, options.at, options.truncate ?? false); await this.#store.set(RecordSchema.parse({ version: 1, diff --git a/src/azure.ts b/src/azure.ts index f556c10..0f144af 100644 --- a/src/azure.ts +++ b/src/azure.ts @@ -330,11 +330,6 @@ function getBodyLength(body: BodyInit | null | undefined): number | undefined { return undefined; } -/** Converts one byte buffer into an owned Fetch body without shared backing state. */ -function getRequestBody(bytes: Uint8Array): ArrayBuffer { - return Uint8Array.from(bytes).buffer; -} - /** Returns the service-version-specific Content-Length field used by Shared Key signing. */ function getSignedContentLength(headers: Headers, version: AzureStorageVersionType): string { const value = headers.get("content-length") ?? ""; @@ -751,7 +746,7 @@ class AzureClient implements AzureClientType { key, query: { comp: "block", blockid: block.id }, headers: { "content-type": "application/octet-stream" }, - body: getRequestBody(block.bytes), + body: Uint8Array.from(block.bytes), ...(signal === undefined ? {} : { signal }), }), `Put Block ${key}#${block.number}`, @@ -802,7 +797,13 @@ class AzureClient implements AzureClientType { const headers = this.#getWriteHeaders(options); headers.set("x-ms-blob-type", "BlockBlob"); await assertResponse( - await this.request({ method: "PUT", key, headers, body: getRequestBody(body), ...(options.signal === undefined ? {} : { signal: options.signal }) }), + await this.request({ + method: "PUT", + key, + headers, + body: Uint8Array.from(body), + ...(options.signal === undefined ? {} : { signal: options.signal }) + }), `Put Blob ${key}`, ); return (await this.head(key, options)) ?? { size: body.byteLength }; diff --git a/src/filesystem.ts b/src/filesystem.ts index 9cc1c7b..11f5981 100644 --- a/src/filesystem.ts +++ b/src/filesystem.ts @@ -310,16 +310,6 @@ function getOptimizations(value: FileSystemOptionsType["optimizations"]): Optimi return OptimizationSchema.parse({ ...DEFAULT_OPTIMIZATIONS, ...value }); } -/** Parses one public enum-like option and normalizes schema failures to TypeError. */ -function getValidatedOption(parse: () => T): T { - try { - return parse(); - } catch (error) { - if (error instanceof TypeError) throw error; - throw new TypeError(error instanceof Error ? error.message : String(error)); - } -} - /** Computes the logical file size produced by one materialized write. */ function getWriteSize( current: number, @@ -470,10 +460,10 @@ class FileSystemFacade implements FileSystemType { this.adapter = adapter; this.maxBufferedWriteBytes = getBufferLimit(options.maxBufferedWriteBytes); this.optimizations = getOptimizations(options.optimizations); - this.metricsMode = getValidatedOption(() => MetricsModeSchema.parse(options.metrics ?? "basic")); + this.metricsMode = MetricsModeSchema.parse(options.metrics ?? "basic"); this.#metrics = new Metrics(this.metricsMode); this.#locks = new MutationLocks( - getValidatedOption(() => CoordinationModeSchema.parse(options.coordination ?? "auto")), + CoordinationModeSchema.parse(options.coordination ?? "auto"), options.lockPrefix ?? DEFAULT_LOCK_PREFIX, ); this.#disposeAdapter = options.disposeAdapter ?? false; @@ -953,6 +943,10 @@ class FileSystemFacade implements FileSystemType { metricBytes = bytes.byteLength; buffered = bytes.byteLength; this.#metrics.buffer(buffered); + metricSupport = this.#support( + metricSupport, + getWriteSize(currentSize, bytes.byteLength, mode, options.at, options.truncate ?? false), + ); await this.adapter.writeFile(normalized, bytes, adapterOptions); } else { const bytes = await toBytes(data); diff --git a/src/metrics.ts b/src/metrics.ts index 358787c..8e8a11e 100644 --- a/src/metrics.ts +++ b/src/metrics.ts @@ -50,13 +50,21 @@ export interface MetricsType { /** Mutable form retained privately so snapshots cannot mutate live counters. */ interface MutableMetricType { + /** Completed attempts, including failures. */ count: number; + /** Attempts that reached a terminal failure. */ failures: number; + /** Logical bytes attributed to this operation at this layer. */ bytes: number; + /** Attempts routed through a backend-native operation. */ native: number; + /** Attempts routed through a portable facade fallback. */ emulated: number; + /** Attempts whose logical value used a partitioned physical representation. */ partitioned: number; + /** Accumulated measured duration when timing mode is enabled. */ durationMs: number; + /** Longest measured attempt when timing mode is enabled. */ maxDurationMs: number; } diff --git a/src/request.ts b/src/request.ts index 378033d..eae2a2a 100644 --- a/src/request.ts +++ b/src/request.ts @@ -15,9 +15,9 @@ export const RequestPolicySchema = z.object({ /** Maximum retry delay in milliseconds. Defaults to 20 seconds. */ maxDelayMs: z.number().int().nonnegative().optional(), /** Exponential delay multiplier. Defaults to 2. */ - multiplier: z.number().finite().min(1).optional(), + multiplier: z.number().min(1).optional(), /** Random delay proportion accepted by `@std/async/retry`. Defaults to 0.5. */ - jitter: z.number().finite().min(0).max(1).optional(), + jitter: z.number().min(0).max(1).optional(), /** Per-attempt deadline in milliseconds. `false` or omission leaves Fetch's own timeout policy unchanged. */ timeoutMs: z.union([z.number().int().positive(), z.literal(false)]).optional(), }).strict(); @@ -59,10 +59,15 @@ export interface RequestMetricsType { export class RequestMetrics { /** Whether monotonic duration is measured. */ readonly #timing: boolean; + /** Concrete Fetch attempts, including retries. */ #requests = 0; + /** Fetch attempts made after the first attempt for one logical request. */ #retries = 0; + /** Logical requests that exhausted retry policy or were canceled. */ #failures = 0; + /** HTTP responses received, including service error status codes. */ #responses = 0; + /** Accumulated Fetch wall-clock time when timing is enabled. */ #durationMs = 0; /** Enables timing only when the caller explicitly requests it. */ @@ -223,13 +228,13 @@ export async function sendRequest( } = {}, ): Promise { const policy = getRequestPolicy(options.policy); - const attempts = options.replayable === false ? 1 : policy.retries + 1; + const attempts = options.replayable === false ? 1 : (policy.retries ?? 0) + 1; let attempt = 0; let lastStarted = 0; try { - const minTimeout = Math.max(1, policy.minDelayMs); - const maxTimeout = Math.max(minTimeout, policy.maxDelayMs); + const minTimeout = Math.max(1, policy.minDelayMs ?? 0); + const maxTimeout = Math.max(minTimeout, policy.maxDelayMs ?? 0); return await retry(async () => { attempt += 1; const scoped = getSignal(options.signal, policy.timeoutMs); @@ -258,8 +263,8 @@ export async function sendRequest( maxAttempts: attempts, minTimeout, maxTimeout, - multiplier: policy.multiplier, - jitter: policy.jitter, + ...(policy.multiplier === undefined ? {} : { multiplier: policy.multiplier }), + ...(policy.jitter === undefined ? {} : { jitter: policy.jitter }), ...(options.signal === undefined ? {} : { signal: options.signal }), isRetriable: (error: unknown) => error instanceof RetryResponseError || error instanceof RequestTransportError, }); diff --git a/src/s3.ts b/src/s3.ts index 95dfaec..181353e 100644 --- a/src/s3.ts +++ b/src/s3.ts @@ -307,23 +307,18 @@ async function getPayloadHash(body: BodyInit | null | undefined): Promise { +async function getHmac(key: BufferSource, value: string): Promise { const cryptoKey = await crypto.subtle.importKey("raw", key, { name: "HMAC", hash: "SHA-256" }, false, ["sign"]); - return await crypto.subtle.sign("HMAC", cryptoKey, textEncoder.encode(value)); + return new Uint8Array(await crypto.subtle.sign("HMAC", cryptoKey, textEncoder.encode(value))); } /** Derives the date, region, and service-specific Signature Version 4 signing key. */ -async function getSigningKey(secret: string, date: string, region: string): Promise { +async function getSigningKey(secret: string, date: string, region: string): Promise { const dateKey = await getHmac(textEncoder.encode(`AWS4${secret}`), date); - const regionKey = await getHmac(dateKey, region); - const serviceKey = await getHmac(regionKey, "s3"); - return await getHmac(serviceKey, "aws4_request"); + const regionKey = await getHmac(Uint8Array.from(dateKey), region); + const serviceKey = await getHmac(Uint8Array.from(regionKey), "s3"); + return await getHmac(Uint8Array.from(serviceKey), "aws4_request"); } /** Formats one UTC instant as the compact timestamp required by Signature Version 4. */ @@ -390,7 +385,7 @@ async function getSuccessXml(response: Response, operation: string) { const body = await response.text(); if (!body.trim().startsWith("<")) return undefined; const root = parseXmlRoot(body); - const error = root.name.local === "Error" ? root : getXmlElements(root, "Error")[0]; + const error = getXmlElements(root, "Error")[0]; if (error === undefined) return root; throw new S3Error(getXmlValue(error, "Message") ?? `${operation} failed after HTTP 200.`, response, { @@ -617,7 +612,7 @@ class S3Client implements S3ClientType { for (const [name, value] of Object.entries(options.metadata ?? {})) headers.set(`x-amz-meta-${name}`, value); await assertResponse( - await this.request({ method: "PUT", key, headers, body: getRequestBody(body), ...(options.signal === undefined ? {} : { signal: options.signal }) }), + await this.request({ method: "PUT", key, headers, body: Uint8Array.from(body), ...(options.signal === undefined ? {} : { signal: options.signal }) }), `PutObject ${key}`, ); return (await this.head(key, options)) ?? { size: body.byteLength }; @@ -717,7 +712,7 @@ class S3Client implements S3ClientType { const scope = `${shortDate}/${this.#region}/s3/aws4_request`; const stringToSign = `AWS4-HMAC-SHA256\n${timestamp}\n${scope}\n${await getSha256(canonicalRequest)}`; const signingKey = await getSigningKey(credentials.secretAccessKey, shortDate, this.#region); - const signature = encodeHex(new Uint8Array(await getHmac(signingKey, stringToSign))).toLowerCase(); + const signature = encodeHex(await getHmac(Uint8Array.from(signingKey), stringToSign)).toLowerCase(); headers.set( "authorization", `AWS4-HMAC-SHA256 Credential=${credentials.accessKeyId}/${scope}, SignedHeaders=${signedHeaders}, Signature=${signature}`, @@ -796,7 +791,7 @@ class S3Client implements S3ClientType { method: "PUT", key: upload.key, query: { partNumber: String(number), uploadId: upload.id }, - body: getRequestBody(bytes), + body: Uint8Array.from(bytes), ...(signal === undefined ? {} : { signal }), }), `UploadPart ${upload.key}#${number}`, -- 2.51.2