From 92cac5d291a160fff39edd67df8dad99afa4d1c2 Mon Sep 17 00:00:00 2001 From: imccyu <276526105+imccyu@users.noreply.github.com> Date: Mon, 24 Aug 2026 03:58:48 +0800 Subject: [PATCH] fix(webworker): close Node compatibility gaps --- .../builtin_modules/implemented/fs-watch.ts | 30 ++++-- .../node/builtin_modules/implemented/fs.ts | 47 +++++---- .../builtin_modules/implemented/stream.ts | 2 +- .../src/shell/process/landlock.ts | 9 +- .../webworker-runtime/src/storage/memory.ts | 8 ++ .../tests/node/child-process.spec.ts | 4 + .../tests/node/fs-watch-stream.spec.ts | 96 ++++++++++++++++++- .../tests/storage/memory-vfs.spec.ts | 37 +++++++ 8 files changed, 201 insertions(+), 32 deletions(-) diff --git a/packages/experimental/webworker-runtime/src/node/builtin_modules/implemented/fs-watch.ts b/packages/experimental/webworker-runtime/src/node/builtin_modules/implemented/fs-watch.ts index 61e3f4419c..a570e54bc2 100644 --- a/packages/experimental/webworker-runtime/src/node/builtin_modules/implemented/fs-watch.ts +++ b/packages/experimental/webworker-runtime/src/node/builtin_modules/implemented/fs-watch.ts @@ -354,12 +354,20 @@ export function watchAsync( let failure: Error | undefined let closed = false - const settleFailure = (reason: unknown): void => { - if (failure !== undefined || closed) return - const error = reason instanceof Error ? reason : new Error(String(reason)) - failure = error + const stopWatcher = (): void => { + options.signal?.removeEventListener('abort', onAbort) watcher?.close() - for (const pending of waiting.splice(0)) pending.reject(error) + } + const settleFailure = (reason: unknown): void => { + if (closed) return + const error = reason instanceof Error ? reason : new Error(String(reason)) + closed = true + queued.length = 0 + stopWatcher() + const failed = waiting.shift() + if (failed === undefined) failure = error + else failed.reject(error) + for (const pending of waiting.splice(0)) pending.resolve({ done: true, value: undefined }) } const onAbort = (): void => { settleFailure(abortError(options.signal?.reason)) } const start = (): void => { @@ -382,11 +390,11 @@ export function watchAsync( } } const close = (): void => { - if (closed) return + const alreadyClosed = closed closed = true queued.length = 0 - options.signal?.removeEventListener('abort', onAbort) - watcher?.close() + failure = undefined + if (!alreadyClosed) stopWatcher() for (const pending of waiting.splice(0)) pending.resolve({ done: true, value: undefined }) } @@ -396,7 +404,11 @@ export function watchAsync( }, next(): Promise> { start() - if (failure !== undefined) return Promise.reject(failure) + if (failure !== undefined) { + const reason = failure + failure = undefined + return Promise.reject(reason) + } const event = queued.shift() if (event !== undefined) return Promise.resolve({ done: false, value: event }) if (closed) return Promise.resolve({ done: true, value: undefined }) diff --git a/packages/experimental/webworker-runtime/src/node/builtin_modules/implemented/fs.ts b/packages/experimental/webworker-runtime/src/node/builtin_modules/implemented/fs.ts index e4d7099f3e..b2f4462a50 100644 --- a/packages/experimental/webworker-runtime/src/node/builtin_modules/implemented/fs.ts +++ b/packages/experimental/webworker-runtime/src/node/builtin_modules/implemented/fs.ts @@ -319,14 +319,16 @@ export function openSync(path: PathArg, flags = 'r', mode?: number): number { return fd } +const badFileDescriptor = (syscall: string): never => { + const error = new Error(`EBADF: bad file descriptor, ${syscall}`) as Error & { code: string; syscall: string } + error.code = 'EBADF' + error.syscall = syscall + throw error +} + const fileOf = (fd: number, syscall: string): OpenFile => { const file = openFiles.get(fd) - if (file === undefined) { - const error = new Error(`EBADF: bad file descriptor, ${syscall}`) as Error & { code: string; syscall: string } - error.code = 'EBADF' - error.syscall = syscall - throw error - } + if (file === undefined) return badFileDescriptor(syscall) return file } @@ -558,15 +560,18 @@ export class ReadStream extends Readable { callback(abortError(this.signal.reason)) return } + let fd: number try { - this.fd = openSync(this.path, this.flags) - this.pending = false - this.emit('open', this.fd) - this.emit('ready') - callback() + fd = openSync(this.path, this.flags) } catch (error) { callback(error as Error) + return } + this.fd = fd + this.pending = false + callback() + this.emit('open', fd) + this.emit('ready') } override _read(size: number): void { @@ -648,16 +653,19 @@ export class WriteStream extends Writable { callback(abortError(this.signal.reason)) return } + let fd: number try { - this.fd = openSync(this.path, this.flags, this.mode) - if (this.start !== undefined) fileOf(this.fd, 'write').position = this.start - this.pending = false - this.emit('open', this.fd) - this.emit('ready') - callback() + fd = openSync(this.path, this.flags, this.mode) } catch (error) { callback(error as Error) + return } + this.fd = fd + if (this.start !== undefined) fileOf(fd, 'write').position = this.start + this.pending = false + callback() + this.emit('open', fd) + this.emit('ready') } override _write( @@ -666,9 +674,10 @@ export class WriteStream extends Writable { callback: (error?: Error | null) => void, ): void { try { - if (this.fd === null) throw new Error('EBADF: bad file descriptor, write') + const fd = this.fd + if (fd === null) return badFileDescriptor('write') const data = typeof chunk === 'string' ? Buffer.from(chunk, encoding) : chunk - this.bytesWritten += writeSync(this.fd, data) + this.bytesWritten += writeSync(fd, data) callback() } catch (error) { callback(error as Error) diff --git a/packages/experimental/webworker-runtime/src/node/builtin_modules/implemented/stream.ts b/packages/experimental/webworker-runtime/src/node/builtin_modules/implemented/stream.ts index a7fca23540..19c60afa15 100644 --- a/packages/experimental/webworker-runtime/src/node/builtin_modules/implemented/stream.ts +++ b/packages/experimental/webworker-runtime/src/node/builtin_modules/implemented/stream.ts @@ -46,7 +46,7 @@ if (getDefaultHighWaterMark(false) !== 64 * 1024) setDefaultHighWaterMark(false, const _isArrayBufferView = (value: unknown): value is ArrayBufferView => ArrayBuffer.isView(value) /** Default-import namespace carrying Node's stream class and static helpers. */ -const streamDefault = Object.assign(Stream, { +const streamDefault = Object.assign(StreamBase, { _isArrayBufferView, getDefaultHighWaterMark, isDestroyed, diff --git a/packages/experimental/webworker-runtime/src/shell/process/landlock.ts b/packages/experimental/webworker-runtime/src/shell/process/landlock.ts index 03ff455c64..5d8c8c3ca0 100644 --- a/packages/experimental/webworker-runtime/src/shell/process/landlock.ts +++ b/packages/experimental/webworker-runtime/src/shell/process/landlock.ts @@ -108,13 +108,18 @@ export async function landlockFileSystem( const readWrite = await Promise.all(invocation.readWrite.map(normalizeGrant)) const readable = [...readOnly, ...readWrite] - const readPath = (path: string, syscall: string): string => { + const checkedPath = (path: string, syscall: string): string => { const target = vfsPath(path, cwd) + if (target.startsWith(`${NULL_PATH}/`)) throw filesystemError('ENOTDIR', syscall, path) + return target + } + const readPath = (path: string, syscall: string): string => { + const target = checkedPath(path, syscall) if (!readable.some(root => contains(root, target))) deny(syscall, path) return target } const writePath = (path: string, syscall: string): string => { - const target = vfsPath(path, cwd) + const target = checkedPath(path, syscall) if (!readWrite.some(root => contains(root, target))) deny(syscall, path) return target } diff --git a/packages/experimental/webworker-runtime/src/storage/memory.ts b/packages/experimental/webworker-runtime/src/storage/memory.ts index 7502e90bb8..5a6826f2e3 100644 --- a/packages/experimental/webworker-runtime/src/storage/memory.ts +++ b/packages/experimental/webworker-runtime/src/storage/memory.ts @@ -699,6 +699,14 @@ export class MemoryVfs implements Vfs { return } if (!this.directories.has(source)) fail('ENOENT', 'rename', source) + if (this.files.has(destination)) fail('ENOTDIR', 'rename', destination) + if (!this.directories.has(dirname(destination))) fail('ENOENT', 'rename', destination) + if (this.directories.has(destination)) { + if (this.readdirSync(destination).length > 0) fail('ENOTEMPTY', 'rename', destination) + this.directories.delete(destination) + this.directoryModes.delete(destination) + this.directoryMtimes.delete(destination) + } const prefix = `${source}${SEP}` const movedFiles: Array<{ path: string; bytes: Uint8Array; mode: number }> = [] for (const [candidate, value] of [...this.files]) { diff --git a/packages/experimental/webworker-runtime/tests/node/child-process.spec.ts b/packages/experimental/webworker-runtime/tests/node/child-process.spec.ts index 3c5304e8df..a1e6fd3e9c 100644 --- a/packages/experimental/webworker-runtime/tests/node/child-process.spec.ts +++ b/packages/experimental/webworker-runtime/tests/node/child-process.spec.ts @@ -160,6 +160,10 @@ it('enforces every ShellFileSystem operation and virtual device edge', async () await expect(guarded.mkdir('/dev/null', false)).rejects.toMatchObject({ code: 'EEXIST' }) await expect(guarded.remove('/dev/null', { recursive: false, force: false })).rejects.toMatchObject({ code: 'EACCES' }) await expect(guarded.rename('/dev/null', `${WORKSPACE}/null`)).rejects.toMatchObject({ code: 'EACCES' }) + await expect(guarded.stat('/dev/null/child')).rejects.toMatchObject({ code: 'ENOTDIR' }) + await expect(guarded.writeText('/dev/null/child', 'not written')).rejects.toMatchObject({ code: 'ENOTDIR' }) + await expect(guarded.mkdir('/dev/null/child', true)).rejects.toMatchObject({ code: 'ENOTDIR' }) + expect(vfs.existsSync('/dev')).toBe(false) await expect(guarded.readText(`${HOME}/private.txt`)).rejects.toMatchObject({ code: 'EACCES' }) await guarded.mkdir('created', false) diff --git a/packages/experimental/webworker-runtime/tests/node/fs-watch-stream.spec.ts b/packages/experimental/webworker-runtime/tests/node/fs-watch-stream.spec.ts index df8c31e0b5..de84fcce85 100644 --- a/packages/experimental/webworker-runtime/tests/node/fs-watch-stream.spec.ts +++ b/packages/experimental/webworker-runtime/tests/node/fs-watch-stream.spec.ts @@ -273,6 +273,9 @@ describe('file streams', () => { const values: string[] = [] for await (const value of workerStream.Readable.from(['one', 'two'])) values.push(String(value)) expect(values).toEqual(['one', 'two']) + expect(workerStream.default).toBe(workerStream.Stream) + expect(new workerStream.Writable({ write: (_chunk, _encoding, callback) => { callback() } })) + .toBeInstanceOf(workerStream.default) expect(typeof workerStream.pipeline).toBe('function') expect(typeof workerStream.finished).toBe('function') expect(workerStream.getDefaultHighWaterMark(false)).toBe(64 * 1024) @@ -385,6 +388,91 @@ describe('file streams', () => { }) expect(events).toEqual(['error', 'close']) }) + + it('publishes descriptors before open and ready listener exceptions escape', () => { + const readPath = `${VFS_ROOT}/listener-read.txt` + vfs.writeFileSync(readPath, 'content') + const readCallback = vi.fn() + const readFailure = new Error('read open listener failed') + const readReceiver: { + path: string + flags: string + start: number + end: number + signal: undefined + pending: boolean + fd: number | null + emit(event: string): boolean + } = { + path: readPath, + flags: 'r', + start: 0, + end: Number.POSITIVE_INFINITY, + signal: undefined, + pending: true, + fd: null, + emit(event) { + expect(readCallback).toHaveBeenCalledOnce() + if (event === 'open') throw readFailure + return true + }, + } + expect(() => { + workerFs.ReadStream.prototype._construct.call( + readReceiver as unknown as workerFs.ReadStream, + readCallback, + ) + }).toThrow(readFailure) + expect(readReceiver.pending).toBe(false) + expect(readReceiver.fd).not.toBeNull() + workerFs.closeSync(readReceiver.fd as number) + + const writeCallback = vi.fn() + const writeFailure = new Error('write ready listener failed') + const writeReceiver: { + path: string + flags: string + mode: undefined + start: undefined + signal: undefined + pending: boolean + fd: number | null + emit(event: string): boolean + } = { + path: `${VFS_ROOT}/listener-write.txt`, + flags: 'w', + mode: undefined, + start: undefined, + signal: undefined, + pending: true, + fd: null, + emit(event) { + expect(writeCallback).toHaveBeenCalledOnce() + if (event === 'ready') throw writeFailure + return true + }, + } + expect(() => { + workerFs.WriteStream.prototype._construct.call( + writeReceiver as unknown as workerFs.WriteStream, + writeCallback, + ) + }).toThrow(writeFailure) + expect(writeReceiver.pending).toBe(false) + expect(writeReceiver.fd).not.toBeNull() + workerFs.closeSync(writeReceiver.fd as number) + }) + + it('codes a write before descriptor publication as EBADF', () => { + let failure: Error | null | undefined + workerFs.WriteStream.prototype._write.call( + { fd: null } as unknown as workerFs.WriteStream, + Buffer.from('x'), + 'utf8', + (error) => { failure = error }, + ) + expect(failure).toMatchObject({ code: 'EBADF', syscall: 'write' }) + }) }) interface StatTransition { @@ -673,8 +761,12 @@ describe('watchers', () => { const event = iterator.next() vfs.writeFileSync(`${VFS_ROOT}/async.txt`, 'x') await expect(event).resolves.toEqual({ done: false, value: { eventType: 'rename', filename: 'async.txt' } }) + const failed = iterator.next() + const completed = iterator.next() controller.abort() - await expect(iterator.next()).rejects.toMatchObject({ name: 'AbortError', code: 'ABORT_ERR' }) + await expect(failed).rejects.toMatchObject({ name: 'AbortError', code: 'ABORT_ERR' }) + await expect(completed).resolves.toEqual({ done: true, value: undefined }) + await expect(iterator.next()).resolves.toEqual({ done: true, value: undefined }) }) it('rejects the first promise-watch read for a pre-aborted signal', async () => { @@ -683,6 +775,7 @@ describe('watchers', () => { controller.abort(reason) const iterator = workerFsp.watch(VFS_ROOT, { signal: controller.signal })[Symbol.asyncIterator]() await expect(iterator.next()).rejects.toMatchObject({ name: 'AbortError', code: 'ABORT_ERR', cause: reason }) + await expect(iterator.next()).resolves.toEqual({ done: true, value: undefined }) }) it('lets promise-watch return interrupt a pending next call', async () => { @@ -697,6 +790,7 @@ describe('watchers', () => { it('propagates promise-watch startup and throw failures', async () => { const missing = workerFsp.watch(`${VFS_ROOT}/missing`)[Symbol.asyncIterator]() await expect(missing.next()).rejects.toMatchObject({ code: 'ENOENT' }) + await expect(missing.next()).resolves.toEqual({ done: true, value: undefined }) const iterator = workerFsp.watch(VFS_ROOT)[Symbol.asyncIterator]() const reason = { reason: 'caller stopped iteration' } diff --git a/packages/experimental/webworker-runtime/tests/storage/memory-vfs.spec.ts b/packages/experimental/webworker-runtime/tests/storage/memory-vfs.spec.ts index 069da0ca2a..ac994e84aa 100644 --- a/packages/experimental/webworker-runtime/tests/storage/memory-vfs.spec.ts +++ b/packages/experimental/webworker-runtime/tests/storage/memory-vfs.spec.ts @@ -220,6 +220,43 @@ describe('mutation publication', () => { }) }) +describe('directory rename', () => { + it('rejects file, non-empty directory, and missing-parent destinations before mutation', () => { + const vfs = new MemoryVfs() + vfs.seed('/dsh/source/nested/file', 'source') + vfs.seed('/dsh/file', 'destination') + vfs.seed('/dsh/non-empty/child', 'destination') + const mutations: VfsMutation[] = [] + vfs.subscribe((mutation) => { mutations.push(mutation) }) + + expect(() => { vfs.renameSync('/dsh/source', '/dsh/file') }) + .toThrow(expect.objectContaining({ code: 'ENOTDIR' })) + expect(() => { vfs.renameSync('/dsh/source', '/dsh/non-empty') }) + .toThrow(expect.objectContaining({ code: 'ENOTEMPTY' })) + expect(() => { vfs.renameSync('/dsh/source', '/missing/destination') }) + .toThrow(expect.objectContaining({ code: 'ENOENT' })) + + expect(vfs.readFileSync('/dsh/source/nested/file', 'utf8')).toBe('source') + expect(vfs.readFileSync('/dsh/file', 'utf8')).toBe('destination') + expect(vfs.readFileSync('/dsh/non-empty/child', 'utf8')).toBe('destination') + expect(mutations).toEqual([]) + }) + + it('replaces an empty directory with the source subtree', () => { + const vfs = new MemoryVfs() + vfs.seedDirectory('/dsh/source/nested', { mode: 0o700 }) + vfs.seed('/dsh/source/nested/file', 'source') + vfs.seedDirectory('/dsh/destination', { mode: 0o711 }) + + vfs.renameSync('/dsh/source', '/dsh/destination') + + expect(vfs.existsSync('/dsh/source')).toBe(false) + expect(vfs.readFileSync('/dsh/destination/nested/file', 'utf8')).toBe('source') + expect((vfs.statSync('/dsh/destination') as VfsStats).mode & 0o777).toBe(0o755) + expect((vfs.statSync('/dsh/destination/nested') as VfsStats).mode & 0o777).toBe(0o700) + }) +}) + describe('hard links', () => { it('shares identity, bytes, and mode until one name is removed', () => { const vfs = new MemoryVfs()