mirror of
https://github.com/stablyai/orca.git
synced 2026-10-20 08:03:28 +00:00
fix(watcher/explorer/relay): coalesce POSIX paths by byte identity; pres
- Remote watcher event coalescing now keeps NFC/NFD-distinct POSIX paths separate while still folding Windows path spellings, fixing cache invalidation when recreating directories with different Unicode compositions. - Relay grace reconfiguration now preserves shutdown-deferred state when grace is set to zero, preventing premature shutdown when the grace timer is reconfigured mid-flight. - File explorer refresh refactored from fixed-wave batching to concurrent task execution with `forEachWithConcurrency`, batching results every N settled reads instead of every wave, and reporting whether cancellation discarded pending work. - Watch handler now resyncs when events arrive after disposal, ensuring refresh requests aren't lost when events race cleanup during worktree switches.
This commit is contained in:
@@ -80,19 +80,20 @@ describe('createRemoteWatcherEventBatch', () => {
|
||||
expect(deliver).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('clears a stale delete when a later update is known to be a file', () => {
|
||||
it('keeps a replacement delete when a directory is recreated as a file', () => {
|
||||
const deliver = vi.fn()
|
||||
const batch = makeBatch(deliver)
|
||||
|
||||
batch.push([
|
||||
{ kind: 'delete', absolutePath: `${ROOT}/a.ts` },
|
||||
{ kind: 'create', absolutePath: `${ROOT}/a.ts` },
|
||||
{ kind: 'update', absolutePath: `${ROOT}/a.ts`, isDirectory: false }
|
||||
{ kind: 'delete', absolutePath: `${ROOT}/dist` },
|
||||
{ kind: 'create', absolutePath: `${ROOT}/dist` },
|
||||
{ kind: 'update', absolutePath: `${ROOT}/dist`, isDirectory: false }
|
||||
])
|
||||
vi.advanceTimersByTime(150)
|
||||
|
||||
expect(deliver.mock.calls[0][0]).toEqual([
|
||||
{ kind: 'update', absolutePath: `${ROOT}/a.ts`, isDirectory: false }
|
||||
{ kind: 'delete', absolutePath: `${ROOT}/dist` },
|
||||
{ kind: 'update', absolutePath: `${ROOT}/dist`, isDirectory: false }
|
||||
])
|
||||
})
|
||||
|
||||
@@ -267,4 +268,39 @@ describe('createRemoteWatcherEventBatch', () => {
|
||||
expect(event.absolutePath).not.toMatch(/^[a-zA-Z]:/)
|
||||
}
|
||||
})
|
||||
|
||||
it('keeps NFC- and NFD-distinct remote POSIX names separate', () => {
|
||||
const deliver = vi.fn()
|
||||
const batch = makeBatch(deliver)
|
||||
const nfcPath = `${ROOT}/caf\u00e9.txt`
|
||||
const nfdPath = `${ROOT}/cafe\u0301.txt`
|
||||
|
||||
batch.push([{ kind: 'update', absolutePath: nfcPath }])
|
||||
batch.push([{ kind: 'update', absolutePath: nfdPath }])
|
||||
vi.advanceTimersByTime(150)
|
||||
|
||||
expect(deliver.mock.calls[0][0]).toEqual([
|
||||
{ kind: 'update', absolutePath: nfcPath },
|
||||
{ kind: 'update', absolutePath: nfdPath }
|
||||
])
|
||||
})
|
||||
|
||||
it('still coalesces equivalent Windows path spellings', () => {
|
||||
const deliver = vi.fn()
|
||||
const batch = createRemoteWatcherEventBatch({
|
||||
rootPath: 'C:\\Repo',
|
||||
deliver,
|
||||
trailingMs: 150,
|
||||
maxWaitMs: 500,
|
||||
maxEvents: 5_000
|
||||
})
|
||||
|
||||
batch.push([{ kind: 'update', absolutePath: 'C:\\Repo\\CAF\u00c9.txt' }])
|
||||
batch.push([{ kind: 'update', absolutePath: 'c:/repo/cafe\u0301.txt' }])
|
||||
vi.advanceTimersByTime(150)
|
||||
|
||||
expect(deliver.mock.calls[0][0]).toEqual([
|
||||
{ kind: 'update', absolutePath: 'c:/repo/cafe\u0301.txt' }
|
||||
])
|
||||
})
|
||||
})
|
||||
|
||||
@@ -1,5 +1,8 @@
|
||||
import type { FsChangeEvent } from '../../shared/types'
|
||||
import { normalizeRuntimePathForComparison } from '../../shared/cross-platform-path'
|
||||
import {
|
||||
isWindowsAbsolutePathLike,
|
||||
normalizeRuntimePathForComparison
|
||||
} from '../../shared/cross-platform-path'
|
||||
|
||||
export type RemoteWatcherEventBatchOptions = {
|
||||
rootPath: string
|
||||
@@ -14,13 +17,18 @@ export type RemoteWatcherEventBatch = {
|
||||
close: () => void
|
||||
}
|
||||
|
||||
// Mirrors coalesceEvents in filesystem-watcher.ts, with two deliberate divergences: a hoisted delete
|
||||
// survives a later event of unknown isDirectory, so delete→create→update over-emits a delete; and
|
||||
// delete→create→delete emits that delete, which local drops — the path is gone, so the renderer needs it.
|
||||
function coalesceRemoteEvents(raw: FsChangeEvent[]): FsChangeEvent[] {
|
||||
// Unlike local coalescing, a recreate keeps its delete through live updates so cached dirs are purged;
|
||||
// delete→create→delete also remains a net delete because the path predated the batch.
|
||||
function posixEventIdentity(absolutePath: string): string {
|
||||
const normalized = absolutePath.replace(/\/+/g, '/')
|
||||
return normalized === '/' ? normalized : normalized.replace(/\/+$/, '')
|
||||
}
|
||||
|
||||
function coalesceRemoteEvents(raw: FsChangeEvent[], rootPath: string): FsChangeEvent[] {
|
||||
const lastByKey = new Map<string, FsChangeEvent>()
|
||||
const deleteBeforeCreate = new Map<string, FsChangeEvent>()
|
||||
const passthrough: FsChangeEvent[] = []
|
||||
const windowsPaths = isWindowsAbsolutePathLike(rootPath)
|
||||
|
||||
for (const event of raw) {
|
||||
if (event.kind !== 'create' && event.kind !== 'update' && event.kind !== 'delete') {
|
||||
@@ -28,10 +36,10 @@ function coalesceRemoteEvents(raw: FsChangeEvent[]): FsChangeEvent[] {
|
||||
passthrough.push(event)
|
||||
continue
|
||||
}
|
||||
// Why: key on the comparison form, never node path.resolve — resolving a remote POSIX path on a
|
||||
// Windows host mints C:\home\u\f, which the renderer's watch canonicalizer then rejects outright.
|
||||
// Residual cost: its NFC fold merges canonically equivalent byte-distinct names on a POSIX host.
|
||||
const key = normalizeRuntimePathForComparison(event.absolutePath)
|
||||
// POSIX event identity keeps byte-distinct names; Windows still folds separators and casing.
|
||||
const key = windowsPaths
|
||||
? normalizeRuntimePathForComparison(event.absolutePath)
|
||||
: posixEventIdentity(event.absolutePath)
|
||||
const prev = lastByKey.get(key)
|
||||
|
||||
if (prev) {
|
||||
@@ -51,9 +59,8 @@ function coalesceRemoteEvents(raw: FsChangeEvent[]): FsChangeEvent[] {
|
||||
|
||||
lastByKey.set(key, event)
|
||||
|
||||
// Why: drop the hoisted delete only when the later event proves the path is a live file, or repeats the
|
||||
// delete. Remote events carry no isDirectory, and discarding it there strands a dead dir's cached subtree.
|
||||
if (event.kind === 'delete' || (event.kind === 'update' && event.isDirectory === false)) {
|
||||
// A final delete subsumes the transition; a live update cannot prove the old entry was not a directory.
|
||||
if (event.kind === 'delete') {
|
||||
deleteBeforeCreate.delete(key)
|
||||
}
|
||||
}
|
||||
@@ -89,7 +96,7 @@ export function createRemoteWatcherEventBatch({
|
||||
deliver([{ kind: 'overflow', absolutePath: rootPath }])
|
||||
return
|
||||
}
|
||||
const coalesced = coalesceRemoteEvents(pending)
|
||||
const coalesced = coalesceRemoteEvents(pending, rootPath)
|
||||
if (coalesced.length > 0) {
|
||||
deliver(coalesced)
|
||||
}
|
||||
|
||||
@@ -1,5 +1,9 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { decideRelayGrace, type RelayGraceDecisionInput } from './relay-grace-branch'
|
||||
import {
|
||||
decideRelayGrace,
|
||||
retryDeferredShutdownAfterGraceReconfigure,
|
||||
type RelayGraceDecisionInput
|
||||
} from './relay-grace-branch'
|
||||
|
||||
const EMPTY_DETACHED_STARTUP_GRACE_MS = 30_000
|
||||
const IDLE_RELAY_GRACE_MS = 15 * 60_000
|
||||
@@ -64,6 +68,17 @@ describe('decideRelayGrace', () => {
|
||||
).toEqual({ branch: 'shutdown-deferred', timeoutMs: IDLE_RELAY_GRACE_MS })
|
||||
})
|
||||
|
||||
it('preserves shutdown-deferred when grace is reconfigured to zero', () => {
|
||||
const current = decide({ retryDeferredShutdown: true })
|
||||
|
||||
expect(
|
||||
decide({
|
||||
configuredGraceMs: 0,
|
||||
retryDeferredShutdown: retryDeferredShutdownAfterGraceReconfigure(current.branch)
|
||||
})
|
||||
).toEqual({ branch: 'shutdown-deferred', timeoutMs: IDLE_RELAY_GRACE_MS })
|
||||
})
|
||||
|
||||
it('prefers startup-empty-detached over the idle cap', () => {
|
||||
expect(
|
||||
decide({
|
||||
|
||||
@@ -23,6 +23,12 @@ export type RelayGraceDecision = {
|
||||
timeoutMs: number
|
||||
}
|
||||
|
||||
export function retryDeferredShutdownAfterGraceReconfigure(
|
||||
currentBranch: RelayGraceBranch | null
|
||||
): boolean {
|
||||
return currentBranch === 'shutdown-deferred'
|
||||
}
|
||||
|
||||
/**
|
||||
* Picks the relay's shutdown grace window.
|
||||
*
|
||||
|
||||
+8
-2
@@ -51,7 +51,11 @@ import {
|
||||
} from '../shared/pi-agent-kind'
|
||||
import { resolveSetupAgentSequenceLaunchCommand } from '../shared/setup-agent-sequencing'
|
||||
import { pickRemoteCliEnv } from './remote-cli-env'
|
||||
import { decideRelayGrace, type RelayGraceBranch } from './relay-grace-branch'
|
||||
import {
|
||||
decideRelayGrace,
|
||||
retryDeferredShutdownAfterGraceReconfigure,
|
||||
type RelayGraceBranch
|
||||
} from './relay-grace-branch'
|
||||
import { relayLogLine } from './relay-diagnostic-log'
|
||||
import { remoteCliRequestTimeoutMs } from './remote-cli-timeout'
|
||||
import { shouldReadRemoteCliStdin } from './remote-cli-stdin'
|
||||
@@ -696,7 +700,9 @@ async function main(): Promise<void> {
|
||||
graceReason !== null &&
|
||||
!shutdownInFlight
|
||||
) {
|
||||
startGrace('grace reconfigured')
|
||||
startGrace('grace reconfigured', {
|
||||
retryDeferredShutdown: retryDeferredShutdownAfterGraceReconfigure(graceBranch)
|
||||
})
|
||||
}
|
||||
}
|
||||
return { graceTimeMs: ptyHandler.configuredGraceTimeMs }
|
||||
|
||||
+87
-13
@@ -45,8 +45,7 @@ describe('refreshFileExplorerExpandedDirs', () => {
|
||||
dirLoadTracker: createFileExplorerDirLoadTracker(),
|
||||
setDirCache,
|
||||
readDirectory,
|
||||
// A limit at or above the dir count must stay a single wave, i.e. exactly
|
||||
// today's two batched writes.
|
||||
// A limit at or above the dir count keeps one result batch.
|
||||
maxConcurrentReads: 16
|
||||
})
|
||||
|
||||
@@ -247,7 +246,7 @@ describe('refreshFileExplorerExpandedDirs', () => {
|
||||
expect(refreshed).toBe(true)
|
||||
expect(peakInFlight).toBeLessThanOrEqual(4)
|
||||
expect(readDirectory).toHaveBeenCalledTimes(20)
|
||||
// One up-front loading write plus one result write per wave (five waves).
|
||||
// One up-front loading write plus one result write per completed group of four.
|
||||
expect(setDirCache).toHaveBeenCalledTimes(6)
|
||||
for (const { dirPath } of dirs) {
|
||||
expect(cache[dirPath]).toMatchObject({
|
||||
@@ -257,7 +256,7 @@ describe('refreshFileExplorerExpandedDirs', () => {
|
||||
}
|
||||
})
|
||||
|
||||
it('marks queued dirs loading up front so only the first wave reads', async () => {
|
||||
it('marks queued dirs loading up front while the initial reads are blocked', async () => {
|
||||
const dirs = Array.from({ length: 9 }, (_, index) => ({
|
||||
dirPath: `/repo/d${index}`,
|
||||
depth: 0
|
||||
@@ -266,13 +265,13 @@ describe('refreshFileExplorerExpandedDirs', () => {
|
||||
const setDirCache = vi.fn((update: CacheUpdate) => {
|
||||
cache = typeof update === 'function' ? update(cache) : update
|
||||
})
|
||||
let releaseFirstWave!: () => void
|
||||
const firstWaveGate = new Promise<void>((resolve) => {
|
||||
releaseFirstWave = resolve
|
||||
let releaseInitialReads!: () => void
|
||||
const initialReadsGate = new Promise<void>((resolve) => {
|
||||
releaseInitialReads = resolve
|
||||
})
|
||||
const readDirectory = vi.fn(async (dirPath: string) => {
|
||||
if (['/repo/d0', '/repo/d1', '/repo/d2'].includes(dirPath)) {
|
||||
await firstWaveGate
|
||||
await initialReadsGate
|
||||
}
|
||||
return { entries: [], operationOwner: { kind: 'local' as const } }
|
||||
})
|
||||
@@ -294,13 +293,89 @@ describe('refreshFileExplorerExpandedDirs', () => {
|
||||
expect(cache['/repo/d6']).toMatchObject({ loading: true })
|
||||
expect(readDirectory).toHaveBeenCalledTimes(3)
|
||||
|
||||
releaseFirstWave()
|
||||
releaseInitialReads()
|
||||
await refreshPromise
|
||||
|
||||
expect(cache['/repo/d6']).toMatchObject({ loading: false })
|
||||
})
|
||||
|
||||
it('drops a later-wave directory superseded while an earlier wave was reading', async () => {
|
||||
it('starts later reads as slots free without waiting for the slowest initial read', async () => {
|
||||
const dirs = Array.from({ length: 5 }, (_, index) => ({
|
||||
dirPath: `/repo/d${index}`,
|
||||
depth: 0
|
||||
}))
|
||||
let cache: Record<string, DirCache> = {}
|
||||
const setDirCache = vi.fn((update: CacheUpdate) => {
|
||||
cache = typeof update === 'function' ? update(cache) : update
|
||||
})
|
||||
let releaseSlowRead!: () => void
|
||||
const slowRead = new Promise<void>((resolve) => {
|
||||
releaseSlowRead = resolve
|
||||
})
|
||||
const readDirectory = vi.fn(async (dirPath: string) => {
|
||||
if (dirPath === '/repo/d0') {
|
||||
await slowRead
|
||||
}
|
||||
return { entries: [entry('x.ts')], operationOwner: { kind: 'local' as const } }
|
||||
})
|
||||
|
||||
const refreshPromise = refreshFileExplorerExpandedDirs({
|
||||
dirs,
|
||||
worktreePath: '/repo',
|
||||
dirLoadTracker: createFileExplorerDirLoadTracker(),
|
||||
setDirCache,
|
||||
readDirectory,
|
||||
maxConcurrentReads: 2
|
||||
})
|
||||
await new Promise((resolve) => setTimeout(resolve, 0))
|
||||
|
||||
expect(readDirectory.mock.calls.map(([dirPath]) => dirPath)).toEqual([
|
||||
'/repo/d0',
|
||||
'/repo/d1',
|
||||
'/repo/d2',
|
||||
'/repo/d3',
|
||||
'/repo/d4'
|
||||
])
|
||||
expect(cache['/repo/d1']).toMatchObject({ loading: false })
|
||||
expect(cache['/repo/d0']).toMatchObject({ loading: true })
|
||||
|
||||
releaseSlowRead()
|
||||
await expect(refreshPromise).resolves.toBe(true)
|
||||
expect(cache['/repo/d0']).toMatchObject({ loading: false })
|
||||
})
|
||||
|
||||
it('does not turn a commit callback failure into an empty directory result', async () => {
|
||||
let cache: Record<string, DirCache> = {}
|
||||
const setDirCache = vi.fn((update: CacheUpdate) => {
|
||||
cache = typeof update === 'function' ? update(cache) : update
|
||||
})
|
||||
const commitError = new Error('commit failed')
|
||||
|
||||
await expect(
|
||||
refreshFileExplorerExpandedDirs({
|
||||
dirs: [{ dirPath: '/repo/src', depth: 0 }],
|
||||
worktreePath: '/repo',
|
||||
dirLoadTracker: createFileExplorerDirLoadTracker(),
|
||||
setDirCache,
|
||||
readDirectory: async () => ({
|
||||
entries: [entry('index.ts')],
|
||||
operationOwner: { kind: 'local' as const }
|
||||
}),
|
||||
maxConcurrentReads: 1,
|
||||
onDirCommitted: () => {
|
||||
throw commitError
|
||||
}
|
||||
})
|
||||
).rejects.toBe(commitError)
|
||||
|
||||
expect(setDirCache).toHaveBeenCalledTimes(2)
|
||||
expect(cache['/repo/src']).toMatchObject({
|
||||
loading: false,
|
||||
children: [{ name: 'index.ts' }]
|
||||
})
|
||||
})
|
||||
|
||||
it('drops a queued directory superseded while an earlier read is blocked', async () => {
|
||||
const tracker = createFileExplorerDirLoadTracker()
|
||||
let cache: Record<string, DirCache> = {}
|
||||
const setDirCache = vi.fn((update: CacheUpdate) => {
|
||||
@@ -330,8 +405,7 @@ describe('refreshFileExplorerExpandedDirs', () => {
|
||||
})
|
||||
await Promise.resolve()
|
||||
|
||||
// A watcher-driven refreshDir supersedes the queued wave-2 dir before it
|
||||
// ever starts reading; its token was already taken up front.
|
||||
// A watcher-driven refreshDir supersedes the queued dir before it starts reading.
|
||||
tracker.begin('/repo/b')
|
||||
const newerBCache: DirCache = { loading: false, children: [] }
|
||||
setDirCache((prev) => ({ ...prev, '/repo/b': newerBCache }))
|
||||
@@ -341,7 +415,7 @@ describe('refreshFileExplorerExpandedDirs', () => {
|
||||
|
||||
expect(refreshed).toBe(false)
|
||||
expect(cache['/repo/a']).toMatchObject({ loading: false, children: [{ name: 'x.ts' }] })
|
||||
// Wave 2 must neither read nor commit the superseded dir.
|
||||
// The queued task must neither read nor commit the superseded dir.
|
||||
expect(readDirectory).toHaveBeenCalledTimes(1)
|
||||
expect(cache['/repo/b']).toEqual(newerBCache)
|
||||
})
|
||||
|
||||
@@ -5,6 +5,7 @@ import {
|
||||
fileExplorerEntriesToTreeNodes,
|
||||
type FileExplorerDirectoryListing
|
||||
} from './file-explorer-directory-listing'
|
||||
import { forEachWithConcurrency } from '../../../../shared/map-with-concurrency'
|
||||
|
||||
export type RefreshFileExplorerTreeDir = {
|
||||
dirPath: string
|
||||
@@ -37,17 +38,23 @@ export async function refreshFileExplorerExpandedDirs({
|
||||
|
||||
const uniqueDirs = Array.from(new Map(dirs.map((dir) => [dir.dirPath, dir])).values())
|
||||
// Why: begin every token before the first read so a concurrent refreshDir or
|
||||
// worktree reset supersedes dirs still queued in a later wave, exactly as the
|
||||
// single-batch version did.
|
||||
// worktree reset supersedes dirs still waiting for a concurrency slot.
|
||||
const loadTokens = new Map(
|
||||
uniqueDirs.map((dir) => [dir.dirPath, dirLoadTracker.begin(dir.dirPath)])
|
||||
)
|
||||
const waveSize = Math.max(1, Math.floor(maxConcurrentReads))
|
||||
const commitBatchSize =
|
||||
maxConcurrentReads === Number.POSITIVE_INFINITY
|
||||
? uniqueDirs.length
|
||||
: Number.isFinite(maxConcurrentReads)
|
||||
? Math.max(1, Math.floor(maxConcurrentReads))
|
||||
: 1
|
||||
const pendingResults: { dirPath: string; cache: DirCache }[] = []
|
||||
let settledSinceCommit = 0
|
||||
let committedDirs = 0
|
||||
|
||||
// Why: mark every dir loading up front, not per wave — FileExplorer's auto-load
|
||||
// Why: mark every dir loading up front — FileExplorer's auto-load
|
||||
// effect re-runs on any `expanded` change and fans out an unbounded loadDir per
|
||||
// dir that is neither cached nor loading, which would defeat the wave cap.
|
||||
// dir that is neither cached nor loading, which would defeat the concurrency cap.
|
||||
setDirCache((prev) => {
|
||||
const next = { ...prev }
|
||||
for (const { dirPath } of uniqueDirs) {
|
||||
@@ -59,65 +66,13 @@ export async function refreshFileExplorerExpandedDirs({
|
||||
return next
|
||||
})
|
||||
|
||||
for (let start = 0; start < uniqueDirs.length; start += waveSize) {
|
||||
// Why: a dir superseded while an earlier wave was reading is left to its
|
||||
// newer load — reading it here would only burn a round trip for a result
|
||||
// that is dropped at commit time.
|
||||
const wave = uniqueDirs
|
||||
.slice(start, start + waveSize)
|
||||
.filter(({ dirPath }) => dirLoadTracker.isCurrent(loadTokens.get(dirPath)!))
|
||||
if (wave.length === 0) {
|
||||
continue
|
||||
}
|
||||
|
||||
// Why: batch each wave's results into one setDirCache write, so the whole
|
||||
// refresh costs 1 + ceil(N / waveSize) cache spreads rather than one per dir.
|
||||
const results = await Promise.all(
|
||||
wave.map(async ({ dirPath, depth }) => {
|
||||
const loadToken = loadTokens.get(dirPath)!
|
||||
try {
|
||||
const listing = await readDirectory(dirPath)
|
||||
if (!dirLoadTracker.isCurrent(loadToken)) {
|
||||
return { current: false as const }
|
||||
}
|
||||
return {
|
||||
current: true as const,
|
||||
dirPath,
|
||||
cache: {
|
||||
children: fileExplorerEntriesToTreeNodes(
|
||||
listing.entries,
|
||||
dirPath,
|
||||
depth,
|
||||
worktreePath,
|
||||
listing.operationOwner
|
||||
),
|
||||
loading: false,
|
||||
operationOwner: listing.operationOwner
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
if (!dirLoadTracker.isCurrent(loadToken)) {
|
||||
return { current: false as const }
|
||||
}
|
||||
return {
|
||||
current: true as const,
|
||||
dirPath,
|
||||
cache: { children: [], loading: false }
|
||||
}
|
||||
}
|
||||
})
|
||||
)
|
||||
|
||||
// Why: the batch commits only after the slowest read in the wave, so a dir
|
||||
// can be superseded (watcher refreshDir, worktree reset) after its own read
|
||||
// resolved. Re-check tokens at commit time so the batched write never
|
||||
// clobbers a newer load — preserving the old per-dir commit ordering.
|
||||
const currentResults = results.filter(
|
||||
(result): result is Extract<typeof result, { current: true }> =>
|
||||
result.current && dirLoadTracker.isCurrent(loadTokens.get(result.dirPath)!)
|
||||
)
|
||||
const commitPendingResults = (): void => {
|
||||
settledSinceCommit = 0
|
||||
const currentResults = pendingResults
|
||||
.splice(0)
|
||||
.filter((result) => dirLoadTracker.isCurrent(loadTokens.get(result.dirPath)!))
|
||||
if (currentResults.length === 0) {
|
||||
continue
|
||||
return
|
||||
}
|
||||
|
||||
setDirCache((prev) => {
|
||||
@@ -133,5 +88,49 @@ export async function refreshFileExplorerExpandedDirs({
|
||||
committedDirs += currentResults.length
|
||||
}
|
||||
|
||||
const settleRead = (result?: { dirPath: string; cache: DirCache }): void => {
|
||||
if (result) {
|
||||
pendingResults.push(result)
|
||||
}
|
||||
settledSinceCommit++
|
||||
if (settledSinceCommit >= commitBatchSize) {
|
||||
commitPendingResults()
|
||||
}
|
||||
}
|
||||
|
||||
await forEachWithConcurrency(uniqueDirs, maxConcurrentReads, async ({ dirPath, depth }) => {
|
||||
const loadToken = loadTokens.get(dirPath)!
|
||||
// A superseding load owns this dir now; do not spend a round trip on a result we must drop.
|
||||
if (!dirLoadTracker.isCurrent(loadToken)) {
|
||||
settleRead()
|
||||
return
|
||||
}
|
||||
let cache: DirCache | undefined
|
||||
try {
|
||||
const listing = await readDirectory(dirPath)
|
||||
if (dirLoadTracker.isCurrent(loadToken)) {
|
||||
cache = {
|
||||
children: fileExplorerEntriesToTreeNodes(
|
||||
listing.entries,
|
||||
dirPath,
|
||||
depth,
|
||||
worktreePath,
|
||||
listing.operationOwner
|
||||
),
|
||||
loading: false,
|
||||
operationOwner: listing.operationOwner
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
if (dirLoadTracker.isCurrent(loadToken)) {
|
||||
cache = { children: [], loading: false }
|
||||
}
|
||||
}
|
||||
settleRead(cache ? { dirPath, cache } : undefined)
|
||||
})
|
||||
if (settledSinceCommit > 0) {
|
||||
commitPendingResults()
|
||||
}
|
||||
|
||||
return committedDirs === uniqueDirs.length
|
||||
}
|
||||
|
||||
+7
-1
@@ -226,13 +226,19 @@ describe('createFileExplorerWatchRefreshScheduler', () => {
|
||||
|
||||
scheduler.requestFullRefresh()
|
||||
scheduler.requestDirRefresh('/repo/src')
|
||||
scheduler.cancel()
|
||||
expect(scheduler.cancel()).toBe(true)
|
||||
await vi.advanceTimersByTimeAsync(MAX_WAIT_MS)
|
||||
|
||||
expect(refreshTree).not.toHaveBeenCalled()
|
||||
expect(refreshDir).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('reports no discarded work when an idle scheduler is cancelled', () => {
|
||||
const { scheduler } = setup()
|
||||
|
||||
expect(scheduler.cancel()).toBe(false)
|
||||
})
|
||||
|
||||
it('stops refreshing queued dirs when cancelled mid-run', async () => {
|
||||
// Regression guard: refreshDir is bound to a live ref that React reassigns
|
||||
// before the effect cleanup runs, so a queued path surviving cancel() would
|
||||
|
||||
@@ -11,7 +11,8 @@ type SchedulerTimer = ReturnType<typeof setTimeout>
|
||||
export type FileExplorerWatchRefreshScheduler = {
|
||||
requestFullRefresh: () => void
|
||||
requestDirRefresh: (dirPath: string) => void
|
||||
cancel: () => void
|
||||
/** True when cancellation discarded a received refresh or interrupted one in flight. */
|
||||
cancel: () => boolean
|
||||
}
|
||||
|
||||
export type CreateFileExplorerWatchRefreshSchedulerParams = {
|
||||
@@ -142,11 +143,14 @@ export function createFileExplorerWatchRefreshScheduler({
|
||||
armForRequest()
|
||||
},
|
||||
cancel: () => {
|
||||
const discardedWork =
|
||||
pendingFull || pendingDirs.size > 0 || timer !== null || inFlight !== null
|
||||
disposed = true
|
||||
clearTimer()
|
||||
pendingFull = false
|
||||
pendingDirs.clear()
|
||||
firstRequestAt = null
|
||||
return discardedWork
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+173
@@ -0,0 +1,173 @@
|
||||
// @vitest-environment happy-dom
|
||||
|
||||
import { act, cleanup, renderHook } from '@testing-library/react'
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import type { FsChangedPayload } from '../../../../shared/types'
|
||||
import type {
|
||||
FileExplorerOperationOwner,
|
||||
FileExplorerTreeRefreshOutcome
|
||||
} from './file-explorer-types'
|
||||
|
||||
const ownerRef = vi.hoisted(() => ({ current: { kind: 'local' } as FileExplorerOperationOwner }))
|
||||
const runtimeWatch = vi.hoisted(() => ({
|
||||
handler: null as ((payload: FsChangedPayload) => void) | null,
|
||||
subscribe: vi.fn()
|
||||
}))
|
||||
|
||||
vi.mock('@/store', () => ({
|
||||
useAppStore: Object.assign((selector: (state: unknown) => unknown) => selector({}), {
|
||||
getState: () => ({})
|
||||
})
|
||||
}))
|
||||
vi.mock('./file-explorer-operation-owner', () => ({
|
||||
getFileExplorerOperationOwner: () => ownerRef.current,
|
||||
getFileExplorerOperationOwnerFromState: () => ownerRef.current
|
||||
}))
|
||||
vi.mock('@/runtime/runtime-file-client', () => ({
|
||||
subscribeRuntimeFileChanges: runtimeWatch.subscribe
|
||||
}))
|
||||
|
||||
import { useFileExplorerWatch } from './useFileExplorerWatch'
|
||||
|
||||
type WatchHandler = (payload: FsChangedPayload) => void
|
||||
|
||||
describe('useFileExplorerWatch pending refreshes', () => {
|
||||
let mainWatchHandler: WatchHandler | null
|
||||
let refreshDir: ReturnType<typeof vi.fn<(dirPath: string) => Promise<void>>>
|
||||
let refreshTree: ReturnType<typeof vi.fn<() => Promise<FileExplorerTreeRefreshOutcome>>>
|
||||
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers()
|
||||
ownerRef.current = { kind: 'local' }
|
||||
mainWatchHandler = null
|
||||
runtimeWatch.handler = null
|
||||
runtimeWatch.subscribe.mockReset()
|
||||
runtimeWatch.subscribe.mockImplementation(
|
||||
async (_context: unknown, handler: WatchHandler): Promise<() => void> => {
|
||||
runtimeWatch.handler = handler
|
||||
return () => undefined
|
||||
}
|
||||
)
|
||||
refreshDir = vi.fn(async () => {})
|
||||
refreshTree = vi.fn(async () => 'refreshed' as const)
|
||||
Object.defineProperty(window, 'api', {
|
||||
configurable: true,
|
||||
value: {
|
||||
fs: {
|
||||
onFsChanged: vi.fn((handler: WatchHandler) => {
|
||||
mainWatchHandler = handler
|
||||
return vi.fn()
|
||||
})
|
||||
}
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
cleanup()
|
||||
vi.useRealTimers()
|
||||
})
|
||||
|
||||
function renderWatch(worktreePath: string | null = '/repo') {
|
||||
return renderHook(
|
||||
({ visiblePath }: { visiblePath: string | null }) =>
|
||||
useFileExplorerWatch({
|
||||
worktreePath: visiblePath,
|
||||
activeWorktreeId: 'wt-1',
|
||||
dirCache: { '/repo': { children: [], loading: false } },
|
||||
setDirCache: vi.fn(),
|
||||
expanded: new Set(),
|
||||
setSelectedPath: vi.fn(),
|
||||
refreshDir,
|
||||
refreshTree,
|
||||
inlineInput: null,
|
||||
dragSourcePath: null,
|
||||
isNativeDragOver: false,
|
||||
operationOwner: ownerRef.current
|
||||
}),
|
||||
{ initialProps: { visiblePath: worktreePath } }
|
||||
)
|
||||
}
|
||||
|
||||
function emit(handler: WatchHandler): void {
|
||||
handler({
|
||||
worktreePath: '/repo',
|
||||
events: [{ kind: 'create', absolutePath: '/repo/new.ts', isDirectory: false }]
|
||||
})
|
||||
}
|
||||
|
||||
it('resyncs on reopen when hiding Files cancelled a received remote event', async () => {
|
||||
ownerRef.current = { kind: 'ssh', connectionId: 'ssh-1' }
|
||||
const hook = renderWatch()
|
||||
expect(mainWatchHandler).not.toBeNull()
|
||||
|
||||
act(() => emit(mainWatchHandler!))
|
||||
hook.rerender({ visiblePath: null })
|
||||
hook.rerender({ visiblePath: '/repo' })
|
||||
await act(async () => vi.advanceTimersByTimeAsync(0))
|
||||
|
||||
expect(refreshTree).toHaveBeenCalledOnce()
|
||||
expect(refreshDir).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('flushes main SSH batches on the next turn without another debounce window', async () => {
|
||||
ownerRef.current = { kind: 'ssh', connectionId: 'ssh-1' }
|
||||
renderWatch()
|
||||
|
||||
act(() => emit(mainWatchHandler!))
|
||||
expect(refreshDir).not.toHaveBeenCalled()
|
||||
await act(async () => vi.advanceTimersByTimeAsync(0))
|
||||
|
||||
expect(refreshDir).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
it('flushes runtime RPC batches on the next turn without another debounce window', async () => {
|
||||
ownerRef.current = {
|
||||
kind: 'runtime',
|
||||
environmentId: 'runtime-1',
|
||||
executionHostId: 'runtime:runtime-1'
|
||||
}
|
||||
renderWatch()
|
||||
expect(runtimeWatch.handler).not.toBeNull()
|
||||
|
||||
act(() => emit(runtimeWatch.handler!))
|
||||
expect(refreshDir).not.toHaveBeenCalled()
|
||||
await act(async () => vi.advanceTimersByTimeAsync(0))
|
||||
|
||||
expect(refreshDir).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
it('resyncs after an event races cleanup of a pending runtime subscription', async () => {
|
||||
ownerRef.current = {
|
||||
kind: 'runtime',
|
||||
environmentId: 'runtime-1',
|
||||
executionHostId: 'runtime:runtime-1'
|
||||
}
|
||||
let releaseSubscription!: () => void
|
||||
const subscriptionReady = new Promise<void>((resolve) => {
|
||||
releaseSubscription = resolve
|
||||
})
|
||||
runtimeWatch.subscribe.mockImplementation(
|
||||
async (_context: unknown, handler: WatchHandler): Promise<() => void> => {
|
||||
runtimeWatch.handler = handler
|
||||
await subscriptionReady
|
||||
return () => undefined
|
||||
}
|
||||
)
|
||||
const hook = renderWatch()
|
||||
const disposedHandler = runtimeWatch.handler!
|
||||
|
||||
hook.rerender({ visiblePath: null })
|
||||
hook.rerender({ visiblePath: '/repo' })
|
||||
act(() => emit(disposedHandler))
|
||||
await act(async () => vi.advanceTimersByTimeAsync(0))
|
||||
|
||||
expect(refreshTree).toHaveBeenCalledOnce()
|
||||
expect(refreshDir).not.toHaveBeenCalled()
|
||||
|
||||
await act(async () => {
|
||||
releaseSubscription()
|
||||
await subscriptionReady
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -102,9 +102,6 @@ export function useFileExplorerWatch({
|
||||
const expandedRef = useRef(expanded)
|
||||
expandedRef.current = expanded
|
||||
|
||||
const worktreeIdRef = useRef(activeWorktreeId)
|
||||
worktreeIdRef.current = activeWorktreeId
|
||||
|
||||
const inlineInputRef = useRef(inlineInput)
|
||||
inlineInputRef.current = inlineInput
|
||||
|
||||
@@ -123,6 +120,8 @@ export function useFileExplorerWatch({
|
||||
|
||||
// Deferred events queue: events that arrive during inline input or drag
|
||||
const deferredRef = useRef<FsChangedPayload[]>([])
|
||||
const resyncWatchKeysRef = useRef(new Set<string>())
|
||||
const activeResyncByWatchKeyRef = useRef(new Map<string, () => void>())
|
||||
|
||||
// Why: a ref bridges processPayload to the flush effect so it can replay deferred payloads without re-subscribing (design §6.2).
|
||||
const processPayloadRef = useRef<((payload: FsChangedPayload) => void) | null>(null)
|
||||
@@ -134,12 +133,19 @@ export function useFileExplorerWatch({
|
||||
}
|
||||
|
||||
const currentWorktreePath = worktreePath
|
||||
const currentWorktreeId = activeWorktreeId
|
||||
const resyncWatchKeys = resyncWatchKeysRef.current
|
||||
const activeResyncByWatchKey = activeResyncByWatchKeyRef.current
|
||||
const currentWatchKey = JSON.stringify([
|
||||
currentWorktreeId,
|
||||
normalizeRuntimePathForComparison(currentWorktreePath)
|
||||
])
|
||||
|
||||
// Why: one scheduler per subscription covers BOTH remote transports (the
|
||||
// Electron fs:changed bus and the runtime-RPC subscription below), and its
|
||||
// lifetime is tied to the watched worktree so a switch can't refresh the
|
||||
// previous root.
|
||||
const owner = getFileExplorerOperationOwner(worktreeIdRef.current)
|
||||
const owner = getFileExplorerOperationOwner(currentWorktreeId)
|
||||
const scheduler = createFileExplorerWatchRefreshScheduler({
|
||||
refreshTree: () => refreshTreeRef.current(),
|
||||
refreshDir: (dirPath) => refreshDirRef.current(dirPath),
|
||||
@@ -152,21 +158,24 @@ export function useFileExplorerWatch({
|
||||
normalizeRuntimePathForComparison(currentWorktreePath) ||
|
||||
expandedRef.current.has(dirPath),
|
||||
dirConcurrency: fileExplorerRefreshConcurrency(owner),
|
||||
// Why: main already coalesced the burst on this same 150/500 window before it reached the
|
||||
// fs:changed bus, so a second debounce here only doubles paint latency. Zero still arms a 0ms
|
||||
// timer, keeping intra-payload dedup and the concurrency cap.
|
||||
...(owner.kind === 'local' ? { trailingMs: 0, maxWaitMs: 0 } : {})
|
||||
// Every producer batches before delivery; a 0ms turn keeps renderer dedup without another 150ms.
|
||||
trailingMs: 0,
|
||||
maxWaitMs: 0
|
||||
})
|
||||
activeResyncByWatchKey.set(currentWatchKey, scheduler.requestFullRefresh)
|
||||
|
||||
if (resyncWatchKeys.delete(currentWatchKey)) {
|
||||
scheduler.requestFullRefresh()
|
||||
}
|
||||
|
||||
function processPayload(payload: FsChangedPayload): void {
|
||||
const wtId = worktreeIdRef.current
|
||||
if (!wtId) {
|
||||
if (!currentWorktreeId) {
|
||||
return
|
||||
}
|
||||
processFileExplorerFsPayload({
|
||||
payload,
|
||||
currentWorktreePath,
|
||||
worktreeId: wtId,
|
||||
worktreeId: currentWorktreeId,
|
||||
cache: dirCacheRef.current,
|
||||
expanded: expandedRef.current,
|
||||
setDirCache,
|
||||
@@ -180,6 +189,20 @@ export function useFileExplorerWatch({
|
||||
processPayloadRef.current = processPayload
|
||||
|
||||
const handleFsChanged = (payload: FsChangedPayload): void => {
|
||||
if (disposed) {
|
||||
if (
|
||||
normalizeRuntimePathForComparison(payload.worktreePath) ===
|
||||
normalizeRuntimePathForComparison(currentWorktreePath)
|
||||
) {
|
||||
const requestActiveResync = activeResyncByWatchKey.get(currentWatchKey)
|
||||
if (requestActiveResync) {
|
||||
requestActiveResync()
|
||||
} else {
|
||||
resyncWatchKeys.add(currentWatchKey)
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
// Why: defer refreshes during inline input/drag so rows don't shift; native drags only set isNativeDragOver (design §6.2).
|
||||
if (
|
||||
inlineInputRef.current !== null ||
|
||||
@@ -234,7 +257,14 @@ export function useFileExplorerWatch({
|
||||
return () => {
|
||||
disposed = true
|
||||
unsubscribeListener?.()
|
||||
scheduler.cancel()
|
||||
if (activeResyncByWatchKey.get(currentWatchKey) === scheduler.requestFullRefresh) {
|
||||
activeResyncByWatchKey.delete(currentWatchKey)
|
||||
}
|
||||
const hadDeferredEvents = deferredRef.current.length > 0
|
||||
if (scheduler.cancel() || hadDeferredEvents) {
|
||||
// The tree cache survives Files being hidden, so remember work canceled after receipt.
|
||||
resyncWatchKeys.add(currentWatchKey)
|
||||
}
|
||||
deferredRef.current = []
|
||||
processPayloadRef.current = null
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user