mirror of
https://github.com/stablyai/orca.git
synced 2026-10-20 00:03:45 +00:00
fix(worktrees): prevent deletion from blocking Orca
Worktree deletion can now proceed even if filesystem watchers or history cleanup operations hang, preventing Orca from freezing. Changes: - Fence install slots with tokens instead of counters so removals can abandon wedged installs without corrupting later removals - Timeout-bound watcher unsubscribe operations with a shared drain budget - Move JSON serialization of large usage caches from queue-time to write-time to avoid blocking main thread - Async tombstone + schedule history tree deletion instead of blocking recursive rmSync during GC, preventing main-thread stalls ~10s after startup
This commit is contained in:
@@ -389,11 +389,8 @@ export class ClaudeUsageStore {
|
||||
}
|
||||
|
||||
private writeToDisk(): Promise<void> {
|
||||
// Why: snapshot synchronously so a concurrent setEnabled/scan cannot tear the JSON mid-stringify.
|
||||
// Pretty-print preserved: humans inspect this analytics cache on disk.
|
||||
const payload = JSON.stringify(this.state, null, 2)
|
||||
const generation = ++this.writeGeneration
|
||||
const write = this.pendingWrite.then(() => this.commitToDisk(payload, generation))
|
||||
const write = this.pendingWrite.then(() => this.commitToDisk(generation))
|
||||
// Why: keep the queue usable after a failure — the awaiting caller still sees the rejection.
|
||||
this.pendingWrite = write.catch((error: unknown) => {
|
||||
console.error('[claude-usage] Failed to persist usage cache:', error)
|
||||
@@ -401,11 +398,15 @@ export class ClaudeUsageStore {
|
||||
return write
|
||||
}
|
||||
|
||||
private async commitToDisk(payload: string, generation: number): Promise<void> {
|
||||
private async commitToDisk(generation: number): Promise<void> {
|
||||
// Why: a newer snapshot is already queued behind this one, so writing 20 MB here would be wasted.
|
||||
if (generation !== this.writeGeneration) {
|
||||
return
|
||||
}
|
||||
// Why stringify here and not at queue time: it blocks the main process for a multi-MB cache, and
|
||||
// superseded generations must not pay for it. Synchronous, so no mutation can tear the JSON.
|
||||
// Pretty-print preserved: humans inspect this analytics cache on disk.
|
||||
const payload = JSON.stringify(this.state, null, 2)
|
||||
const usageFile = getClaudeUsageFile()
|
||||
await mkdir(dirname(usageFile), { recursive: true }).catch(() => {})
|
||||
// Why: atomic temp-file + durable rename so a crash cannot leave a truncated analytics file.
|
||||
|
||||
@@ -407,10 +407,8 @@ export class CodexUsageStore {
|
||||
}
|
||||
|
||||
private writeToDisk(): Promise<void> {
|
||||
// Why: snapshot synchronously so a concurrent setEnabled/scan cannot tear the JSON mid-stringify.
|
||||
const payload = JSON.stringify(this.state)
|
||||
const generation = ++this.writeGeneration
|
||||
const write = this.pendingWrite.then(() => this.commitToDisk(payload, generation))
|
||||
const write = this.pendingWrite.then(() => this.commitToDisk(generation))
|
||||
// Why: keep the queue usable after a failure — the awaiting caller still sees the rejection.
|
||||
this.pendingWrite = write.catch((error: unknown) => {
|
||||
console.error('[codex-usage] Failed to persist usage cache:', error)
|
||||
@@ -418,11 +416,14 @@ export class CodexUsageStore {
|
||||
return write
|
||||
}
|
||||
|
||||
private async commitToDisk(payload: string, generation: number): Promise<void> {
|
||||
private async commitToDisk(generation: number): Promise<void> {
|
||||
// Why: a newer snapshot is already queued behind this one, so writing 60 MB here would be wasted.
|
||||
if (generation !== this.writeGeneration) {
|
||||
return
|
||||
}
|
||||
// Why stringify here and not at queue time: it blocks the main process for a multi-MB cache, and
|
||||
// superseded generations must not pay for it. Synchronous, so no mutation can tear the JSON.
|
||||
const payload = JSON.stringify(this.state)
|
||||
const usageFile = getCodexUsageFile()
|
||||
await mkdir(dirname(usageFile), { recursive: true }).catch(() => {})
|
||||
await writeFileDurableIfCurrent(
|
||||
|
||||
@@ -42,10 +42,6 @@ import {
|
||||
registerFilesystemWatcherHandlers,
|
||||
restoreLocalWatcherAfterFailedRemoval
|
||||
} from './filesystem-watcher'
|
||||
import {
|
||||
createWatcherRemovalDeadline,
|
||||
WATCHER_REMOVAL_DRAIN_BUDGET_MS
|
||||
} from './watcher-removal-drain'
|
||||
import { stat } from 'node:fs/promises'
|
||||
import { subscribe as subscribeParcelWatcher } from '@parcel/watcher'
|
||||
import { subscribeViaWatcherProcess } from './parcel-watcher-process'
|
||||
@@ -705,85 +701,6 @@ describe('local filesystem watcher unsubscribe cleanup', () => {
|
||||
expect(unsubscribeMock).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('bounds a wedged watcher install so worktree deletion cannot hang forever', async () => {
|
||||
vi.useFakeTimers()
|
||||
try {
|
||||
vi.mocked(stat).mockResolvedValue({ isDirectory: () => true } as never)
|
||||
// Why mock the process-backed subscribe: the in-process fallback rejects on abort, so only a
|
||||
// subscribe that ignores the abort signal exercises the deadline rather than the cancel path.
|
||||
// Once, so the wedge cannot leak into the next test.
|
||||
vi.mocked(subscribeViaWatcherProcess).mockImplementationOnce(() => new Promise(() => {}))
|
||||
const sender = {
|
||||
isDestroyed: () => false,
|
||||
send: vi.fn(),
|
||||
once: vi.fn(),
|
||||
id: 1
|
||||
}
|
||||
|
||||
const watchPromise = handlers['fs:watchWorktree'](
|
||||
{ sender },
|
||||
{ worktreePath: '/tmp/repo' }
|
||||
) as Promise<unknown>
|
||||
await vi.waitFor(() => {
|
||||
expect(subscribeViaWatcherProcess).toHaveBeenCalled()
|
||||
})
|
||||
|
||||
let closed = false
|
||||
const closePromise = closeLocalWatcherForWorktreePath('/tmp/repo').then(() => {
|
||||
closed = true
|
||||
})
|
||||
|
||||
await vi.advanceTimersByTimeAsync(WATCHER_REMOVAL_DRAIN_BUDGET_MS - 1)
|
||||
expect(closed).toBe(false)
|
||||
|
||||
await vi.advanceTimersByTimeAsync(1)
|
||||
await closePromise
|
||||
expect(closed).toBe(true)
|
||||
|
||||
void watchPromise
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('spends one shared budget across both drains instead of one per await', async () => {
|
||||
vi.useFakeTimers()
|
||||
try {
|
||||
vi.mocked(stat).mockResolvedValue({ isDirectory: () => true } as never)
|
||||
vi.mocked(subscribeViaWatcherProcess).mockImplementationOnce(() => new Promise(() => {}))
|
||||
const sender = {
|
||||
isDestroyed: () => false,
|
||||
send: vi.fn(),
|
||||
once: vi.fn(),
|
||||
id: 1
|
||||
}
|
||||
|
||||
const watchPromise = handlers['fs:watchWorktree'](
|
||||
{ sender },
|
||||
{ worktreePath: '/tmp/repo' }
|
||||
) as Promise<unknown>
|
||||
await vi.waitFor(() => {
|
||||
expect(subscribeViaWatcherProcess).toHaveBeenCalled()
|
||||
})
|
||||
|
||||
// Half the budget is already gone before the close starts; the drain may only spend the rest.
|
||||
const deadline = createWatcherRemovalDeadline()
|
||||
await vi.advanceTimersByTimeAsync(WATCHER_REMOVAL_DRAIN_BUDGET_MS / 2)
|
||||
|
||||
let closed = false
|
||||
const closePromise = closeLocalWatcherForWorktreePath('/tmp/repo', deadline).then(() => {
|
||||
closed = true
|
||||
})
|
||||
await vi.advanceTimersByTimeAsync(WATCHER_REMOVAL_DRAIN_BUDGET_MS / 2)
|
||||
await closePromise
|
||||
expect(closed).toBe(true)
|
||||
|
||||
void watchPromise
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('re-arms active local listeners after worktree deletion fails', async () => {
|
||||
vi.mocked(stat).mockResolvedValue({ isDirectory: () => true } as never)
|
||||
const firstUnsubscribe = vi.fn()
|
||||
|
||||
@@ -0,0 +1,189 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import type * as ParcelWatcherProcess from './parcel-watcher-process'
|
||||
|
||||
const { handleMock } = vi.hoisted(() => ({
|
||||
handleMock: vi.fn()
|
||||
}))
|
||||
|
||||
vi.mock('electron', () => ({
|
||||
ipcMain: {
|
||||
handle: handleMock
|
||||
}
|
||||
}))
|
||||
|
||||
vi.mock('fs/promises', () => ({
|
||||
stat: vi.fn()
|
||||
}))
|
||||
|
||||
vi.mock('@parcel/watcher', () => ({
|
||||
subscribe: vi.fn()
|
||||
}))
|
||||
|
||||
vi.mock('./filesystem-watcher-wsl', () => ({
|
||||
createWslWatcher: vi.fn()
|
||||
}))
|
||||
|
||||
vi.mock('./parcel-watcher-process', async (importOriginal) => {
|
||||
const actual = await importOriginal<typeof ParcelWatcherProcess>()
|
||||
return {
|
||||
...actual,
|
||||
subscribeViaWatcherProcess: vi.fn(actual.subscribeViaWatcherProcess)
|
||||
}
|
||||
})
|
||||
|
||||
vi.mock('../providers/ssh-filesystem-dispatch', () => ({
|
||||
getSshFilesystemProvider: vi.fn(),
|
||||
onSshFilesystemProviderRegistered: () => () => {}
|
||||
}))
|
||||
|
||||
import {
|
||||
closeAllWatchers,
|
||||
closeLocalWatcherForWorktreePath,
|
||||
registerFilesystemWatcherHandlers
|
||||
} from './filesystem-watcher'
|
||||
import {
|
||||
createWatcherRemovalDeadline,
|
||||
WATCHER_REMOVAL_DRAIN_BUDGET_MS
|
||||
} from './watcher-removal-drain'
|
||||
import { stat } from 'node:fs/promises'
|
||||
import { subscribe as subscribeParcelWatcher } from '@parcel/watcher'
|
||||
import { subscribeViaWatcherProcess } from './parcel-watcher-process'
|
||||
|
||||
type HandlerMap = Record<string, (_event: unknown, args: unknown) => Promise<unknown> | unknown>
|
||||
|
||||
describe('local filesystem watcher removal deadline', () => {
|
||||
const handlers: HandlerMap = {}
|
||||
|
||||
beforeEach(async () => {
|
||||
handleMock.mockReset()
|
||||
vi.mocked(stat).mockReset()
|
||||
vi.mocked(subscribeParcelWatcher).mockReset()
|
||||
vi.mocked(subscribeViaWatcherProcess).mockClear()
|
||||
for (const key of Object.keys(handlers)) {
|
||||
delete handlers[key]
|
||||
}
|
||||
handleMock.mockImplementation((channel, handler) => {
|
||||
handlers[channel] = handler
|
||||
})
|
||||
registerFilesystemWatcherHandlers()
|
||||
await closeAllWatchers()
|
||||
})
|
||||
|
||||
afterEach(async () => {
|
||||
await closeAllWatchers()
|
||||
})
|
||||
|
||||
it('bounds a wedged watcher install so worktree deletion cannot hang forever', async () => {
|
||||
vi.useFakeTimers()
|
||||
try {
|
||||
vi.mocked(stat).mockResolvedValue({ isDirectory: () => true } as never)
|
||||
// Why mock the process-backed subscribe: the in-process fallback rejects on abort, so only a
|
||||
// subscribe that ignores the abort signal exercises the deadline rather than the cancel path.
|
||||
// Once, so the wedge cannot leak into the next test.
|
||||
vi.mocked(subscribeViaWatcherProcess).mockImplementationOnce(() => new Promise(() => {}))
|
||||
const sender = {
|
||||
isDestroyed: () => false,
|
||||
send: vi.fn(),
|
||||
once: vi.fn(),
|
||||
id: 1
|
||||
}
|
||||
|
||||
const watchPromise = handlers['fs:watchWorktree'](
|
||||
{ sender },
|
||||
{ worktreePath: '/tmp/repo' }
|
||||
) as Promise<unknown>
|
||||
await vi.waitFor(() => {
|
||||
expect(subscribeViaWatcherProcess).toHaveBeenCalled()
|
||||
})
|
||||
|
||||
let closed = false
|
||||
const closePromise = closeLocalWatcherForWorktreePath('/tmp/repo').then(() => {
|
||||
closed = true
|
||||
})
|
||||
|
||||
await vi.advanceTimersByTimeAsync(WATCHER_REMOVAL_DRAIN_BUDGET_MS - 1)
|
||||
expect(closed).toBe(false)
|
||||
|
||||
await vi.advanceTimersByTimeAsync(1)
|
||||
await closePromise
|
||||
expect(closed).toBe(true)
|
||||
|
||||
void watchPromise
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('bounds a wedged live unsubscribe so worktree deletion cannot hang forever', async () => {
|
||||
vi.mocked(stat).mockResolvedValue({ isDirectory: () => true } as never)
|
||||
// The in-process Parcel fallback has no unsubscribe timeout of its own, so only the shared
|
||||
// removal deadline can stop this from hanging delete forever.
|
||||
let resolveUnsubscribe: () => void = () => {}
|
||||
const unsubscribeMock = vi.fn(
|
||||
() =>
|
||||
new Promise<void>((resolve) => {
|
||||
resolveUnsubscribe = resolve
|
||||
})
|
||||
)
|
||||
vi.mocked(subscribeParcelWatcher).mockResolvedValue({ unsubscribe: unsubscribeMock } as never)
|
||||
const sender = { isDestroyed: () => false, send: vi.fn(), once: vi.fn(), id: 1 }
|
||||
|
||||
await handlers['fs:watchWorktree']({ sender }, { worktreePath: '/tmp/repo' })
|
||||
|
||||
vi.useFakeTimers()
|
||||
try {
|
||||
let closed = false
|
||||
const closePromise = closeLocalWatcherForWorktreePath('/tmp/repo').then(() => {
|
||||
closed = true
|
||||
})
|
||||
await vi.advanceTimersByTimeAsync(WATCHER_REMOVAL_DRAIN_BUDGET_MS - 1)
|
||||
expect(unsubscribeMock).toHaveBeenCalledTimes(1)
|
||||
expect(closed).toBe(false)
|
||||
|
||||
await vi.advanceTimersByTimeAsync(1)
|
||||
await closePromise
|
||||
expect(closed).toBe(true)
|
||||
} finally {
|
||||
resolveUnsubscribe()
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('spends one shared budget across both drains instead of one per await', async () => {
|
||||
vi.useFakeTimers()
|
||||
try {
|
||||
vi.mocked(stat).mockResolvedValue({ isDirectory: () => true } as never)
|
||||
vi.mocked(subscribeViaWatcherProcess).mockImplementationOnce(() => new Promise(() => {}))
|
||||
const sender = {
|
||||
isDestroyed: () => false,
|
||||
send: vi.fn(),
|
||||
once: vi.fn(),
|
||||
id: 1
|
||||
}
|
||||
|
||||
const watchPromise = handlers['fs:watchWorktree'](
|
||||
{ sender },
|
||||
{ worktreePath: '/tmp/repo' }
|
||||
) as Promise<unknown>
|
||||
await vi.waitFor(() => {
|
||||
expect(subscribeViaWatcherProcess).toHaveBeenCalled()
|
||||
})
|
||||
|
||||
// Half the budget is already gone before the close starts; the drain may only spend the rest.
|
||||
const deadline = createWatcherRemovalDeadline()
|
||||
await vi.advanceTimersByTimeAsync(WATCHER_REMOVAL_DRAIN_BUDGET_MS / 2)
|
||||
|
||||
let closed = false
|
||||
const closePromise = closeLocalWatcherForWorktreePath('/tmp/repo', deadline).then(() => {
|
||||
closed = true
|
||||
})
|
||||
await vi.advanceTimersByTimeAsync(WATCHER_REMOVAL_DRAIN_BUDGET_MS / 2)
|
||||
await closePromise
|
||||
expect(closed).toBe(true)
|
||||
|
||||
void watchPromise
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
})
|
||||
@@ -860,7 +860,14 @@ export async function closeLocalWatcherForWorktreePath(
|
||||
clearTimeout(root.batch.timer)
|
||||
}
|
||||
watchedRoots.delete(rootKey)
|
||||
await trackLocalUnsubscribe(rootKey, root)
|
||||
// Why: the in-process Parcel fallback has no unsubscribe timeout of its own, so an unbounded await
|
||||
// here would hang delete forever and hold the removal gate. The promise stays tracked in
|
||||
// pendingLocalUnsubscribesByRoot, so a later close still observes its failure.
|
||||
await drainBeforeWatcherRemoval(
|
||||
trackLocalUnsubscribe(rootKey, root),
|
||||
deadline,
|
||||
`local watcher unsubscribe for ${rootKey}`
|
||||
)
|
||||
}
|
||||
|
||||
export async function restoreLocalWatcherAfterFailedRemoval(worktreePath: string): Promise<void> {
|
||||
|
||||
@@ -29,6 +29,47 @@ describe('watcher removal gate', () => {
|
||||
finishLaterInstall()
|
||||
})
|
||||
|
||||
it('drops abandoned install fences so a later removal is not fenced by a wedged install', async () => {
|
||||
beginWatcherInstall('/repo/nested')
|
||||
const removal = acquireWatcherRemovalGate('/repo')
|
||||
let ready = false
|
||||
void removal.ready.then(() => {
|
||||
ready = true
|
||||
})
|
||||
await Promise.resolve()
|
||||
expect(ready).toBe(false)
|
||||
|
||||
removal.abandonPendingInstalls()
|
||||
await removal.ready
|
||||
removal.release()
|
||||
|
||||
const retry = acquireWatcherRemovalGate('/repo')
|
||||
await retry.ready
|
||||
retry.release()
|
||||
})
|
||||
|
||||
it('keeps a fresh install fenced after an unrelated install was abandoned', async () => {
|
||||
const wedged = beginWatcherInstall('/repo')
|
||||
const removal = acquireWatcherRemovalGate('/repo')
|
||||
removal.abandonPendingInstalls()
|
||||
removal.release()
|
||||
// Why: a late finishInstall from the abandoned slot must not release a newer install's fence.
|
||||
const finishFresh = beginWatcherInstall('/repo')
|
||||
wedged()
|
||||
|
||||
const retry = acquireWatcherRemovalGate('/repo')
|
||||
let ready = false
|
||||
void retry.ready.then(() => {
|
||||
ready = true
|
||||
})
|
||||
await Promise.resolve()
|
||||
expect(ready).toBe(false)
|
||||
|
||||
finishFresh()
|
||||
await retry.ready
|
||||
retry.release()
|
||||
})
|
||||
|
||||
it('scopes identical roots to their execution host', async () => {
|
||||
const removal = acquireWatcherRemovalGate('/repo', 'ssh-a')
|
||||
await removal.ready
|
||||
|
||||
@@ -7,16 +7,25 @@ import {
|
||||
WATCHER_REMOVAL_IN_PROGRESS_MESSAGE
|
||||
} from '../../shared/worktree-removal-fence-error'
|
||||
|
||||
// Why: fence slots are identities, not a counter, so a removal that gives up waiting can drop exactly
|
||||
// the installs it waited on without a late finishInstall corrupting the count for a newer install.
|
||||
type WatcherInstallToken = symbol
|
||||
|
||||
type WatcherRemovalGateState = {
|
||||
key: string
|
||||
connectionId: string | null
|
||||
rootPath: string
|
||||
installCount: number
|
||||
installs: Set<WatcherInstallToken>
|
||||
removalCount: number
|
||||
installDrainWaiters: Set<() => void>
|
||||
}
|
||||
|
||||
export type WatcherRemovalGate = {
|
||||
ready: Promise<void>
|
||||
/** Drop the install fence slots this removal waited on. Call only after `ready` timed out: the
|
||||
* installs are presumed wedged, and leaving them counted makes every later removal of this root
|
||||
* wait out the full drain budget again. */
|
||||
abandonPendingInstalls(): void
|
||||
release(): void
|
||||
}
|
||||
|
||||
@@ -76,20 +85,11 @@ function beginRemovalSensitiveInstall(
|
||||
}
|
||||
const key = watcherRemovalGateKey(normalizedRoot, connectionId)
|
||||
const state = states.get(key) ?? createState(key, normalizedRoot, connectionId)
|
||||
state.installCount++
|
||||
let released = false
|
||||
const token: WatcherInstallToken = Symbol(normalizedRoot)
|
||||
state.installs.add(token)
|
||||
return () => {
|
||||
if (released) {
|
||||
return
|
||||
}
|
||||
released = true
|
||||
state.installCount--
|
||||
if (state.installCount === 0) {
|
||||
for (const resolve of state.installDrainWaiters) {
|
||||
resolve()
|
||||
}
|
||||
state.installDrainWaiters.clear()
|
||||
deleteIdleState(key, state)
|
||||
if (state.installs.delete(token) && state.installs.size === 0) {
|
||||
resolveInstallDrain(state)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -117,16 +117,30 @@ export function acquireWatcherRemovalGate(
|
||||
state.removalCount++
|
||||
// Why: deleting a parent root must wait for native installs already admitted
|
||||
// under that root, not only installs keyed to the exact same spelling.
|
||||
const drains = matchingHostStates(connectionId)
|
||||
const fenced = matchingHostStates(connectionId)
|
||||
.filter(
|
||||
(candidate) =>
|
||||
candidate.installCount > 0 && isPathInsideOrEqual(normalizedRoot, candidate.rootPath)
|
||||
candidate.installs.size > 0 && isPathInsideOrEqual(normalizedRoot, candidate.rootPath)
|
||||
)
|
||||
.map((candidate) => new Promise<void>((resolve) => candidate.installDrainWaiters.add(resolve)))
|
||||
.map((candidate) => ({ state: candidate, tokens: new Set(candidate.installs) }))
|
||||
const drains = fenced.map(
|
||||
({ state: candidate }) =>
|
||||
new Promise<void>((resolve) => candidate.installDrainWaiters.add(resolve))
|
||||
)
|
||||
const ready = drains.length === 0 ? Promise.resolve() : Promise.all(drains).then(() => undefined)
|
||||
let released = false
|
||||
return {
|
||||
ready,
|
||||
abandonPendingInstalls: () => {
|
||||
for (const { state: candidate, tokens } of fenced) {
|
||||
for (const token of tokens) {
|
||||
candidate.installs.delete(token)
|
||||
}
|
||||
if (candidate.installs.size === 0) {
|
||||
resolveInstallDrain(candidate)
|
||||
}
|
||||
}
|
||||
},
|
||||
release: () => {
|
||||
if (released) {
|
||||
return
|
||||
@@ -161,9 +175,10 @@ function createState(
|
||||
connectionId?: string
|
||||
): WatcherRemovalGateState {
|
||||
const state = {
|
||||
key,
|
||||
connectionId: connectionId ?? null,
|
||||
rootPath,
|
||||
installCount: 0,
|
||||
installs: new Set<WatcherInstallToken>(),
|
||||
removalCount: 0,
|
||||
installDrainWaiters: new Set<() => void>()
|
||||
}
|
||||
@@ -171,8 +186,16 @@ function createState(
|
||||
return state
|
||||
}
|
||||
|
||||
function resolveInstallDrain(state: WatcherRemovalGateState): void {
|
||||
for (const resolve of state.installDrainWaiters) {
|
||||
resolve()
|
||||
}
|
||||
state.installDrainWaiters.clear()
|
||||
deleteIdleState(state.key, state)
|
||||
}
|
||||
|
||||
function deleteIdleState(key: string, state: WatcherRemovalGateState): void {
|
||||
if (state.installCount === 0 && state.removalCount === 0 && states.get(key) === state) {
|
||||
if (state.installs.size === 0 && state.removalCount === 0 && states.get(key) === state) {
|
||||
states.delete(key)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -193,10 +193,8 @@ export class OpenCodeUsageStore {
|
||||
}
|
||||
|
||||
private writeToDisk(): Promise<void> {
|
||||
// Why: snapshot synchronously so a concurrent setEnabled/scan cannot tear the JSON mid-stringify.
|
||||
const payload = JSON.stringify(this.state, null, 2)
|
||||
const generation = ++this.writeGeneration
|
||||
const write = this.pendingWrite.then(() => this.commitToDisk(payload, generation))
|
||||
const write = this.pendingWrite.then(() => this.commitToDisk(generation))
|
||||
// Why: keep the queue usable after a failure — the awaiting caller still sees the rejection.
|
||||
this.pendingWrite = write.catch((error: unknown) => {
|
||||
console.error('[opencode-usage] Failed to persist usage cache:', error)
|
||||
@@ -204,11 +202,14 @@ export class OpenCodeUsageStore {
|
||||
return write
|
||||
}
|
||||
|
||||
private async commitToDisk(payload: string, generation: number): Promise<void> {
|
||||
private async commitToDisk(generation: number): Promise<void> {
|
||||
// Why: a newer snapshot is already queued behind this one, so rewriting the cache here is wasted.
|
||||
if (generation !== this.writeGeneration) {
|
||||
return
|
||||
}
|
||||
// Why stringify here and not at queue time: it blocks the main process for a multi-MB cache, and
|
||||
// superseded generations must not pay for it. Synchronous, so no mutation can tear the JSON.
|
||||
const payload = JSON.stringify(this.state, null, 2)
|
||||
const usageFile = getOpenCodeUsageFile()
|
||||
await mkdir(dirname(usageFile), { recursive: true }).catch(() => {})
|
||||
await writeFileDurableIfCurrent(
|
||||
|
||||
@@ -36591,8 +36591,9 @@ describe('OrcaRuntimeService', () => {
|
||||
|
||||
it('proceeds when a wedged install never releases the removal fence', async () => {
|
||||
vi.useFakeTimers()
|
||||
// Held across the whole acquire — models a native subscribe that ignores abort and never settles.
|
||||
const wedgedInstall = beginWatcherInstall(TEST_WORKTREE_PATH)
|
||||
// Held across the whole acquire and never released — models a native subscribe that ignores abort
|
||||
// and never settles. The removal must abandon the fence slot rather than leak it into later suites.
|
||||
beginWatcherInstall(TEST_WORKTREE_PATH)
|
||||
try {
|
||||
const runtime = createRuntime()
|
||||
|
||||
@@ -36614,8 +36615,30 @@ describe('OrcaRuntimeService', () => {
|
||||
const finishRetry = beginWatcherInstall(TEST_WORKTREE_PATH)
|
||||
finishRetry()
|
||||
} finally {
|
||||
// Why: the gate's install count is module-global; leaving it armed would stall later suites.
|
||||
wedgedInstall()
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('does not re-spend the drain budget on a removal after a wedged install was abandoned', async () => {
|
||||
vi.useFakeTimers()
|
||||
beginWatcherInstall(TEST_WORKTREE_PATH)
|
||||
try {
|
||||
const runtime = createRuntime()
|
||||
|
||||
const firstAcquiring = runtime.acquireFileWatcherRemoval(TEST_WORKTREE_PATH)
|
||||
await vi.advanceTimersByTimeAsync(WATCHER_REMOVAL_DRAIN_BUDGET_MS)
|
||||
await (await firstAcquiring).finish(true)
|
||||
|
||||
let secondAcquired = false
|
||||
const secondAcquiring = runtime.acquireFileWatcherRemoval(TEST_WORKTREE_PATH).then((gate) => {
|
||||
secondAcquired = true
|
||||
return gate
|
||||
})
|
||||
await vi.advanceTimersByTimeAsync(1)
|
||||
|
||||
expect(secondAcquired).toBe(true)
|
||||
await (await secondAcquiring).finish(true)
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
@@ -7225,11 +7225,16 @@ export class OrcaRuntimeService {
|
||||
await this.closeFileWatchersForRemoval(worktreePath, connectionId, deadline)
|
||||
// Why: a wedged install never releases its fence slot, so gate.ready can hang forever; the delete
|
||||
// must proceed instead, or the gate stays held and every later install under this root is rejected.
|
||||
await drainBeforeWatcherRemoval(
|
||||
const fenceDrain = await drainBeforeWatcherRemoval(
|
||||
gate.ready,
|
||||
deadline,
|
||||
`watcher install fence for ${worktreePath}`
|
||||
)
|
||||
if (fenceDrain === 'timeout') {
|
||||
// Why: a wedged install holds its fence slot for the process lifetime, so leaving it counted
|
||||
// makes every later removal of this root burn the whole drain budget again.
|
||||
gate.abandonPendingInstalls()
|
||||
}
|
||||
await this.closeFileWatchersForRemoval(worktreePath, connectionId, deadline)
|
||||
let finished = false
|
||||
return {
|
||||
|
||||
@@ -58,6 +58,17 @@ function scheduleHistoryTreeRemoval(dir: string): void {
|
||||
pendingHistoryTreeRemovals.set(dir, removal)
|
||||
}
|
||||
|
||||
/** Tombstone one history tree and queue its recursive removal off the caller's critical path.
|
||||
* Returns false when the rename failed, leaving the tree for a later GC pass to reclaim. */
|
||||
export function scheduleWorktreeHistoryTreeDeletion(dir: string, historyRoot: string): boolean {
|
||||
const tombstone = tombstoneHistoryTree(dir, historyRoot)
|
||||
if (!tombstone) {
|
||||
return false
|
||||
}
|
||||
scheduleHistoryTreeRemoval(tombstone)
|
||||
return true
|
||||
}
|
||||
|
||||
/** Schedule tombstoned trees under one history root for async removal — the retry after a quit mid-rm. */
|
||||
export function schedulePendingHistoryTreeRemovals(historyRoot: string): void {
|
||||
const pendingRoot = getPendingDeleteRoot(historyRoot)
|
||||
@@ -93,9 +104,7 @@ export function deleteWorktreeHistoryDir(worktreeId: string): void {
|
||||
const worktreeHash = hashWorktreeId(worktreeId)
|
||||
const historyRoot = getHistoryRoot()
|
||||
try {
|
||||
const tombstone = tombstoneHistoryTree(join(historyRoot, worktreeHash), historyRoot)
|
||||
if (tombstone) {
|
||||
scheduleHistoryTreeRemoval(tombstone)
|
||||
if (scheduleWorktreeHistoryTreeDeletion(join(historyRoot, worktreeHash), historyRoot)) {
|
||||
console.log(`[pty:history] Scheduled history delete for worktree ${worktreeId}`)
|
||||
}
|
||||
} catch (err) {
|
||||
@@ -106,9 +115,6 @@ export function deleteWorktreeHistoryDir(worktreeId: string): void {
|
||||
|
||||
// Also clean up WSL history for this worktree; listWslHistoryRoots is empty where WSL never ran.
|
||||
for (const distroRoot of listWslHistoryRoots()) {
|
||||
const tombstone = tombstoneHistoryTree(join(distroRoot, worktreeHash), distroRoot)
|
||||
if (tombstone) {
|
||||
scheduleHistoryTreeRemoval(tombstone)
|
||||
}
|
||||
scheduleWorktreeHistoryTreeDeletion(join(distroRoot, worktreeHash), distroRoot)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,11 +1,14 @@
|
||||
import { join } from 'node:path'
|
||||
import { existsSync, readFileSync, readdirSync, rmSync, statSync } from 'node:fs'
|
||||
import { existsSync, readFileSync, readdirSync, statSync } from 'node:fs'
|
||||
import {
|
||||
getHistoryRoot,
|
||||
listWslHistoryRoots,
|
||||
PENDING_DELETE_DIR_NAME
|
||||
} from './terminal-history-paths'
|
||||
import { schedulePendingHistoryTreeRemovals } from './terminal-history-deletion'
|
||||
import {
|
||||
schedulePendingHistoryTreeRemovals,
|
||||
scheduleWorktreeHistoryTreeDeletion
|
||||
} from './terminal-history-deletion'
|
||||
|
||||
// Why 5 minutes: GC runs ~10s after startup, and the live-worktree snapshot is
|
||||
// taken just before. A worktree created between the snapshot and GC execution
|
||||
@@ -78,9 +81,12 @@ function gcScanRoot(
|
||||
}
|
||||
|
||||
result.orphaned++
|
||||
rmSync(entryPath, { recursive: true, force: true })
|
||||
result.pruned++
|
||||
console.log(`[pty:history:gc] Pruned orphaned history: ${meta.worktreeId}`)
|
||||
// Why: a large orphaned tree recursive-rm'd here would stall the main process ~10s after
|
||||
// launch — the same freeze the explicit-delete path already tombstones its way out of.
|
||||
if (scheduleWorktreeHistoryTreeDeletion(entryPath, root)) {
|
||||
result.pruned++
|
||||
console.log(`[pty:history:gc] Pruned orphaned history: ${meta.worktreeId}`)
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
// Skip individual entries that fail.
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import type * as FsPromises from 'node:fs/promises'
|
||||
import { sep } from 'node:path'
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
const {
|
||||
@@ -61,11 +62,11 @@ vi.mock('./wsl', () => ({
|
||||
|
||||
import {
|
||||
resolveShellKind,
|
||||
hashWorktreeId,
|
||||
ensureHistoryDir,
|
||||
injectHistoryEnv,
|
||||
updateHistFileForFallback
|
||||
} from './terminal-history'
|
||||
import { hashWorktreeId } from './terminal-history-paths'
|
||||
import {
|
||||
deleteWorktreeHistoryDir,
|
||||
flushPendingWorktreeHistoryDeletions
|
||||
@@ -79,6 +80,8 @@ describe('terminal-history', () => {
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks()
|
||||
// Why: clearAllMocks keeps implementations, so a throwing rename from one test would leak forward.
|
||||
renameSyncMock.mockReset()
|
||||
getPathMock.mockReturnValue('/fake/userData')
|
||||
existsSyncMock.mockReturnValue(true)
|
||||
statSyncMock.mockReturnValue({ isDirectory: () => true, size: 100 })
|
||||
@@ -372,12 +375,17 @@ describe('terminal-history', () => {
|
||||
const liveIds = new Set(['live-wt'])
|
||||
runHistoryGc(liveIds)
|
||||
|
||||
// Should only prune dir2 (dead-wt), not dir1 (live-wt)
|
||||
expect(rmSyncMock).toHaveBeenCalledTimes(1)
|
||||
expect(rmSyncMock).toHaveBeenCalledWith(expect.stringContaining('dir2'), {
|
||||
recursive: true,
|
||||
force: true
|
||||
})
|
||||
// Should only prune dir2 (dead-wt), not dir1 (live-wt), and never recursive-rm on the main thread.
|
||||
expect(rmSyncMock).not.toHaveBeenCalled()
|
||||
expect(renameSyncMock).toHaveBeenCalledTimes(1)
|
||||
expect(renameSyncMock).toHaveBeenCalledWith(
|
||||
expect.stringContaining('dir2'),
|
||||
expect.stringContaining(`.pending-delete${sep}dir2.`)
|
||||
)
|
||||
expect(rmAsyncMock).toHaveBeenCalledWith(
|
||||
expect.stringContaining(`.pending-delete${sep}dir2.`),
|
||||
expect.objectContaining({ recursive: true, force: true })
|
||||
)
|
||||
})
|
||||
|
||||
it('skips recently-created directories to avoid TOCTOU race', () => {
|
||||
@@ -403,6 +411,7 @@ describe('terminal-history', () => {
|
||||
|
||||
// Should NOT prune because the directory is too young
|
||||
expect(rmSyncMock).not.toHaveBeenCalled()
|
||||
expect(renameSyncMock).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('does not throw when history root does not exist', () => {
|
||||
|
||||
@@ -3,8 +3,6 @@ import { mkdirSync, existsSync, writeFileSync } from 'node:fs'
|
||||
import { parseWslPath, toLinuxPath } from './wsl'
|
||||
import { getHistoryRoot, getHistoryRootWsl, hashWorktreeId } from './terminal-history-paths'
|
||||
|
||||
export { hashWorktreeId } from './terminal-history-paths'
|
||||
|
||||
type ShellKind = 'zsh' | 'bash' | 'fish' | 'pwsh' | 'powershell' | 'cmd' | 'unknown'
|
||||
|
||||
// ─── Shell Detection ───────────────────────────────────────────────
|
||||
|
||||
Reference in New Issue
Block a user