fix(webworker): close Node compatibility gaps

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 stopWatcher = (): void => {
+    options.signal?.removeEventListener('abort', onAbort)
+    watcher?.close()
+  }
   const settleFailure = (reason: unknown): void => {
-    if (failure !== undefined || closed) return
+    if (closed) return
     const error = reason instanceof Error ? reason : new Error(String(reason))
-    failure = error
-    watcher?.close()
-    for (const pending of waiting.splice(0)) pending.reject(error)
+    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<IteratorResult<WatchEvent>> {
       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 })

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)

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,

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
   }

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]) {

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)

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' }

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()