mirror of
https://github.com/stablyai/orca.git
synced 2026-10-20 00:03:45 +00:00
fix(watcher/explorer): resilient commit batches on callback error
Cache writes precede callbacks, so a throwing callback cannot strand the batch with stale marks. All callbacks complete despite errors, with the first error thrown after. - Fix commitBatchSize calculation for empty dirs - Add scheduler discard-report test - Move disposed variable before handleFsChanged - Expand batching-strategy comments
This commit is contained in:
@@ -69,7 +69,9 @@ describe('decideRelayGrace', () => {
|
||||
})
|
||||
|
||||
it('preserves shutdown-deferred when grace is reconfigured to zero', () => {
|
||||
const current = decide({ retryDeferredShutdown: true })
|
||||
// Why a non-zero starting grace: configureRelayGraceTime only re-arms on an actual change,
|
||||
// so the reconfiguration this models must cross a real boundary, not 0 → 0.
|
||||
const current = decide({ configuredGraceMs: 10_000, retryDeferredShutdown: true })
|
||||
|
||||
expect(
|
||||
decide({
|
||||
|
||||
@@ -375,6 +375,44 @@ describe('refreshFileExplorerExpandedDirs', () => {
|
||||
})
|
||||
})
|
||||
|
||||
it('still notifies the rest of a commit batch after one commit callback throws', 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')
|
||||
const onDirCommitted = vi.fn((dirPath: string) => {
|
||||
if (dirPath === '/repo/a') {
|
||||
throw commitError
|
||||
}
|
||||
})
|
||||
|
||||
await expect(
|
||||
refreshFileExplorerExpandedDirs({
|
||||
dirs: [
|
||||
{ dirPath: '/repo/a', depth: 0 },
|
||||
{ dirPath: '/repo/b', depth: 0 }
|
||||
],
|
||||
worktreePath: '/repo',
|
||||
dirLoadTracker: createFileExplorerDirLoadTracker(),
|
||||
setDirCache,
|
||||
readDirectory: async () => ({
|
||||
entries: [entry('index.ts')],
|
||||
operationOwner: { kind: 'local' as const }
|
||||
}),
|
||||
// Both dirs land in one commit batch, so the throw must not strand the other's mark.
|
||||
maxConcurrentReads: 2,
|
||||
onDirCommitted
|
||||
})
|
||||
).rejects.toBe(commitError)
|
||||
|
||||
expect(onDirCommitted.mock.calls.map(([dirPath]) => dirPath).sort()).toEqual([
|
||||
'/repo/a',
|
||||
'/repo/b'
|
||||
])
|
||||
expect(cache['/repo/b']).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> = {}
|
||||
|
||||
@@ -44,7 +44,7 @@ export async function refreshFileExplorerExpandedDirs({
|
||||
)
|
||||
const commitBatchSize =
|
||||
maxConcurrentReads === Number.POSITIVE_INFINITY
|
||||
? uniqueDirs.length
|
||||
? Math.max(1, uniqueDirs.length)
|
||||
: Number.isFinite(maxConcurrentReads)
|
||||
? Math.max(1, Math.floor(maxConcurrentReads))
|
||||
: 1
|
||||
@@ -82,10 +82,24 @@ export async function refreshFileExplorerExpandedDirs({
|
||||
}
|
||||
return next
|
||||
})
|
||||
for (const result of currentResults) {
|
||||
onDirCommitted?.(result.dirPath)
|
||||
}
|
||||
committedDirs += currentResults.length
|
||||
// Why: the cache write above already landed for every result, so a throwing callback must not
|
||||
// strand the rest of the batch with a staleness mark no later commit will clear.
|
||||
let firstCommitError: unknown
|
||||
let commitFailed = false
|
||||
for (const result of currentResults) {
|
||||
try {
|
||||
onDirCommitted?.(result.dirPath)
|
||||
} catch (error) {
|
||||
if (!commitFailed) {
|
||||
commitFailed = true
|
||||
firstCommitError = error
|
||||
}
|
||||
}
|
||||
}
|
||||
if (commitFailed) {
|
||||
throw firstCommitError
|
||||
}
|
||||
}
|
||||
|
||||
const settleRead = (result?: { dirPath: string; cache: DirCache }): void => {
|
||||
|
||||
+15
@@ -239,6 +239,21 @@ describe('createFileExplorerWatchRefreshScheduler', () => {
|
||||
expect(scheduler.cancel()).toBe(false)
|
||||
})
|
||||
|
||||
it('reports discarded work when cancelled with a refresh in flight and nothing queued', async () => {
|
||||
const gate = createDeferred()
|
||||
const { refreshTree, scheduler } = setup({ refreshTree: () => gate.promise })
|
||||
|
||||
scheduler.requestFullRefresh()
|
||||
await vi.advanceTimersByTimeAsync(TRAILING_MS)
|
||||
expect(refreshTree).toHaveBeenCalledTimes(1)
|
||||
|
||||
// Nothing pending and no timer armed: the in-flight run is the only discarded work.
|
||||
expect(scheduler.cancel()).toBe(true)
|
||||
|
||||
gate.resolve()
|
||||
await vi.advanceTimersByTimeAsync(MAX_WAIT_MS)
|
||||
})
|
||||
|
||||
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
|
||||
|
||||
@@ -158,7 +158,10 @@ export function useFileExplorerWatch({
|
||||
normalizeRuntimePathForComparison(currentWorktreePath) ||
|
||||
expandedRef.current.has(dirPath),
|
||||
dirConcurrency: fileExplorerRefreshConcurrency(owner),
|
||||
// Every producer batches before delivery; a 0ms turn keeps renderer dedup without another 150ms.
|
||||
// Why 0ms for every transport: each producer already flushes on the shared
|
||||
// WATCH_BATCH_* window — local/SSH via filesystem-watcher, runtime files.watch via
|
||||
// createFileWatchEventBatcher — so a second 150ms here only adds latency, not coalescing.
|
||||
// A new transport must batch at the producer before joining this path.
|
||||
trailingMs: 0,
|
||||
maxWaitMs: 0
|
||||
})
|
||||
@@ -188,6 +191,9 @@ export function useFileExplorerWatch({
|
||||
// Why: expose processPayload to the flush effect so it can replay deferred payloads without re-subscribing.
|
||||
processPayloadRef.current = processPayload
|
||||
|
||||
// Declared before handleFsChanged so its `disposed` read can never hit the temporal dead zone.
|
||||
let disposed = false
|
||||
|
||||
const handleFsChanged = (payload: FsChangedPayload): void => {
|
||||
if (disposed) {
|
||||
if (
|
||||
@@ -216,7 +222,6 @@ export function useFileExplorerWatch({
|
||||
processPayload(payload)
|
||||
}
|
||||
|
||||
let disposed = false
|
||||
let unsubscribeListener: (() => void) | null = null
|
||||
if (activeRuntimeEnvironmentId?.trim() && activeWorktreeId) {
|
||||
// Why: remote runtime watch events don't enter the local Electron fs:changed bus, so subscribe directly.
|
||||
|
||||
Reference in New Issue
Block a user