fix: unreachable UNC paths in the vfs can cause "not found" error on the frontend #1115

Author: rejetto

src/stat.ts

@@ -1,42 +1,39 @@
-import { defineConfig } from './config'
+import { Worker } from 'node:worker_threads'
 import { Stats } from 'node:fs'
-import { haveTimeout, pendingPromise } from './cross'
-import { stat } from 'fs/promises'
+import { pendingPromise } from './cross'
 
-const fileTimeout = defineConfig('file_timeout', 3, x => x * 1000)
+// all stat requests for the same worker are serialized, potentially introducing extra latency
 
-// since nodejs' UV_THREADPOOL_SIZE is limited, avoid using multiple slots for the same UNC host, and always leave one free for local operations
-const poolSize = Number(process.env.UV_THREADPOOL_SIZE || 4)
-const previous = new Map<string, Promise<Stats>>() // wrapped promises with haveTimeout
-const working = new Set<Promise<Stats>>() // plain stat's promise
-export async function statWithTimeout(path: string) {
-    const uncHost = /^\\\\([^\\]+)\\/.exec(path)?.[1]
-    if (!uncHost)
-        return haveTimeout(fileTimeout.compiled(), stat(path))
-    const busy = process.env.HFS_PARALLEL_UNC ? null : previous.get(uncHost) // by default we serialize requests on the same UNC host, to keep threadpool usage low
-    const ret = pendingPromise<Stats>()
-    previous.set(uncHost, ret) // reserve the slot before starting the operation
-    const err = await busy?.then(() => false, e => e.message === 'timeout' && e) // only timeout error is shared with pending requests
-    if (err) {
-        if (previous.get(uncHost) === ret) // but we don't want to block forever, only involve those that were already waiting
-            previous.delete(uncHost)
-        ret.reject(err)
+const pool = new Map<string, (path: string) => Promise<Stats>>()
+
+export function getStatWorker(key: string) {
+    const existing = pool.get(key)
+    if (existing)
+        return existing
+    const worker = new Worker(__dirname + '/statWorker.js')
+    worker.unref()
+    const requests = new Map()
+    worker.on('message', (msg: any) => {
+        requests.get(msg.path)?.resolve(msg.error
+            ? Promise.reject(new Error(msg.error))
+            : Object.setPrototypeOf(msg.result, Stats.prototype)
+        )
+        requests.delete(msg.path)
+    })
+    worker.on('error', (err) => {
+        for (const p of requests.values())
+            p.reject(err)
+        requests.clear()
+        worker.terminate().catch(() => {})
+        pool.delete(key)
+    })
+    pool.set(key, query)
+    return query
+
+    function query(path: string) {
+        const ret = pendingPromise<Stats>()
+        requests.set(path, ret)
+        worker.postMessage(path)
         return ret
     }
-    while (working.size >= poolSize - 1) // always leave one slot free for local operations
-        await Promise.race(working.values()).catch(() => {}) // we are assuming UV_THREADPOOL_SIZE > 1, otherwise race() will deadlock
-    const op = stat(path)
-    working.add(op)
-    try {
-        ret.resolve(await haveTimeout(fileTimeout.compiled(),
-            op.finally(() => working.delete(op)) ))
-    }
-    catch (e) {
-        ret.reject(e)
-    }
-    finally {
-        if (previous.get(uncHost) === ret)
-            previous.delete(uncHost)
-    }
-    return ret
 }

src/statWorker.js

@@ -0,0 +1,10 @@
+const { parentPort } = require('node:worker_threads')
+const { statSync } = require('node:fs')
+
+parentPort.on('message', async path => {
+    try { // we use statSync to not use (and not risk to saturate) libuv's thread pool
+        parentPort.postMessage({ path, result: { ...statSync(path) } })
+    } catch (err) {
+        parentPort.postMessage({ path, error: err?.message || String(err) })
+    }
+})

src/util-files.ts

@@ -1,18 +1,33 @@
 // This file is part of HFS - Copyright 2021-2023, Massimo Melina <[email protected]> - License https://www.gnu.org/licenses/gpl-3.0.txt
 
-import { access, mkdir, readFile } from 'fs/promises'
-import { Promisable, try_, wait, isWindowsDrive } from './cross'
+import { access, mkdir, readFile, stat } from 'fs/promises'
+import { Promisable, try_, wait, isWindowsDrive, haveTimeout } from './cross'
+import { defineConfig } from './config'
 import { createWriteStream, mkdirSync, watch, ftruncate } from 'fs'
 import { basename, dirname } from 'path'
 import glob from 'fast-glob'
 import { IS_WINDOWS } from './const'
 import { once } from 'events'
 import { Readable } from 'stream'
-import { statWithTimeout } from './stat'
+import { getStatWorker } from './stat'
 // @ts-ignore
 import unzipper from 'unzip-stream'
 
-export { statWithTimeout }
+const fileTimeout = defineConfig('file_timeout', 3, x => x * 1000)
+// a smart (and a bit arbitrary) way to decide if we need the stat-workers functionality. Without it, we may be a bit faster. We'll see with experience if we need a dedicated configuration.
+const disableStatWorkers = Number(process.env.UV_THREADPOOL_SIZE) >= 10
+
+// some paths (mostly offline networked ones) can take 20+ seconds to fail
+export async function statWithTimeout(path: string) {
+    // with just 4 requests we saturate the default UV_THREADPOOL_SIZE, blocking even local fs operations, so we move all UNC requests to worker threads
+    const workerKey = !disableStatWorkers && IS_WINDOWS && getUncHost(path) // pool requests by server name
+    const op = workerKey ? getStatWorker(workerKey)(path) : stat(path)
+    return haveTimeout(fileTimeout.compiled(), op)
+}
+
+function getUncHost(path: string) {
+    return /^\\\\([^\\]+)\\/.exec(path)?.[1]
+}
 
 export async function isDirectory(path: string) {
     try { return (await statWithTimeout(path)).isDirectory() }