mirror of
https://github.com/stablyai/orca.git
synced 2026-10-16 00:03:09 +00:00
Address review: worker error handling + awaited shutdown drain (#5308)
- file-watcher-worker: add a .catch to the per-batch stat mapping chain so a throwing postMessage/stat surfaces as an error+overflow instead of crashing the worker with an unhandled rejection. Factored the error+overflow emit into reportWatchError(). - file-watcher-host: the disposer is now async and awaits worker.terminate(), so awaitRuntimeFileWatcherUnsubscribes() doesn't resolve until the worker is actually down (the shutdown-drain guarantee the tracking machinery provides). Co-authored-by: Orca <help@stably.ai>
This commit is contained in:
@@ -134,12 +134,12 @@ describe('watchFileExplorerInWorker', () => {
|
||||
worker.emit('message', { type: 'ready' })
|
||||
const dispose = await promise
|
||||
|
||||
dispose()
|
||||
await dispose()
|
||||
expect(worker.postedMessages).toContainEqual({ type: 'unsubscribe' })
|
||||
expect(worker.terminated).toBe(true)
|
||||
|
||||
// Idempotent: a second dispose does nothing further.
|
||||
dispose()
|
||||
await dispose()
|
||||
expect(
|
||||
worker.postedMessages.filter((m) => (m as { type?: string }).type === 'unsubscribe')
|
||||
).toHaveLength(1)
|
||||
@@ -151,7 +151,7 @@ describe('watchFileExplorerInWorker', () => {
|
||||
const worker = lastWorker()
|
||||
worker.emit('message', { type: 'ready' })
|
||||
const dispose = await promise
|
||||
dispose()
|
||||
await dispose()
|
||||
onEvents.mockClear()
|
||||
|
||||
worker.emit('message', {
|
||||
|
||||
@@ -36,7 +36,7 @@ function getFileWatcherWorkerPath(): string {
|
||||
export function watchFileExplorerInWorker(
|
||||
rootPath: string,
|
||||
callback: (events: FsChangeEvent[]) => void
|
||||
): Promise<() => void> {
|
||||
): Promise<() => Promise<void>> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const worker = new Worker(getFileWatcherWorkerPath(), {
|
||||
workerData: { rootPath, ignore: RUNTIME_FILE_WATCH_IGNORE }
|
||||
@@ -45,7 +45,10 @@ export function watchFileExplorerInWorker(
|
||||
let ready = false
|
||||
let disposed = false
|
||||
|
||||
const dispose = (): void => {
|
||||
// Why: returns a promise that resolves once the worker is actually down, so
|
||||
// the shutdown drain (awaitRuntimeFileWatcherUnsubscribes) doesn't finish
|
||||
// while the native watcher thread is still alive.
|
||||
const dispose = async (): Promise<void> => {
|
||||
if (disposed) {
|
||||
return
|
||||
}
|
||||
@@ -57,7 +60,10 @@ export function watchFileExplorerInWorker(
|
||||
} catch {
|
||||
// Worker already gone — terminate below covers it.
|
||||
}
|
||||
void worker.terminate()
|
||||
await worker.terminate().then(
|
||||
() => undefined,
|
||||
() => undefined
|
||||
)
|
||||
}
|
||||
|
||||
worker.on('message', (message: FileWatcherWorkerMessage) => {
|
||||
|
||||
@@ -36,6 +36,20 @@ if (!parentPort) {
|
||||
|
||||
const port = parentPort
|
||||
|
||||
/** Report a watcher failure to the host and ask the renderer to refresh from
|
||||
* scratch (the overflow event), so a mid-stream error never leaves the
|
||||
* explorer silently stale. */
|
||||
function reportWatchError(err: unknown): void {
|
||||
port.postMessage({
|
||||
type: 'error',
|
||||
message: err instanceof Error ? err.message : String(err)
|
||||
} satisfies FileWatcherWorkerMessage)
|
||||
port.postMessage({
|
||||
type: 'events',
|
||||
events: [{ kind: 'overflow', absolutePath: data.rootPath }]
|
||||
} satisfies FileWatcherWorkerMessage)
|
||||
}
|
||||
|
||||
/** Run an async mapper over items with a bounded number in flight at once, so a
|
||||
* large batch can't occupy every libuv threadpool thread in this worker. */
|
||||
async function mapWithConcurrency<T, R>(
|
||||
@@ -71,15 +85,7 @@ async function main(): Promise<void> {
|
||||
data.rootPath,
|
||||
(err, events) => {
|
||||
if (err) {
|
||||
// Surface the error and ask the renderer to refresh from scratch.
|
||||
port.postMessage({
|
||||
type: 'error',
|
||||
message: err.message
|
||||
} satisfies FileWatcherWorkerMessage)
|
||||
port.postMessage({
|
||||
type: 'events',
|
||||
events: [{ kind: 'overflow', absolutePath: data.rootPath }]
|
||||
} satisfies FileWatcherWorkerMessage)
|
||||
reportWatchError(err)
|
||||
return
|
||||
}
|
||||
// Why: large watcher batches usually mean a generated directory or branch
|
||||
@@ -103,9 +109,13 @@ async function main(): Promise<void> {
|
||||
}
|
||||
return { kind: event.type, absolutePath: event.path, isDirectory }
|
||||
}
|
||||
).then((mapped) => {
|
||||
port.postMessage({ type: 'events', events: mapped } satisfies FileWatcherWorkerMessage)
|
||||
})
|
||||
)
|
||||
.then((mapped) => {
|
||||
port.postMessage({ type: 'events', events: mapped } satisfies FileWatcherWorkerMessage)
|
||||
})
|
||||
// Why: without this, a throwing postMessage / stat becomes an unhandled
|
||||
// rejection that crashes the worker silently. Surface it instead.
|
||||
.catch((err: unknown) => reportWatchError(err))
|
||||
},
|
||||
{ ignore: data.ignore }
|
||||
)
|
||||
|
||||
@@ -298,7 +298,7 @@ export class RuntimeFileCommands {
|
||||
// recursive crawl can't starve the main/`serve` process (issue #5308).
|
||||
const dispose = await watchFileExplorerInWorker(rootPath, callback)
|
||||
return () => {
|
||||
trackRuntimeFileWatcherUnsubscribe(rootPath, async () => dispose())
|
||||
trackRuntimeFileWatcherUnsubscribe(rootPath, dispose)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user