Fix remaining PTY and bundle watch races

Keep inherited child prompt markers bounded by the normal silence fallback. Stage web-plugin rescans atomically and retain missing watch state until a successful rebuild.
This commit is contained in:
Tianyi Cui
2026-07-25 00:33:33 +08:00
parent f9a638b8a6
commit 1d066e0f74
9 changed files with 158 additions and 37 deletions

View File

@@ -8,7 +8,7 @@ Client-disconnect detection hangs off the **response** `close` event, not the re
A request whose handling throws (a malformed %-escape hitting `decodeURIComponent`, a client dropping mid-body) is answered 400 — or the socket destroyed when headers are already out — and reported to `onError`; it never becomes a process-killing unhandled rejection.
In development, the client-plugin registry synchronously captures each built bundle's stat baseline before it returns, then polls those baselines and re-hashes changed content. An immediate rebuild therefore cannot disappear into an asynchronously established watch baseline; a rename window retains the last successful baseline and retries when the bundle reappears.
In development, the client-plugin registry synchronously captures each built bundle's stat baseline before it returns, then polls those baselines and re-hashes changed content. Each rescan stages its candidate table, graph, and watch map before publishing them, so a baseline failure preserves the prior graph. An immediate rebuild therefore cannot disappear into an asynchronously established watch baseline; a rename window marks the path dirty, retains the last successful baseline, and forces a re-hash when the bundle reappears even with identical metadata.
## Model Experience

View File

@@ -129,6 +129,13 @@ interface WebPluginRecord {
clientPath: string
}
interface WatchedBundle {
path: string
mtimeMs: number
size: number
dirty: boolean
}
/** Narrow an unknown parsed JSON value to the dshClient declaration, throwing on malformed fields. */
function parseDshClient(name: string, value: unknown): DshClientDeclaration | undefined {
if (value === undefined) return undefined
@@ -203,8 +210,32 @@ export function createHostWebPluginRegistry(deps: WebPluginRegistryDeps): HostWe
throw new Error(`web-plugins: watch.intervalMs must be a positive integer (got ${String(deps.watch?.intervalMs)})`)
}
const stageWatches = (
candidateTable: Map<string, WebPluginRecord>,
currentWatches: Map<string, WatchedBundle>,
): Map<string, WatchedBundle> => {
const candidateWatches = new Map<string, WatchedBundle>()
if (watchInterval === undefined) return candidateWatches
for (const [id, record] of candidateTable) {
const current = currentWatches.get(id)
if (current?.path === record.clientPath) {
candidateWatches.set(id, { ...current })
continue
}
const baseline = statSync(record.clientPath)
candidateWatches.set(id, {
path: record.clientPath,
mtimeMs: baseline.mtimeMs,
size: baseline.size,
dirty: false,
})
}
return candidateWatches
}
let table = scan(deps)
let graph = composeGraph(table)
let watched = stageWatches(table, new Map())
const rebuildListeners = new Set<(id: string, rev: string) => void>()
const rebuilt = (id: string): string | undefined => {
@@ -220,21 +251,6 @@ export function createHostWebPluginRegistry(deps: WebPluginRegistryDeps): HostWe
// registry is returned, then poll those baselines. fs.watchFile establishes
// its first baseline asynchronously, so an immediate rebuild can otherwise
// become the baseline and disappear without an observed delta.
const watched = new Map<string, { path: string; mtimeMs: number; size: number }>()
const syncWatches = (): void => {
if (watchInterval === undefined) return
for (const [id, watch] of watched) {
if (table.get(id)?.clientPath === watch.path) continue
watched.delete(id)
}
for (const [id, record] of table) {
if (watched.has(id)) continue
const baseline = statSync(record.clientPath)
watched.set(id, { path: record.clientPath, mtimeMs: baseline.mtimeMs, size: baseline.size })
}
}
syncWatches()
const pollWatches = (): void => {
for (const [id, watch] of watched) {
let current: Stats
@@ -242,18 +258,24 @@ export function createHostWebPluginRegistry(deps: WebPluginRegistryDeps): HostWe
current = statSync(watch.path)
} catch (error) {
const code = (error as NodeJS.ErrnoException).code
if (code === 'ENOENT') continue // mid-rename window; retry against the retained baseline
if (code === 'ENOENT') {
watch.dirty = true
continue
}
deps.onError(error instanceof Error ? error : new Error(String(error)))
continue
}
if (current.mtimeMs === watch.mtimeMs && current.size === watch.size) continue
if (!watch.dirty && current.mtimeMs === watch.mtimeMs && current.size === watch.size) continue
const before = table.get(id)?.entry.rev
let rev: string | undefined
try {
rev = rebuilt(id)
} catch (error) {
const code = (error as NodeJS.ErrnoException).code
if (code === 'ENOENT') continue // mid-rename window; retry against the retained baseline
if (code === 'ENOENT') {
watch.dirty = true
continue
}
watch.mtimeMs = current.mtimeMs
watch.size = current.size
deps.onError(error instanceof Error ? error : new Error(String(error)))
@@ -261,6 +283,7 @@ export function createHostWebPluginRegistry(deps: WebPluginRegistryDeps): HostWe
}
watch.mtimeMs = current.mtimeMs
watch.size = current.size
watch.dirty = false
if (rev === undefined || rev === before) continue
for (const notify of rebuildListeners) {
// A throwing subscriber must not skip later subscribers or escape the
@@ -283,9 +306,12 @@ export function createHostWebPluginRegistry(deps: WebPluginRegistryDeps): HostWe
queueMicrotask(() => {
pending = false
try {
table = scan(deps)
graph = composeGraph(table)
syncWatches()
const candidateTable = scan(deps)
const candidateGraph = composeGraph(candidateTable)
const candidateWatches = stageWatches(candidateTable, watched)
table = candidateTable
graph = candidateGraph
watched = candidateWatches
} catch (error) {
// Keep serving the previous graph: a mid-flight rescan failure must not
// take down the boot manifest for plugins that were fine.

View File

@@ -1,11 +1,41 @@
import { mkdirSync, mkdtempSync, writeFileSync } from 'node:fs'
import {
mkdirSync,
mkdtempSync,
statSync,
type PathLike,
type Stats,
unlinkSync,
utimesSync,
writeFileSync,
} from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { Context } from 'cordis'
import { describe, expect, it, vi } from 'vitest'
import { afterEach, describe, expect, it, vi } from 'vitest'
import { createHostWebPluginRegistry, injectBootManifest } from '../src/index.ts'
import type { LoaderEntryView, WebPluginRegistryDeps } from '../src/index.ts'
const fsControl = vi.hoisted(() => ({ failNextStatPath: undefined as string | undefined }))
vi.mock('node:fs', async (importOriginal) => {
const actual = await importOriginal<typeof import('node:fs')>()
return {
...actual,
statSync: (path: PathLike): Stats => {
if (String(path) === fsControl.failNextStatPath) {
fsControl.failNextStatPath = undefined
throw Object.assign(new Error('staged bundle missing'), { code: 'ENOENT' })
}
return actual.statSync(path)
},
}
})
afterEach(() => {
fsControl.failNextStatPath = undefined
vi.useRealTimers()
})
/** Write a fake installed package (package.json + optional client bundle) and return its package.json path. */
function makePkg(root: string, name: string, pkg: Record<string, unknown>, withBundle = true): string {
const dir = join(root, name.replaceAll('/', '__'))
@@ -147,6 +177,58 @@ describe('createHostWebPluginRegistry', () => {
expect(rebuilds).toHaveLength(1)
})
it('watch mode: a failed rescan baseline preserves the published table and graph', async () => {
const { deps, entries, errors, ctx, root } = makeDeps([
{ name: 'stable', pkg: webDecl() },
{ name: 'late', pkg: webDecl(), loaded: false },
])
deps.watch = { intervalMs: 1_000 }
const registry = createHostWebPluginRegistry(deps)
const before = registry.graph()
;(entries[1] as { fiber?: unknown }).fiber = {}
fsControl.failNextStatPath = join(root, 'late', 'lib', 'client.js')
ctx.emit('internal/plugin', ctx.fiber)
await Promise.resolve()
expect(errors[0]?.message).toContain('staged bundle missing')
expect(registry.graph()).toBe(before)
expect(registry.clientPath('late')).toBeUndefined()
ctx.emit('internal/plugin', ctx.fiber)
await Promise.resolve()
expect(registry.graph().entries.map(row => row.id)).toEqual(['stable', 'late'])
registry.dispose()
})
it('watch mode: a missing bundle forces a re-hash when identical metadata reappears', async () => {
vi.useFakeTimers()
const { deps, root } = makeDeps([{ name: 'watched', pkg: webDecl() }])
const bundle = join(root, 'watched', 'lib', 'client.js')
const fixedTime = new Date(1_600_000_000_000)
utimesSync(bundle, fixedTime, fixedTime)
deps.watch = { intervalMs: 20 }
const registry = createHostWebPluginRegistry(deps)
const baseline = statSync(bundle)
const rebuilds: { id: string; rev: string }[] = []
registry.onRebuilt((id, rev) => rebuilds.push({ id, rev }))
unlinkSync(bundle)
await vi.advanceTimersByTimeAsync(20)
writeFileSync(bundle, 'x'.repeat(baseline.size))
utimesSync(bundle, fixedTime, fixedTime)
const restored = statSync(bundle)
expect({ mtimeMs: restored.mtimeMs, size: restored.size }).toEqual({
mtimeMs: baseline.mtimeMs,
size: baseline.size,
})
await vi.advanceTimersByTimeAsync(20)
expect(rebuilds).toHaveLength(1)
expect(registry.graph().entries[0]?.rev).toBe(rebuilds[0]?.rev)
registry.dispose()
})
it('rejects a non-positive or non-integer watch interval at build time', () => {
for (const intervalMs of [0, -5, 1.5]) {
const { deps } = makeDeps([{ name: 'p', pkg: webDecl() }])