mirror of
https://github.com/stablyai/orca.git
synced 2026-10-23 08:03:39 +00:00
refactor(git-watch): share one single-flight helper between watcher fallbacks
Both fallbacks tracked their in-flight promise with the same self-comparison on settle, duplicated verbatim. Hoisting it removes a subtle invariant that was being hand-maintained in two places.
This commit is contained in:
@@ -0,0 +1,24 @@
|
||||
/** Collapses concurrent callers onto one run; the slot clears on settle so a
|
||||
* later caller starts fresh. Shared by both watcher fallbacks. */
|
||||
export function createSingleFlight(): {
|
||||
run: (start: () => Promise<void>) => Promise<void>
|
||||
pending: () => Promise<void> | null
|
||||
} {
|
||||
let inFlight: Promise<void> | null = null
|
||||
return {
|
||||
run: (start) => {
|
||||
if (inFlight) {
|
||||
return inFlight
|
||||
}
|
||||
// Self-comparison: a run settling after a newer one started must not clear it.
|
||||
const tracked: Promise<void> = start().finally(() => {
|
||||
if (inFlight === tracked) {
|
||||
inFlight = null
|
||||
}
|
||||
})
|
||||
inFlight = tracked
|
||||
return tracked
|
||||
},
|
||||
pending: () => inFlight
|
||||
}
|
||||
}
|
||||
@@ -8,6 +8,7 @@ import type {
|
||||
WorktreeBaseSubscription,
|
||||
WorktreePollerWindowVisibility
|
||||
} from './worktree-base-directory-poller'
|
||||
import { createSingleFlight } from './single-flight-promise'
|
||||
import { createGitCommonWatchReconciliation } from './worktree-git-common-watch-reconciliation'
|
||||
import { startGitCommonPolling } from './worktree-git-common-polling'
|
||||
|
||||
@@ -30,7 +31,7 @@ export async function startGitCommonNarrowWatch(
|
||||
let disposed = false
|
||||
let subscription: WorktreeBaseSubscription | null = null
|
||||
let existenceTimer: ReturnType<typeof setInterval> | null = null
|
||||
let pollingFallbackPromise: Promise<void> | null = null
|
||||
const pollingFallback = createSingleFlight()
|
||||
let subscribing = false
|
||||
let parkedWhileHidden = false
|
||||
let usingPollingFallback = false
|
||||
@@ -64,33 +65,31 @@ export async function startGitCommonNarrowWatch(
|
||||
isWatcherProcessFailure(error) &&
|
||||
(error.code === 'supervisor_crash_fuse' || error.code === 'process_unavailable')
|
||||
|
||||
const ensurePollingFallback = (): Promise<void> => {
|
||||
if (pollingFallbackPromise) {
|
||||
return pollingFallbackPromise
|
||||
}
|
||||
stopExistencePoll()
|
||||
usingPollingFallback = true
|
||||
const pending = reconciliation
|
||||
.unsubscribe()
|
||||
.catch(() => {})
|
||||
.then(() =>
|
||||
startGitCommonPolling(target.path, onEvents, pollIntervalMs, visibility, onFullScan, false)
|
||||
)
|
||||
.then(async (fallback) => {
|
||||
if (disposed || subscription) {
|
||||
await fallback.unsubscribe()
|
||||
return
|
||||
}
|
||||
subscription = fallback
|
||||
})
|
||||
const tracked = pending.finally(() => {
|
||||
if (pollingFallbackPromise === tracked) {
|
||||
pollingFallbackPromise = null
|
||||
}
|
||||
const ensurePollingFallback = (): Promise<void> =>
|
||||
pollingFallback.run(() => {
|
||||
stopExistencePoll()
|
||||
usingPollingFallback = true
|
||||
return reconciliation
|
||||
.unsubscribe()
|
||||
.catch(() => {})
|
||||
.then(() =>
|
||||
startGitCommonPolling(
|
||||
target.path,
|
||||
onEvents,
|
||||
pollIntervalMs,
|
||||
visibility,
|
||||
onFullScan,
|
||||
false
|
||||
)
|
||||
)
|
||||
.then(async (fallback) => {
|
||||
if (disposed || subscription) {
|
||||
await fallback.unsubscribe()
|
||||
return
|
||||
}
|
||||
subscription = fallback
|
||||
})
|
||||
})
|
||||
pollingFallbackPromise = tracked
|
||||
return pollingFallbackPromise
|
||||
}
|
||||
|
||||
const tryUpgradeToNarrowWatch = async (): Promise<void> => {
|
||||
if (disposed || subscribing || subscription) {
|
||||
@@ -232,7 +231,7 @@ export async function startGitCommonNarrowWatch(
|
||||
}
|
||||
if (disposed || errored) {
|
||||
void sub.unsubscribe().catch(() => {})
|
||||
await pollingFallbackPromise?.catch(() => {})
|
||||
await pollingFallback.pending()?.catch(() => {})
|
||||
return !errored || subscription !== null
|
||||
}
|
||||
subscription = { unsubscribe: () => sub.unsubscribe() }
|
||||
@@ -261,7 +260,7 @@ export async function startGitCommonNarrowWatch(
|
||||
disposed = true
|
||||
stopExistencePoll()
|
||||
unsubscribeVisibility()
|
||||
await pollingFallbackPromise?.catch(() => {})
|
||||
await pollingFallback.pending()?.catch(() => {})
|
||||
nativeSubscriptionGeneration++
|
||||
const current = subscription
|
||||
subscription = null
|
||||
|
||||
@@ -6,6 +6,7 @@ import type {
|
||||
WorktreeBaseSubscription,
|
||||
WorktreePollerWindowVisibility
|
||||
} from './worktree-base-directory-poller'
|
||||
import { createSingleFlight } from './single-flight-promise'
|
||||
import { PRIMARY_CHECKOUT_METADATA_FILES } from './worktree-git-common-metadata-files'
|
||||
import { startGitCommonPrimaryPolling } from './worktree-git-common-primary-polling'
|
||||
|
||||
@@ -34,34 +35,25 @@ export async function startGitCommonPrimaryWatch(
|
||||
let watcher: WatcherProcessSubscription | null = null
|
||||
let statusRefPolling: WorktreeBaseSubscription | null = null
|
||||
let fallback: WorktreeBaseSubscription | null = null
|
||||
let fallbackPromise: Promise<void> | null = null
|
||||
const fallbackFlight = createSingleFlight()
|
||||
|
||||
const startFallback = (): Promise<void> => {
|
||||
if (fallbackPromise) {
|
||||
return fallbackPromise
|
||||
}
|
||||
const pending = startGitCommonPrimaryPolling(
|
||||
commonDirPath,
|
||||
getStatusRefPaths,
|
||||
onEvents,
|
||||
pollIntervalMs,
|
||||
visibility,
|
||||
onFullScan
|
||||
).then(async (nextFallback) => {
|
||||
if (disposed || watcher || statusRefPolling) {
|
||||
await nextFallback.unsubscribe()
|
||||
return
|
||||
}
|
||||
fallback = nextFallback
|
||||
})
|
||||
const tracked = pending.finally(() => {
|
||||
if (fallbackPromise === tracked) {
|
||||
fallbackPromise = null
|
||||
}
|
||||
})
|
||||
fallbackPromise = tracked
|
||||
return fallbackPromise
|
||||
}
|
||||
const startFallback = (): Promise<void> =>
|
||||
fallbackFlight.run(() =>
|
||||
startGitCommonPrimaryPolling(
|
||||
commonDirPath,
|
||||
getStatusRefPaths,
|
||||
onEvents,
|
||||
pollIntervalMs,
|
||||
visibility,
|
||||
onFullScan
|
||||
).then(async (nextFallback) => {
|
||||
if (disposed || watcher || statusRefPolling) {
|
||||
await nextFallback.unsubscribe()
|
||||
return
|
||||
}
|
||||
fallback = nextFallback
|
||||
})
|
||||
)
|
||||
|
||||
const stopStatusRefPolling = async (): Promise<void> => {
|
||||
const current = statusRefPolling
|
||||
@@ -115,7 +107,7 @@ export async function startGitCommonPrimaryWatch(
|
||||
}
|
||||
}
|
||||
)
|
||||
if (watcher && !disposed && !fallbackPromise) {
|
||||
if (watcher && !disposed && !fallbackFlight.pending()) {
|
||||
statusRefPolling = await startGitCommonPrimaryPolling(
|
||||
commonDirPath,
|
||||
getStatusRefPaths,
|
||||
@@ -139,7 +131,7 @@ export async function startGitCommonPrimaryWatch(
|
||||
await current.unsubscribe().catch(() => {})
|
||||
}
|
||||
await stopStatusRefPolling()
|
||||
await fallbackPromise?.catch(() => {})
|
||||
await fallbackFlight.pending()?.catch(() => {})
|
||||
if (fallback) {
|
||||
await fallback.unsubscribe().catch(() => {})
|
||||
fallback = null
|
||||
|
||||
Reference in New Issue
Block a user