From eabcf7ce64433f6c517dffd1506abaa7669c8003 Mon Sep 17 00:00:00 2001 From: Okiki Ojo Date: Fri, 14 Aug 2026 01:44:13 -0400 Subject: [PATCH] feat: implement long-lived asynchronous positional writes with openWritableFile support Signed-off-by: Okiki Ojo --- mod.ts | 2 + src/adapter/definition.ts | 31 ++++++++- src/adapter/deno.ts | 38 ++++++++++ src/adapter/node.ts | 45 +++++++++++- src/adapter/opfs.ts | 39 ++++++++++- src/adapter/record.ts | 1 + src/filesystem.ts | 127 +++++++++++++++++++++++++++++++--- src/schema.ts | 1 + src/writable.ts | 142 ++++++++++++++++++++++++++++++++++++++ 9 files changed, 414 insertions(+), 12 deletions(-) create mode 100644 src/writable.ts diff --git a/mod.ts b/mod.ts index 391d6da..6f015d4 100644 --- a/mod.ts +++ b/mod.ts @@ -53,6 +53,7 @@ export type { MakeDirectoryOptionsType, MoveOptionsType, OpenSyncFileOptionsType, + OpenWritableFileOptionsType, ReadOptionsType, ReadTextOptionsType, RemoveOptionsType, @@ -73,6 +74,7 @@ export type { WriteCommandType, } from "./src/handle.ts"; export type { SyncFileType } from "./src/sync.ts"; +export type { WritableFileType } from "./src/writable.ts"; export type { WriteDataType } from "./src/stream.ts"; export type { AdapterCapabilitiesType, diff --git a/src/adapter/definition.ts b/src/adapter/definition.ts index a64c070..16e0f38 100644 --- a/src/adapter/definition.ts +++ b/src/adapter/definition.ts @@ -65,12 +65,36 @@ export interface AdapterDirectoryStatType { /** Portable entry metadata returned by an adapter. */ export type AdapterStatType = AdapterFileStatType | AdapterDirectoryStatType; +/** + * Long-lived asynchronous positional file owned by an adapter. + * + * This contract exists for callers such as media muxers and database engines + * that rewrite earlier byte ranges while a file stays open. It is deliberately + * separate from `writeFile()`, which represents one complete write operation. + * + * `abort()` may discard staged changes when the backend can do so. Backends + * without transactional staging still close the native resource, so callers + * that need rollback should write to a staging path and remove it after abort. + */ +export interface AdapterWritableFileType { + /** Writes bytes at one explicit zero-based file position. */ + write(buffer: ArrayBufferView, options: { readonly at: number }): Promise; + /** Changes current byte length. */ + truncate(size: number): Promise; + /** Requests backend durability without closing the file. */ + flush(): Promise; + /** Commits staged backend state where the backend uses staging, then closes. */ + close(): Promise; + /** Discards staged state when possible, then releases the native resource. */ + abort(reason?: unknown): Promise; +} + /** * Synchronous random-access file owned by an adapter. * * The adapter owns the native runtime object. The caller owns the returned - * resource and must call `close()`. `flush()` means "ask the backend to make - * current writes durable"; the exact storage guarantee remains backend-specific. + * resource and must call `close()`. `flush()` asks the backend to make current + * writes durable; the exact storage guarantee remains backend-specific. */ export interface AdapterSyncFileType { /** Reads bytes into `buffer` and returns the number of bytes read. */ @@ -124,6 +148,8 @@ export interface AdapterType { writeStream?(path: PathType, source: ReadableStream, options: AdapterWriteOptionsType): Promise; /** Performs an adapter-native move when `capabilities.nativeMove` is true. */ move?(source: PathType, destination: PathType, options: AdapterMoveOptionsType): Promise; + /** Opens long-lived asynchronous positional writes when `capabilities.positionalWrite` is true. */ + openWritableFile?(path: PathType): Promise; /** Opens synchronous random access when `capabilities.syncAccess` is true. */ openSyncFile?(path: PathType): Promise; /** Releases resources that this adapter explicitly owns. */ @@ -165,6 +191,7 @@ export interface FileSystemOptionsType { * streamWrite: false, * rangeRead: false, * nativeMove: false, + * positionalWrite: false, * syncAccess: false, * }, * async stat(path) { return null; }, diff --git a/src/adapter/deno.ts b/src/adapter/deno.ts index d9795fa..10eb404 100644 --- a/src/adapter/deno.ts +++ b/src/adapter/deno.ts @@ -42,6 +42,7 @@ export function createDenoAdapter(options: DenoAdapterOptionsType): AdapterType streamWrite: true, rangeRead: true, nativeMove: true, + positionalWrite: true, syncAccess: true, }, async stat(path, operationOptions) { @@ -169,6 +170,43 @@ export function createDenoAdapter(options: DenoAdapterOptionsType): AdapterType throwIfAborted(operationOptions.signal, "move", source); await Deno.rename(hostPath(source), hostPath(destination)); }, + async openWritableFile(path) { + const file = await Deno.open(hostPath(path), { read: true, write: true }); + let closed = false; + const getFile = () => { + if (closed) throw new Error(`Writable file '${path}' is closed.`); + return file; + }; + return { + async write(buffer, writeOptions) { + const source = new Uint8Array(buffer.buffer, buffer.byteOffset, buffer.byteLength); + const target = getFile(); + await target.seek(writeOptions.at, Deno.SeekMode.Start); + let offset = 0; + while (offset < source.byteLength) { + const count = await target.write(source.subarray(offset)); + if (count <= 0) throw new Error(`Deno positional write made no progress for '${path}'.`); + offset += count; + } + }, + async truncate(size) { + await getFile().truncate(size); + }, + async flush() { + await getFile().sync(); + }, + async close() { + if (closed) return; + closed = true; + file.close(); + }, + async abort() { + if (closed) return; + closed = true; + file.close(); + }, + }; + }, async openSyncFile(path) { const file = Deno.openSync(hostPath(path), { read: true, write: true }); let cursor = 0; diff --git a/src/adapter/node.ts b/src/adapter/node.ts index d74677b..542091e 100644 --- a/src/adapter/node.ts +++ b/src/adapter/node.ts @@ -129,6 +129,7 @@ export function createNodeAdapter(options: NodeAdapterOptionsType): AdapterType streamWrite: true, rangeRead: true, nativeMove: true, + positionalWrite: true, syncAccess: true, }, async stat(path, operationOptions) { @@ -173,7 +174,7 @@ export function createNodeAdapter(options: NodeAdapterOptionsType): AdapterType const start = readOptions.at ?? 0; const end = readOptions.length === undefined ? undefined : Math.max(start, start + readOptions.length - 1); const stream = createReadStream(hostPath(path), { start, ...(end === undefined ? {} : { end }) }); - return Readable.toWeb(stream, { type: "bytes" }) as unknown as ReadableStream; + return Readable.toWeb(stream) as unknown as ReadableStream; }, async writeFile(path, data, writeOptions) { throwIfAborted(writeOptions.signal, "write", path); @@ -228,6 +229,48 @@ export function createNodeAdapter(options: NodeAdapterOptionsType): AdapterType throwIfAborted(operationOptions.signal, "move", source); await rename(hostPath(source), hostPath(destination)); }, + async openWritableFile(path) { + const file = await open(hostPath(path), "r+"); + let closed = false; + const getFile = () => { + if (closed) throw new Error(`Writable file '${path}' is closed.`); + return file; + }; + return { + async write(buffer, writeOptions) { + const source = new Uint8Array(buffer.buffer, buffer.byteOffset, buffer.byteLength); + let offset = 0; + while (offset < source.byteLength) { + const result = await getFile().write( + source, + offset, + source.byteLength - offset, + writeOptions.at + offset, + ); + if (result.bytesWritten <= 0) { + throw new Error(`Node positional write made no progress for '${path}'.`); + } + offset += result.bytesWritten; + } + }, + async truncate(size) { + await getFile().truncate(size); + }, + async flush() { + await getFile().sync(); + }, + async close() { + if (closed) return; + closed = true; + await file.close(); + }, + async abort() { + if (closed) return; + closed = true; + await file.close(); + }, + }; + }, async openSyncFile(path) { const descriptor = openSync(hostPath(path), "r+"); let cursor = 0; diff --git a/src/adapter/opfs.ts b/src/adapter/opfs.ts index 124df3a..43ceb0e 100644 --- a/src/adapter/opfs.ts +++ b/src/adapter/opfs.ts @@ -4,6 +4,7 @@ import type { AdapterStatType, AdapterSyncFileType, AdapterType, + AdapterWritableFileType, AdapterWriteOptionsType, FileSystemOptionsType, } from "./definition.ts"; @@ -130,7 +131,7 @@ async function writeToNative( throwIfAborted(options.signal, "write", path); const next = await reader.read(); if (next.done) break; - await writable.write(next.value); + await writable.write(next.value as BufferSource); cursor += next.value.byteLength; } } catch (error) { @@ -194,6 +195,7 @@ export function createOpfsAdapter(root: FileSystemDirectoryHandle): OpfsAdapterT streamWrite: true, rangeRead: true, nativeMove: false, + positionalWrite: true, syncAccess: supportsSyncAccessHandle(), }, async stat(path, options) { @@ -239,6 +241,41 @@ export function createOpfsAdapter(root: FileSystemDirectoryHandle): OpfsAdapterT const parent = await getDirectory(nativeRoot, dirname(path)); await parent.removeEntry(basename(path)); }, + async openWritableFile(path): Promise { + const writable = await (await getFile(nativeRoot, path, true)).createWritable({ keepExistingData: true }); + let closed = false; + const getWritable = () => { + if (closed) throw new Error(`Writable file '${path}' is closed.`); + return writable; + }; + return { + async write(buffer, writeOptions) { + const view = new Uint8Array(buffer.buffer, buffer.byteOffset, buffer.byteLength); + const data = buffer.buffer instanceof ArrayBuffer + ? new Uint8Array(buffer.buffer, buffer.byteOffset, buffer.byteLength) + : Uint8Array.from(view); + await getWritable().write({ type: "write", position: writeOptions.at, data }); + }, + async truncate(size) { + await getWritable().truncate(size); + }, + async flush() { + // FileSystemWritableFileStream has no separate durability primitive. + // Native staging is committed by close(). + getWritable(); + }, + async close() { + if (closed) return; + closed = true; + await writable.close(); + }, + async abort(reason) { + if (closed) return; + closed = true; + await writable.abort(reason); + }, + }; + }, async openSyncFile(path) { const handle = await getFile(nativeRoot, path); if (handle.createSyncAccessHandle === undefined) { diff --git a/src/adapter/record.ts b/src/adapter/record.ts index 2e0c5ec..90a49e7 100644 --- a/src/adapter/record.ts +++ b/src/adapter/record.ts @@ -113,6 +113,7 @@ export function createRecordAdapter(store: RecordStoreType, options: RecordAdapt streamWrite: false, rangeRead: false, nativeMove: false, + positionalWrite: false, syncAccess: false, }, async stat(path, operationOptions) { diff --git a/src/filesystem.ts b/src/filesystem.ts index 7d4ad49..a7dcae8 100644 --- a/src/filesystem.ts +++ b/src/filesystem.ts @@ -19,6 +19,7 @@ import { withAbortSignal, } from "./stream.ts"; import { ManagedSyncFile, type SyncFileType } from "./sync.ts"; +import { ManagedWritableFile, type WritableFileType } from "./writable.ts"; import { DirectoryHandle, FileHandle, type DirectoryHandleType, type FileHandleType } from "./handle.ts"; /** Default lock namespace used when the caller does not provide one. */ @@ -125,6 +126,14 @@ export interface EmptyDirectoryOptionsType extends SignalOptionsType { readonly concurrency?: number; } +/** Options for opening long-lived asynchronous positional writes. */ +export interface OpenWritableFileOptionsType extends SignalOptionsType { + /** Creates the file when it does not exist. */ + readonly create?: boolean; + /** Creates missing parent directories when creating the file. */ + readonly parents?: boolean; +} + /** Options for opening synchronous random access. */ export interface OpenSyncFileOptionsType extends SignalOptionsType { /** Creates the file when it does not exist. */ @@ -240,6 +249,8 @@ export interface FileSystemType extends AsyncDisposable { remove(path: string, options?: RemoveOptionsType): Promise; /** Removes every direct or nested child while preserving the requested directory. */ emptyDir(path?: string, options?: EmptyDirectoryOptionsType): Promise; + /** Opens long-lived asynchronous positional writes when the selected adapter supports them. */ + openWritableFile(path: string, options?: OpenWritableFileOptionsType): Promise; /** Opens synchronous random access when the selected adapter supports it. */ openSyncFile(path: string, options?: OpenSyncFileOptionsType): Promise; /** Releases an adapter only when ownership was transferred at creation. */ @@ -983,6 +994,76 @@ class FileSystemFacade implements FileSystemType { } } + /** + * Opens long-lived asynchronous positional writes and transfers the file lock to the returned resource. + * + * This operation is capability-gated rather than emulated with repeated + * `writeFile(..., { mode: "update" })` calls. Record-oriented backends would + * otherwise rematerialize an increasingly large file for each chunk, which + * can turn a linear media write into quadratic work. + */ + async openWritableFile( + path: string, + options: OpenWritableFileOptionsType = {}, + ): Promise { + this.#assertOpen(); + const normalized = normalizePath(path); + if (normalized === ROOT_PATH) { + throw new FileSystemError("type-mismatch", "open-writable-file", normalized, "The virtual root is a directory."); + } + throwIfAborted(options.signal, "open-writable-file", normalized); + if (!this.adapter.capabilities.positionalWrite || this.adapter.openWritableFile === undefined) { + throw new FileSystemError( + "not-supported", + "open-writable-file", + normalized, + `Adapter '${this.adapter.name}' does not provide long-lived positional writes.`, + ); + } + + const lock = await this.#locks.acquireFile(normalized, options.signal); + try { + let stat = await this.adapter.stat(normalized, getAdapterSignalOptions(options.signal)); + if (stat?.kind === "directory") { + throw new FileSystemError("type-mismatch", "open-writable-file", normalized, `'${normalized}' is a directory.`); + } + if (stat === null) { + if (!options.create) { + throw new FileSystemError("not-found", "open-writable-file", normalized, `File '${normalized}' does not exist.`); + } + if (options.parents) await ensureParents(this.adapter, dirname(normalized), options.signal); + const parent = await this.adapter.stat(dirname(normalized), getAdapterSignalOptions(options.signal)); + if (parent?.kind !== "directory") { + throw new FileSystemError( + "not-found", + "open-writable-file", + normalized, + `Parent directory '${dirname(normalized)}' does not exist.`, + ); + } + await this.adapter.writeFile(normalized, new Uint8Array(), { + mode: "replace", + ...getAdapterSignalOptions(options.signal), + }); + stat = await this.adapter.stat(normalized, getAdapterSignalOptions(options.signal)); + if (stat?.kind !== "file") { + throw new FileSystemError( + "unknown", + "open-writable-file", + normalized, + `Adapter '${this.adapter.name}' did not expose the file after creating it.`, + ); + } + } + + const file = await this.adapter.openWritableFile(normalized); + return new ManagedWritableFile(normalized, file, lock, options.signal); + } catch (error) { + lock.release(); + throw toFileSystemError(error, "open-writable-file", normalized); + } + } + /** * Opens synchronous random access and transfers the file mutation lock to the returned resource. * @@ -992,6 +1073,10 @@ class FileSystemFacade implements FileSystemType { async openSyncFile(path: string, options: OpenSyncFileOptionsType = {}): Promise { this.#assertOpen(); const normalized = normalizePath(path); + if (normalized === ROOT_PATH) { + throw new FileSystemError("type-mismatch", "open-sync-file", normalized, "The virtual root is a directory."); + } + throwIfAborted(options.signal, "open-sync-file", normalized); if (!this.adapter.capabilities.syncAccess || this.adapter.openSyncFile === undefined) { throw new FileSystemError( "not-supported", @@ -1000,16 +1085,42 @@ class FileSystemFacade implements FileSystemType { `Adapter '${this.adapter.name}' does not provide synchronous file access.`, ); } - if (options.create) { - await this.getFileHandle(normalized, { - create: true, - parents: options.parents ?? false, - ...getAdapterSignalOptions(options.signal), - }); - } - else await this.getFileHandle(normalized, getAdapterSignalOptions(options.signal)); + const lock = await this.#locks.acquireFile(normalized, options.signal); try { + let stat = await this.adapter.stat(normalized, getAdapterSignalOptions(options.signal)); + if (stat?.kind === "directory") { + throw new FileSystemError("type-mismatch", "open-sync-file", normalized, `'${normalized}' is a directory.`); + } + if (stat === null) { + if (!options.create) { + throw new FileSystemError("not-found", "open-sync-file", normalized, `File '${normalized}' does not exist.`); + } + if (options.parents) await ensureParents(this.adapter, dirname(normalized), options.signal); + const parent = await this.adapter.stat(dirname(normalized), getAdapterSignalOptions(options.signal)); + if (parent?.kind !== "directory") { + throw new FileSystemError( + "not-found", + "open-sync-file", + normalized, + `Parent directory '${dirname(normalized)}' does not exist.`, + ); + } + await this.adapter.writeFile(normalized, new Uint8Array(), { + mode: "replace", + ...getAdapterSignalOptions(options.signal), + }); + stat = await this.adapter.stat(normalized, getAdapterSignalOptions(options.signal)); + if (stat?.kind !== "file") { + throw new FileSystemError( + "unknown", + "open-sync-file", + normalized, + `Adapter '${this.adapter.name}' did not expose the file after creating it.`, + ); + } + } + const file = await this.adapter.openSyncFile(normalized); return new ManagedSyncFile(normalized, file, lock); } catch (error) { diff --git a/src/schema.ts b/src/schema.ts index d612a1d..d9d6172 100644 --- a/src/schema.ts +++ b/src/schema.ts @@ -96,6 +96,7 @@ export const AdapterCapabilitiesSchema = z.object({ streamWrite: z.boolean(), rangeRead: z.boolean(), nativeMove: z.boolean(), + positionalWrite: z.boolean(), syncAccess: z.boolean(), }); diff --git a/src/writable.ts b/src/writable.ts new file mode 100644 index 0000000..3596d53 --- /dev/null +++ b/src/writable.ts @@ -0,0 +1,142 @@ +import type { AdapterWritableFileType } from "./adapter/definition.ts"; +import { FileSystemError, throwIfAborted, toFileSystemError } from "./error.ts"; +import type { HeldLockType } from "./lock.ts"; + +/** + * Long-lived asynchronous positional file returned by `openWritableFile()`. + * + * The resource owns the facade mutation lock for its complete lifetime. Calls + * can therefore rewrite earlier byte ranges without reopening the backend file + * and without racing another mutation through the same filesystem facade. + * + * `close()` commits backend staging where the adapter supports it. `abort()` + * discards staged changes when possible. A host filesystem cannot generally + * roll back bytes already written, so callers that need all-or-nothing output + * should write to a staging path and move it only after a successful close. + */ +export interface WritableFileType { + /** Canonical virtual path whose mutation lock is owned by this resource. */ + readonly path: string; + /** True after `close()` or `abort()` releases the backend file. */ + readonly closed: boolean; + /** Writes all bytes at one explicit zero-based position. */ + write(buffer: ArrayBufferView, options: { readonly at: number }): Promise; + /** Changes current byte length. */ + truncate(size: number): Promise; + /** Requests backend durability without releasing the resource. */ + flush(): Promise; + /** Commits backend staging when applicable and releases the mutation lock. */ + close(): Promise; + /** Discards backend staging when possible and releases the mutation lock. */ + abort(reason?: unknown): Promise; +} + +/** Owns one adapter writable file for the same lifetime as one facade lock. */ +export class ManagedWritableFile implements WritableFileType { + readonly path: string; + #file: AdapterWritableFileType | undefined; + readonly #lock: HeldLockType; + readonly #signal: AbortSignal | undefined; + + constructor( + path: string, + file: AdapterWritableFileType, + lock: HeldLockType, + signal?: AbortSignal, + ) { + this.path = path; + this.#file = file; + this.#lock = lock; + this.#signal = signal; + } + + get closed(): boolean { + return this.#file === undefined; + } + + /** Returns the live backend file and rejects ordinary work after termination. */ + #getFile(operation: string): AdapterWritableFileType { + if (this.#file === undefined) { + throw new FileSystemError( + "invalid-operation", + operation, + this.path, + `Writable file '${this.path}' is already closed.`, + ); + } + throwIfAborted(this.#signal, operation, this.path); + return this.#file; + } + + async write(buffer: ArrayBufferView, options: { readonly at: number }): Promise { + if (!Number.isSafeInteger(options.at) || options.at < 0) { + throw new RangeError("write position must be a non-negative safe integer."); + } + try { + await this.#getFile("positional-write").write(buffer, options); + throwIfAborted(this.#signal, "positional-write", this.path); + } catch (error) { + throw toFileSystemError(error, "positional-write", this.path); + } + } + + async truncate(size: number): Promise { + if (!Number.isSafeInteger(size) || size < 0) { + throw new RangeError("truncate size must be a non-negative safe integer."); + } + try { + await this.#getFile("positional-truncate").truncate(size); + throwIfAborted(this.#signal, "positional-truncate", this.path); + } catch (error) { + throw toFileSystemError(error, "positional-truncate", this.path); + } + } + + async flush(): Promise { + try { + await this.#getFile("positional-flush").flush(); + throwIfAborted(this.#signal, "positional-flush", this.path); + } catch (error) { + throw toFileSystemError(error, "positional-flush", this.path); + } + } + + /** + * Closes once and always releases the facade lock. + * + * The backend file is detached before close starts so a failed close cannot + * leave an apparently reusable resource that no longer has lock ownership. + */ + async close(): Promise { + const file = this.#file; + if (file === undefined) return; + this.#file = undefined; + try { + await file.close(); + } catch (error) { + throw toFileSystemError(error, "positional-close", this.path); + } finally { + this.#lock.release(); + } + } + + /** + * Aborts once and always releases the facade lock. + * + * Cleanup deliberately ignores the operation signal. Cancellation is the + * reason this method is often needed, so an already-aborted signal must not + * prevent native resources from being released. + */ + async abort(reason?: unknown): Promise { + const file = this.#file; + if (file === undefined) return; + this.#file = undefined; + try { + await file.abort(reason); + } catch (error) { + throw toFileSystemError(error, "positional-abort", this.path); + } finally { + this.#lock.release(); + } + } +} -- 2.51.2