diff --git a/packages/server/scripts/supervisor.logging.test.ts b/packages/server/scripts/supervisor.logging.test.ts index 3d1d926f6..61255358f 100644 --- a/packages/server/scripts/supervisor.logging.test.ts +++ b/packages/server/scripts/supervisor.logging.test.ts @@ -13,9 +13,11 @@ const supervisorPath = fileURLToPath(new URL("./supervisor.ts", import.meta.url) async function runSupervisorFixture(options: { workerSource: string; restartOnCrash?: boolean; + timeoutMs?: number; }): Promise<{ code: number | null; signal: NodeJS.Signals | null; + elapsedMs: number; log: string; stdout: string; stderr: string; @@ -47,6 +49,7 @@ async function runSupervisorFixture(options: { `, ); + const startedAt = Date.now(); const child = spawn(process.execPath, ["--import", "tsx", runnerPath], { cwd: repoRoot, env: { ...process.env }, @@ -71,7 +74,7 @@ async function runSupervisorFixture(options: { const timeout = setTimeout(() => { child.kill("SIGKILL"); reject(new Error("supervisor fixture timed out")); - }, 10000); + }, options.timeoutMs ?? 10_000); child.on("error", (error) => { clearTimeout(timeout); @@ -84,7 +87,7 @@ async function runSupervisorFixture(options: { }); const log = await readFile(logPath, "utf8"); - return { code, signal, log, stdout, stderr }; + return { code, signal, elapsedMs: Date.now() - startedAt, log, stdout, stderr }; } describe("supervisor durable logging", () => { @@ -186,6 +189,152 @@ describe("supervisor durable logging", () => { expect(result.log).toContain('"workerPid":'); }); + test("keeps a worker alive while heartbeats continue", async () => { + const result = await runSupervisorFixture({ + timeoutMs: 10_000, + workerSource: ` + setInterval(() => { + process.send?.({ type: "paseo:worker-heartbeat" }); + }, 250); + setTimeout(() => { + process.send?.({ type: "paseo:shutdown", reason: "healthy_heartbeat_test_complete" }); + }, 6000); + `, + }); + + expect(result.code).toBe(0); + expect(result.signal).toBeNull(); + expect(result.log).toContain('"reason":"healthy_heartbeat_test_complete"'); + expect(result.log).not.toContain('"msg":"Worker heartbeat timed out; restarting worker"'); + }, 10_000); + + test("tolerates a transient seven-second heartbeat pause", async () => { + const result = await runSupervisorFixture({ + timeoutMs: 10_000, + workerSource: ` + import { existsSync, writeFileSync } from "node:fs"; + + const marker = process.argv[1] + ".started"; + if (!existsSync(marker)) { + writeFileSync(marker, "started"); + let heartbeatCount = 0; + const heartbeat = setInterval(() => { + process.send?.({ type: "paseo:worker-heartbeat" }); + heartbeatCount += 1; + if (heartbeatCount === 3) clearInterval(heartbeat); + }, 100); + setTimeout(() => { + process.send?.({ type: "paseo:shutdown", reason: "heartbeat_pause_tolerated" }); + }, 7_000); + setInterval(() => {}, 1_000); + } else { + process.send?.({ type: "paseo:shutdown", reason: "unexpected_heartbeat_restart" }); + setInterval(() => {}, 1_000); + } + `, + }); + + expect(result.code).toBe(0); + expect(result.signal).toBeNull(); + expect(result.log).toContain('"reason":"heartbeat_pause_tolerated"'); + expect(result.log).not.toContain('"reason":"unexpected_heartbeat_restart"'); + expect(result.log).not.toContain('"msg":"Worker heartbeat timed out; restarting worker"'); + }, 10_000); + + test.skipIf(isPlatform("win32"))( + "forces shutdown when a worker ignores SIGTERM", + async () => { + const result = await runSupervisorFixture({ + timeoutMs: 15_000, + workerSource: ` + process.on("SIGTERM", () => {}); + process.send?.({ type: "paseo:shutdown", reason: "stalled_worker_shutdown" }); + setInterval(() => {}, 1_000); + `, + }); + + expect(result.code).toBe(0); + expect(result.signal).toBeNull(); + expect(result.log).toContain('"reason":"stalled_worker_shutdown"'); + expect(result.log).toContain('"msg":"Worker did not exit after SIGTERM; forcing SIGKILL"'); + expect(result.log).toContain('"signal":"SIGKILL"'); + }, + 20_000, + ); + + // POSIX-only: the watchdog uses SIGKILL after its graceful shutdown window. + test.skipIf(isPlatform("win32"))( + "restarts a worker that stops heartbeating", + async () => { + const result = await runSupervisorFixture({ + timeoutMs: 35_000, + workerSource: ` + import { existsSync, writeFileSync } from "node:fs"; + + const marker = process.argv[1] + ".started"; + if (!existsSync(marker)) { + writeFileSync(marker, "started"); + let heartbeatCount = 0; + const heartbeat = setInterval(() => { + process.send?.({ type: "paseo:worker-heartbeat" }); + heartbeatCount += 1; + if (heartbeatCount === 3) clearInterval(heartbeat); + }, 100); + process.on("SIGTERM", () => {}); + setInterval(() => {}, 1000); + } else { + process.send?.({ type: "paseo:shutdown", reason: "watchdog_test_complete" }); + setInterval(() => {}, 1000); + } + `, + }); + + expect(result.code).toBe(0); + expect(result.signal).toBeNull(); + expect(result.log).toContain('"msg":"Worker heartbeat timed out; restarting worker"'); + expect(result.log).toContain('"msg":"Worker did not exit after SIGTERM; forcing SIGKILL"'); + expect(result.log).toContain('"signal":"SIGKILL"'); + }, + 40_000, + ); + + test.skipIf(isPlatform("win32"))( + "restarts after worker exit while a descendant retains the worker stdio", + async () => { + const result = await runSupervisorFixture({ + timeoutMs: 7_000, + workerSource: ` + import { spawn } from "node:child_process"; + import { existsSync, writeFileSync } from "node:fs"; + + const marker = process.argv[1] + ".started"; + if (!existsSync(marker)) { + writeFileSync(marker, "started"); + const descendant = spawn( + process.execPath, + ["-e", "process.on('SIGTERM', () => {}); setTimeout(() => process.exit(0), 4000)"], + { detached: true, stdio: ["ignore", "inherit", "inherit"] }, + ); + descendant.unref(); + process.on("SIGTERM", () => process.exit(0)); + process.send?.({ type: "paseo:restart", reason: "stdio_descendant" }); + setInterval(() => {}, 1000); + } else { + process.send?.({ type: "paseo:shutdown", reason: "stdio_restart_complete" }); + setInterval(() => {}, 1000); + } + `, + }); + + expect(result.code).toBe(0); + expect(result.signal).toBeNull(); + expect(result.elapsedMs).toBeLessThan(2_500); + expect(result.log).toContain('"reason":"stdio_descendant"'); + expect(result.log).toContain("Restarting worker"); + }, + 7_000, + ); + // POSIX-only: Windows reports the worker self-kill as an exit code, not SIGKILL. test.skipIf(isPlatform("win32"))( "logs worker signal exits even when the worker cannot log", diff --git a/packages/server/scripts/supervisor.ts b/packages/server/scripts/supervisor.ts index 4e56c9612..4554e8d54 100644 --- a/packages/server/scripts/supervisor.ts +++ b/packages/server/scripts/supervisor.ts @@ -2,6 +2,11 @@ import { fork, spawn, type ChildProcess } from "child_process"; import { mkdirSync } from "node:fs"; import path from "node:path"; import { createStream as createRotatingFileStream } from "rotating-file-stream"; +import { signalProcessTree } from "../src/utils/tree-kill.js"; + +const WORKER_HEARTBEAT_INTERVAL_MS = 1_000; +const WORKER_HEARTBEAT_TIMEOUT_MS = 15_000; +const WORKER_TERMINATION_GRACE_MS = 10_000; interface SupervisorLogFileOptions { path: string; @@ -29,6 +34,10 @@ interface SupervisorHeartbeatMessage { type: "paseo:supervisor-heartbeat"; } +interface WorkerHeartbeatMessage { + type: "paseo:worker-heartbeat"; +} + interface SupervisorOptions { name: string; startupMessage: string; @@ -84,6 +93,15 @@ function parseLifecycleMessage(msg: unknown): WorkerLifecycleMessage | null { return null; } +function isWorkerHeartbeatMessage(msg: unknown): msg is WorkerHeartbeatMessage { + return ( + typeof msg === "object" && + msg !== null && + "type" in msg && + (msg as { type?: unknown }).type === "paseo:worker-heartbeat" + ); +} + function toRotatingFileStreamSize(size: string): string { const trimmed = size.trim(); const match = trimmed.match(/^(\d+)\s*([bBkKmMgG])?$/); @@ -120,6 +138,7 @@ export function runSupervisor(options: SupervisorOptions): SupervisorController let restarting = false; let shuttingDown = false; let exiting = false; + let forceKillTimer: NodeJS.Timeout | null = null; const logStream = createSupervisorLogStream(options.logFile); const writeDurableChunk = (chunk: string | Buffer): void => { @@ -169,6 +188,40 @@ export function runSupervisor(options: SupervisorOptions): SupervisorController }); }; + const clearForceKillTimer = (): void => { + if (forceKillTimer) { + clearTimeout(forceKillTimer); + forceKillTimer = null; + } + }; + + const scheduleForceKill = (reason: string): void => { + if (!child) { + return; + } + const currentChild = child; + clearForceKillTimer(); + forceKillTimer = setTimeout(() => { + forceKillTimer = null; + if (child !== currentChild) { + return; + } + writeLifecycleLog("Worker did not exit after SIGTERM; forcing SIGKILL", { + reason, + supervisorPid: process.pid, + workerPid: currentChild.pid ?? null, + }); + void signalProcessTree(currentChild, "SIGKILL").catch((error) => { + writeLifecycleLog("Failed to force-kill worker process tree", { + error: error instanceof Error ? error.message : String(error), + supervisorPid: process.pid, + workerPid: currentChild.pid ?? null, + }); + }); + }, WORKER_TERMINATION_GRACE_MS); + forceKillTimer.unref(); + }; + const spawnWorker = () => { let workerEntry: string; try { @@ -197,6 +250,7 @@ export function runSupervisor(options: SupervisorOptions): SupervisorController } const currentChild = child; + let lastWorkerHeartbeatAt = Date.now(); const heartbeat = setInterval(() => { const message: SupervisorHeartbeatMessage = { type: "paseo:supervisor-heartbeat" }; if (currentChild.connected) { @@ -210,9 +264,26 @@ export function runSupervisor(options: SupervisorOptions): SupervisorController } else { writeLifecycleLog("Worker heartbeat skipped because IPC channel is disconnected"); } - }, 1000); + }, WORKER_HEARTBEAT_INTERVAL_MS); heartbeat.unref(); + const workerWatchdog = setInterval(() => { + if (child !== currentChild || restarting || shuttingDown) { + return; + } + const heartbeatAgeMs = Date.now() - lastWorkerHeartbeatAt; + if (heartbeatAgeMs < WORKER_HEARTBEAT_TIMEOUT_MS) { + return; + } + writeLifecycleLog("Worker heartbeat timed out; restarting worker", { + heartbeatAgeMs, + supervisorPid: process.pid, + workerPid: currentChild.pid ?? null, + }); + requestRestart("worker_heartbeat_timeout"); + }, WORKER_HEARTBEAT_INTERVAL_MS); + workerWatchdog.unref(); + child.on("disconnect", () => { writeLifecycleLog("Worker IPC channel disconnected"); }); @@ -228,6 +299,10 @@ export function runSupervisor(options: SupervisorOptions): SupervisorController }); child.on("message", (msg: unknown) => { + if (isWorkerHeartbeatMessage(msg)) { + lastWorkerHeartbeatAt = Date.now(); + return; + } const lifecycleMessage = parseLifecycleMessage(msg); if (!lifecycleMessage) { return; @@ -256,8 +331,10 @@ export function runSupervisor(options: SupervisorOptions): SupervisorController requestRestart(reason); }); - child.on("close", (code, signal) => { + child.on("exit", (code, signal) => { clearInterval(heartbeat); + clearInterval(workerWatchdog); + clearForceKillTimer(); const exitDescriptor = describeExit(code, signal); writeLifecycleLog("Worker exited", { code, signal, exit: exitDescriptor }); @@ -308,6 +385,7 @@ export function runSupervisor(options: SupervisorOptions): SupervisorController writeLifecycleLog("Restart requested", { reason }); log(`${reason}. Stopping worker for restart...`); signalWorker("SIGTERM", reason); + scheduleForceKill(reason); }; const requestShutdown = (reason: string) => { @@ -323,6 +401,7 @@ export function runSupervisor(options: SupervisorOptions): SupervisorController return; } signalWorker("SIGTERM", reason); + scheduleForceKill(reason); }; const forwardSignal = (signal: NodeJS.Signals) => { diff --git a/packages/server/src/server/daemon-worker.ts b/packages/server/src/server/daemon-worker.ts index 4656523ef..4188e97d2 100644 --- a/packages/server/src/server/daemon-worker.ts +++ b/packages/server/src/server/daemon-worker.ts @@ -27,6 +27,10 @@ interface SupervisorHeartbeatMessage { type: "paseo:supervisor-heartbeat"; } +interface WorkerHeartbeatMessage { + type: "paseo:worker-heartbeat"; +} + interface BootstrapResult { paseoHome: string; logger: ReturnType; @@ -228,6 +232,13 @@ async function main() { const supervisorPid = process.ppid; let lastSupervisorHeartbeatAt = Date.now(); let supervisorExitRequested = false; + const sendWorkerHeartbeat = () => { + try { + process.send?.({ type: "paseo:worker-heartbeat" } satisfies WorkerHeartbeatMessage); + } catch { + // The disconnect handler below owns supervisor-loss shutdown. + } + }; const exitAfterSupervisorLoss = (reason: string) => { if (supervisorExitRequested) { return; @@ -261,6 +272,10 @@ async function main() { }); process.on("disconnect", () => exitAfterSupervisorLoss("ipc_disconnect_event")); + sendWorkerHeartbeat(); + const workerHeartbeat = setInterval(sendWorkerHeartbeat, 1_000); + workerHeartbeat.unref(); + const timer = setInterval(() => { const ipcConnected = typeof process.connected === "boolean" ? process.connected : true; const heartbeatExpired = Date.now() - lastSupervisorHeartbeatAt > 3500; diff --git a/packages/server/src/server/test-utils/workspace-git-service-stub.ts b/packages/server/src/server/test-utils/workspace-git-service-stub.ts index 8c82bad82..2cfaf659f 100644 --- a/packages/server/src/server/test-utils/workspace-git-service-stub.ts +++ b/packages/server/src/server/test-utils/workspace-git-service-stub.ts @@ -82,6 +82,10 @@ export function createNoopWorkspaceGitService( workingTreeWatchSetupInFlightCount: 0, workspaceRefreshInFlightCount: 0, workspaceRefreshQueuedCount: 0, + workspaceRefreshAdmissionActiveCount: 0, + workspaceRefreshAdmissionPendingCount: 0, + workspaceObservationSetupAdmissionActiveCount: 0, + workspaceObservationSetupAdmissionPendingCount: 0, fetchInFlightCount: 0, snapshotUpdatedListenerCount: 0, watcherErrorCallbackCount: 0, diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index 0b95c9a4b..7797e466c 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -246,6 +246,10 @@ function createFallbackWorkspaceGitService(): WorkspaceGitService { workingTreeWatchSetupInFlightCount: 0, workspaceRefreshInFlightCount: 0, workspaceRefreshQueuedCount: 0, + workspaceRefreshAdmissionActiveCount: 0, + workspaceRefreshAdmissionPendingCount: 0, + workspaceObservationSetupAdmissionActiveCount: 0, + workspaceObservationSetupAdmissionPendingCount: 0, fetchInFlightCount: 0, snapshotUpdatedListenerCount: 0, watcherErrorCallbackCount: 0, diff --git a/packages/server/src/server/workspace-git-service.observation.test.ts b/packages/server/src/server/workspace-git-service.observation.test.ts index 99b6802eb..6ca28b81d 100644 --- a/packages/server/src/server/workspace-git-service.observation.test.ts +++ b/packages/server/src/server/workspace-git-service.observation.test.ts @@ -1093,7 +1093,8 @@ describe("WorkspaceGitService checkout observation", () => { expect(watcher.subscribe).toHaveBeenCalledTimes(1); }); await vi.advanceTimersByTimeAsync(35_000); - expect(service.getMetrics().workingTreeWatchSetupInFlightCount).toBe(1); + expect(service.getMetrics().workingTreeWatchSetupInFlightCount).toBe(0); + expect(service.getMetrics().workspaceObservationSetupAdmissionActiveCount).toBe(0); expect(getWatcherSubscribeCallCount(watcher, REPO_CWD)).toBe(1); openedSubscription.resolve({ unsubscribe: erroredUnsubscribe }); diff --git a/packages/server/src/server/workspace-git-service.test.ts b/packages/server/src/server/workspace-git-service.test.ts index a97d02e84..f7f51ad56 100644 --- a/packages/server/src/server/workspace-git-service.test.ts +++ b/packages/server/src/server/workspace-git-service.test.ts @@ -9,6 +9,9 @@ import type { PullRequestStatusResult, } from "../utils/checkout-git.js"; import { + WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY, + WORKSPACE_GIT_REFRESH_CONCURRENCY, + WORKSPACE_GIT_WATCHER_SUBSCRIBE_TIMEOUT_MS, WorkspaceGitServiceImpl, type WorkspaceGitRuntimeSnapshot, } from "./workspace-git-service.js"; @@ -507,6 +510,279 @@ describe("WorkspaceGitServiceImpl", () => { service.dispose(); }); + test("bounds refresh generations and re-enqueues hot workspace successors fairly", async () => { + const hotCwds = Array.from({ length: WORKSPACE_GIT_REFRESH_CONCURRENCY }, (_, index) => + path.resolve(`/tmp/hot-repo-${index}`), + ); + const fifthCwd = path.resolve("/tmp/fifth-repo"); + const firstReadGates = new Map( + [...hotCwds, fifthCwd].map((cwd) => [cwd, createDeferred()]), + ); + const pendingFirstReads = new Set(firstReadGates.keys()); + const startedReads: string[] = []; + const getCheckoutStatus = vi.fn((cwd: string) => { + startedReads.push(cwd); + const firstRead = firstReadGates.get(cwd); + if (firstRead && pendingFirstReads.delete(cwd)) { + return firstRead.promise; + } + return Promise.resolve(createCheckoutStatus(cwd, { currentBranch: "successor" })); + }); + const service = createService({ getCheckoutStatus }); + + const hotInitialReads = hotCwds.map((cwd) => service.getSnapshot(cwd, { includeForge: false })); + await vi.waitFor(() => { + expect(startedReads).toEqual(hotCwds); + }); + + const fifthRead = service.getSnapshot(fifthCwd, { includeForge: false }); + const hotSuccessorReads = hotCwds.map((cwd) => + service.getSnapshot(cwd, { + force: true, + includeForge: false, + reason: "hot-successor", + }), + ); + expect(service.getMetrics()).toMatchObject({ + workspaceRefreshAdmissionActiveCount: WORKSPACE_GIT_REFRESH_CONCURRENCY, + workspaceRefreshAdmissionPendingCount: 1, + workspaceRefreshQueuedCount: WORKSPACE_GIT_REFRESH_CONCURRENCY, + }); + + firstReadGates.get(hotCwds[0])?.resolve(createCheckoutStatus(hotCwds[0])); + await vi.waitFor(() => { + expect(startedReads).toContain(fifthCwd); + }); + expect(startedReads.filter((cwd) => cwd === hotCwds[0])).toHaveLength(1); + + for (const [cwd, gate] of firstReadGates) { + gate.resolve(createCheckoutStatus(cwd)); + } + await Promise.all([...hotInitialReads, fifthRead, ...hotSuccessorReads]); + + for (const cwd of hotCwds) { + expect(startedReads.filter((startedCwd) => startedCwd === cwd)).toHaveLength(2); + } + expect(startedReads.filter((cwd) => cwd === fifthCwd)).toHaveLength(1); + + service.dispose(); + }); + + test("bounds workspace observation setup pipelines", async () => { + const cwds = Array.from( + { length: WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY + 2 }, + (_, index) => `/tmp/observation-repo-${index}`, + ); + const setupGate = createDeferred(); + let activeSetups = 0; + let maxActiveSetups = 0; + const subscribe = vi.fn(async () => { + activeSetups += 1; + maxActiveSetups = Math.max(maxActiveSetups, activeSetups); + await setupGate.promise; + activeSetups -= 1; + return createAsyncSubscription(); + }); + const runGitCommand = vi.fn(async (_args: string[], options: { cwd: string }) => ({ + stdout: `${options.cwd}\n`, + stderr: "", + truncated: false, + exitCode: 0, + signal: null, + })); + const service = createService({ subscribe, runGitCommand }); + + await Promise.all(cwds.map((cwd) => service.getSnapshot(cwd, { includeForge: false }))); + const subscriptions = cwds.map((cwd) => service.registerWorkspace({ cwd }, vi.fn())); + + await vi.waitFor(() => { + expect(subscribe).toHaveBeenCalledTimes(WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY); + }); + expect(maxActiveSetups).toBe(WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY); + expect(service.getMetrics()).toMatchObject({ + workspaceObservationSetupAdmissionActiveCount: WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY, + workspaceObservationSetupAdmissionPendingCount: 2, + }); + + setupGate.resolve(); + await vi.waitFor(() => { + expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0); + }); + expect(maxActiveSetups).toBe(WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY); + + for (const subscription of subscriptions) { + subscription.unsubscribe(); + } + service.dispose(); + }); + + test("watcher subscription deadlines release observation admission slots", async () => { + const cwds = Array.from( + { length: WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY + 2 }, + (_, index) => path.resolve(`/tmp/deadline-repo-${index}`), + ); + const stalledSubscriptions = Array.from( + { length: WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY }, + () => createDeferred>(), + ); + const subscribedPaths: string[] = []; + const subscribe = vi.fn((watchPath: string) => { + subscribedPaths.push(watchPath); + const stalled = stalledSubscriptions.shift(); + return stalled?.promise ?? Promise.resolve(createAsyncSubscription()); + }); + const runGitCommand = vi.fn(async (_args: string[], options: { cwd: string }) => ({ + stdout: `${options.cwd}\n`, + stderr: "", + truncated: false, + exitCode: 0, + signal: null, + })); + const service = createService({ subscribe, runGitCommand }); + + await Promise.all(cwds.map((cwd) => service.getSnapshot(cwd, { includeForge: false }))); + const subscriptions = cwds.map((cwd) => service.registerWorkspace({ cwd }, vi.fn())); + + await vi.waitFor(() => { + expect(subscribe).toHaveBeenCalledTimes(WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY); + }); + + await vi.advanceTimersByTimeAsync(WORKSPACE_GIT_WATCHER_SUBSCRIBE_TIMEOUT_MS); + await vi.waitFor(() => { + expect(subscribedPaths).toEqual(expect.arrayContaining(cwds)); + expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0); + }); + + for (const subscription of subscriptions) { + subscription.unsubscribe(); + } + service.dispose(); + }); + + test("dispose settles queued refresh and observation work and cleans late watchers", async () => { + const cwds = Array.from({ length: WORKSPACE_GIT_REFRESH_CONCURRENCY + 1 }, (_, index) => + path.resolve(`/tmp/dispose-repo-${index}`), + ); + const refreshGates = new Map(cwds.map((cwd) => [cwd, createDeferred()])); + const getCheckoutStatus = vi.fn((cwd: string) => { + const gate = refreshGates.get(cwd); + if (!gate) { + throw new Error(`Unexpected cwd: ${cwd}`); + } + return gate.promise; + }); + const watcherGates = Array.from({ length: WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY }, () => + createDeferred>(), + ); + const lateSubscriptions = watcherGates.map(() => createAsyncSubscription()); + const lateUnsubscribes = lateSubscriptions.map(({ unsubscribe }) => unsubscribe); + const subscribe = vi.fn( + () => watcherGates[subscribe.mock.calls.length - 1]?.promise ?? Promise.reject(), + ); + const runGitCommand = vi.fn(async (_args: string[], options: { cwd: string }) => ({ + stdout: `${options.cwd}\n`, + stderr: "", + truncated: false, + exitCode: 0, + signal: null, + })); + const service = createService({ getCheckoutStatus, subscribe, runGitCommand }); + + cwds.forEach((cwd) => service.registerWorkspace({ cwd }, vi.fn())); + await vi.waitFor(() => { + expect(service.getMetrics()).toMatchObject({ + workspaceRefreshAdmissionActiveCount: WORKSPACE_GIT_REFRESH_CONCURRENCY, + workspaceRefreshAdmissionPendingCount: 1, + workspaceObservationSetupAdmissionActiveCount: WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY, + workspaceObservationSetupAdmissionPendingCount: + cwds.length - WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY, + }); + }); + const queuedRefreshSettlement = service.getSnapshot(cwds.at(-1)!).then( + () => ({ status: "fulfilled" as const, error: null }), + (error: unknown) => ({ status: "rejected" as const, error }), + ); + + service.dispose(); + + await expect(queuedRefreshSettlement).resolves.toMatchObject({ + status: "rejected", + error: { name: "AbortError" }, + }); + await vi.waitFor(() => { + expect(service.getMetrics()).toMatchObject({ + workspaceRefreshAdmissionPendingCount: 0, + workspaceObservationSetupAdmissionActiveCount: 0, + workspaceObservationSetupAdmissionPendingCount: 0, + }); + }); + expect(subscribe).toHaveBeenCalledTimes(WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY); + + watcherGates.forEach((gate, index) => gate.resolve(lateSubscriptions[index]!)); + refreshGates.forEach((gate, cwd) => gate.resolve(createCheckoutStatus(cwd))); + await vi.waitFor(() => { + for (const unsubscribe of lateUnsubscribes) { + expect(unsubscribe).toHaveBeenCalledTimes(1); + } + expect(service.getMetrics().workspaceRefreshAdmissionActiveCount).toBe(0); + }); + }); + + test("cleans a watcher subscription that resolves after its deadline", async () => { + const lateWatcher = createDeferred>(); + const lateSubscription = createAsyncSubscription(); + const subscribe = vi + .fn() + .mockImplementationOnce(() => lateWatcher.promise) + .mockResolvedValue(createAsyncSubscription()); + const service = createService({ subscribe }); + + await service.getSnapshot(REPO_CWD, { includeForge: false }); + const workspaceSubscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn()); + await vi.waitFor(() => { + expect(subscribe).toHaveBeenCalledTimes(1); + }); + + await vi.advanceTimersByTimeAsync(WORKSPACE_GIT_WATCHER_SUBSCRIBE_TIMEOUT_MS); + await vi.waitFor(() => { + expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0); + }); + lateWatcher.resolve(lateSubscription); + + await vi.waitFor(() => { + expect(lateSubscription.unsubscribe).toHaveBeenCalledTimes(1); + }); + workspaceSubscription.unsubscribe(); + service.dispose(); + expect(lateSubscription.unsubscribe).toHaveBeenCalledTimes(1); + }); + + test("rejects new work after disposal", async () => { + const getCheckoutSnapshotFacts = vi.fn(async (cwd: string) => createCheckoutSnapshotFacts(cwd)); + const runGitCommand = vi.fn(); + const service = createService({ getCheckoutSnapshotFacts, runGitCommand }); + + service.dispose(); + + expect(() => service.registerWorkspace({ cwd: REPO_CWD }, vi.fn())).toThrow( + "WorkspaceGitService is disposed", + ); + await expect(service.getSnapshot(REPO_CWD)).rejects.toThrow("WorkspaceGitService is disposed"); + await expect(service.refresh(REPO_CWD)).rejects.toThrow("WorkspaceGitService is disposed"); + await expect(service.requestWorkingTreeWatch(REPO_CWD, vi.fn())).rejects.toThrow( + "WorkspaceGitService is disposed", + ); + expect(() => service.scheduleRefreshForCwd(REPO_CWD)).toThrow( + "WorkspaceGitService is disposed", + ); + expect(() => service.onWorkspaceStateMayHaveChanged(REPO_CWD)).toThrow( + "WorkspaceGitService is disposed", + ); + expect(() => service.invalidateForge(REPO_CWD)).toThrow("WorkspaceGitService is disposed"); + expect(getCheckoutSnapshotFacts).not.toHaveBeenCalled(); + expect(runGitCommand).not.toHaveBeenCalled(); + }); + test("multiple listeners on the same workspace share one observation setup", async () => { const getPullRequestStatus = vi.fn(async () => createPullRequestStatusResult()); const getCheckoutSnapshotFacts = vi.fn(async (cwd: string) => createCheckoutSnapshotFacts(cwd)); diff --git a/packages/server/src/server/workspace-git-service.ts b/packages/server/src/server/workspace-git-service.ts index 83ca876ac..01909810d 100644 --- a/packages/server/src/server/workspace-git-service.ts +++ b/packages/server/src/server/workspace-git-service.ts @@ -3,6 +3,7 @@ import { readFile } from "node:fs/promises"; import { join, resolve } from "node:path"; import parcelWatcher from "@parcel/watcher"; import { LRUCache } from "lru-cache"; +import pLimit from "p-limit"; import type pino from "pino"; import type { ProjectCheckoutLitePayload } from "@getpaseo/protocol/messages"; import { parseGitRemoteLocation } from "@getpaseo/protocol/git-remote"; @@ -58,6 +59,11 @@ const FORGE_PR_STATUS_POLL_FAST_INTERVAL_MS = 20_000; const FORGE_PR_STATUS_POLL_SLOW_INTERVAL_MS = 120_000; const FORGE_PR_STATUS_POLL_ERROR_BACKOFF_CAP_MS = 300_000; const DEGRADED_GIT_POLL_INTERVAL_MS = 5_000; +// Keep whole workspace pipelines below the lower-level Git process pool so daemon control work +// retains subprocess and event-loop headroom during large workspace reconciliation bursts. +export const WORKSPACE_GIT_REFRESH_CONCURRENCY = 4; +export const WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY = 2; +export const WORKSPACE_GIT_WATCHER_SUBSCRIBE_TIMEOUT_MS = 10_000; const WATCH_RECOVERY_BASE_DELAY_MS = 30_000; const WATCH_RECOVERY_MAX_ATTEMPTS = 3; // Auxiliary reads may reuse cached values within this window; snapshots do not expire on read. @@ -196,6 +202,10 @@ export interface WorkspaceGitServiceMetrics { workingTreeWatchSetupInFlightCount: number; workspaceRefreshInFlightCount: number; workspaceRefreshQueuedCount: number; + workspaceRefreshAdmissionActiveCount: number; + workspaceRefreshAdmissionPendingCount: number; + workspaceObservationSetupAdmissionActiveCount: number; + workspaceObservationSetupAdmissionPendingCount: number; fetchInFlightCount: number; snapshotUpdatedListenerCount: number; watcherErrorCallbackCount: number; @@ -340,6 +350,22 @@ export function subscribeToWorkspaceFileChanges( }); } +class WorkspaceGitServiceDisposedError extends Error { + constructor() { + super("WorkspaceGitService is disposed"); + this.name = "WorkspaceGitServiceDisposedError"; + } +} + +class WorkspaceGitWatcherSubscriptionTimeoutError extends Error { + constructor(watchPath: string) { + super( + `Watcher subscription for ${watchPath} timed out after ${WORKSPACE_GIT_WATCHER_SUBSCRIBE_TIMEOUT_MS}ms`, + ); + this.name = "WorkspaceGitWatcherSubscriptionTimeoutError"; + } +} + interface WorkspaceGitTarget { cwd: string; listeners: Set; @@ -460,6 +486,16 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { private readonly worktreesRoot: string | undefined; private readonly deps: WorkspaceGitServiceDependencies; private readonly forgeResolver: ForgeResolver; + private readonly workspaceRefreshLimit = pLimit({ + concurrency: WORKSPACE_GIT_REFRESH_CONCURRENCY, + rejectOnClear: true, + }); + private readonly workspaceObservationSetupLimit = pLimit({ + concurrency: WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY, + rejectOnClear: true, + }); + private readonly disposeController = new AbortController(); + private disposed = false; private readonly snapshotUpdatedListeners = new Set(); private readonly workspaceTargets = new Map(); private readonly repoTargets = new Map(); @@ -507,6 +543,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { } resolveForge(cwd: string): Promise { + this.assertNotDisposed(); return this.forgeResolver.resolve(resolve(cwd)); } @@ -514,6 +551,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { params: { cwd: string }, listener: WorkspaceGitListener, ): WorkspaceGitSubscription { + this.assertNotDisposed(); const cwd = resolve(params.cwd); const target = this.ensureWorkspaceTarget(cwd); target.listeners.add(listener); @@ -533,6 +571,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { } onSnapshotUpdated(listener: WorkspaceGitSnapshotUpdatedListener): WorkspaceGitSubscription { + this.assertNotDisposed(); this.snapshotUpdatedListeners.add(listener); return { unsubscribe: () => { @@ -583,6 +622,12 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { workingTreeWatchSetupInFlightCount: this.workingTreeWatchSetups.size, workspaceRefreshInFlightCount, workspaceRefreshQueuedCount, + workspaceRefreshAdmissionActiveCount: this.workspaceRefreshLimit.activeCount, + workspaceRefreshAdmissionPendingCount: this.workspaceRefreshLimit.pendingCount, + workspaceObservationSetupAdmissionActiveCount: + this.workspaceObservationSetupLimit.activeCount, + workspaceObservationSetupAdmissionPendingCount: + this.workspaceObservationSetupLimit.pendingCount, fetchInFlightCount, snapshotUpdatedListenerCount: this.snapshotUpdatedListeners.size, watcherErrorCallbackCount: this.watcherErrorCallbackCount, @@ -593,6 +638,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { cwd: string, options?: WorkspaceGitSnapshotOptions, ): Promise { + this.assertNotDisposed(); cwd = resolve(cwd); const request = this.normalizeRefreshRequest(options, "getSnapshot", true); const target = this.ensureWorkspaceTarget(cwd); @@ -604,6 +650,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { } async getCheckout(cwd: string): Promise { + this.assertNotDisposed(); const normalizedCwd = resolve(cwd); const status = await this.deps.getCheckoutStatus(normalizedCwd, { paseoHome: this.paseoHome, @@ -640,6 +687,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { options: CheckoutDiffCompare, readOptions?: WorkspaceGitReadOptions, ): Promise { + this.assertNotDisposed(); const normalizedCwd = resolve(cwd); const normalizedOptions = this.normalizeCheckoutDiffOptions(options); const key = this.buildCheckoutDiffCacheKey(normalizedCwd, normalizedOptions); @@ -689,6 +737,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { ref: string, options?: WorkspaceGitReadOptions, ): Promise { + this.assertNotDisposed(); const normalizedCwd = resolve(cwd); const normalizedRef = ref.trim(); const key = JSON.stringify(["branch-validation", normalizedCwd, normalizedRef]); @@ -698,6 +747,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { } hasLocalBranch(cwd: string, branch: string, options?: WorkspaceGitReadOptions): Promise { + this.assertNotDisposed(); const normalizedCwd = resolve(cwd); const normalizedBranch = branch.trim(); const ref = `refs/heads/${normalizedBranch}`; @@ -717,6 +767,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { options?: WorkspaceGitBranchSuggestionsOptions, readOptions?: WorkspaceGitReadOptions, ): Promise { + this.assertNotDisposed(); const normalizedCwd = resolve(cwd); const query = options?.query ?? ""; const limit = options?.limit; @@ -731,6 +782,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { options?: WorkspaceGitStashListOptions, readOptions?: WorkspaceGitReadOptions, ): Promise { + this.assertNotDisposed(); const normalizedCwd = resolve(cwd); const paseoOnly = options?.paseoOnly !== false; const key = JSON.stringify(["stashes", normalizedCwd, paseoOnly]); @@ -747,6 +799,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { cwdOrRepoRoot: string, options?: WorkspaceGitReadOptions, ): Promise { + this.assertNotDisposed(); const repoRoot = await this.resolveRepoRoot(cwdOrRepoRoot, options); const key = JSON.stringify(["worktrees", repoRoot]); return this.readAuxiliaryCache(this.worktreeListCache, key, options, () => @@ -773,6 +826,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { cwdOrRepoRoot: string, options?: WorkspaceGitReadOptions, ): Promise { + this.assertNotDisposed(); const cwd = resolve(cwdOrRepoRoot); const key = JSON.stringify(["default-branch", cwd]); return this.readAuxiliaryCache(this.defaultBranchCache, key, options, async () => { @@ -798,6 +852,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { } async refresh(cwd: string, _options?: { priority?: "normal" | "high" }): Promise { + this.assertNotDisposed(); cwd = resolve(cwd); const target = this.ensureWorkspaceTarget(cwd); await this.refreshWorkspaceTarget(target, { @@ -816,6 +871,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { cwd: string, onChange: () => void, ): Promise<{ repoRoot: string | null; unsubscribe: () => void }> { + this.assertNotDisposed(); cwd = resolve(cwd); const target = await this.ensureWorkingTreeWatchTarget(cwd); target.listeners.add(onChange); @@ -829,6 +885,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { } scheduleRefreshForCwd(cwd: string): void { + this.assertNotDisposed(); cwd = resolve(cwd); const target = this.workspaceTargets.get(cwd); if (target) { @@ -837,6 +894,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { } onWorkspaceStateMayHaveChanged(cwd: string): void { + this.assertNotDisposed(); const normalizedCwd = resolve(cwd); const target = this.workspaceTargets.get(normalizedCwd); if (!target || target.closed) { @@ -856,10 +914,19 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { * git mutations to force a fresh forge status on the next refresh. */ invalidateForge(cwd: string): void { + this.assertNotDisposed(); this.forgeResolver.invalidate(resolve(cwd)); } dispose(): void { + if (this.disposed) { + return; + } + this.disposed = true; + this.disposeController.abort(new WorkspaceGitServiceDisposedError()); + this.workspaceRefreshLimit.clearQueue(); + this.workspaceObservationSetupLimit.clearQueue(); + for (const target of this.workspaceTargets.values()) { this.closeWorkspaceTarget(target); } @@ -880,7 +947,14 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { this.snapshotUpdatedListeners.clear(); } + private assertNotDisposed(): void { + if (this.disposed) { + throw new WorkspaceGitServiceDisposedError(); + } + } + private ensureWorkspaceTarget(cwd: string): WorkspaceGitTarget { + this.assertNotDisposed(); const existingTarget = this.workspaceTargets.get(cwd); if (existingTarget) { return existingTarget; @@ -895,6 +969,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { options: WorkspaceGitReadOptions | undefined, load: () => Promise, ): Promise { + this.assertNotDisposed(); if (options?.force && !options.reason) { throw new Error("WorkspaceGitService forced read requires a reason"); } @@ -951,6 +1026,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { } private ensureWorkingTreeWatchTarget(cwd: string): Promise { + this.assertNotDisposed(); const targetCwd = this.workingTreeWatchAliases.get(cwd); if (targetCwd) { const existingTarget = this.workingTreeWatchTargets.get(targetCwd); @@ -1067,9 +1143,16 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { return; } - target.observationSetupPromise = Promise.resolve() - .then(() => this.setupWorkspaceObservation(target)) + target.observationSetupPromise = this.workspaceObservationSetupLimit(async () => { + if (!this.isActiveObservedWorkspaceTarget(target)) { + return; + } + await this.setupWorkspaceObservation(target); + }) .catch((error) => { + if (this.disposed || !this.isActiveObservedWorkspaceTarget(target)) { + return; + } this.logger.warn( { err: error, cwd: target.cwd }, "Failed to set up workspace git observation", @@ -1183,6 +1266,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { this.workingTreeWatchTargets.set(cwd, target); this.workingTreeWatchAliases.set(cwd, cwd); await this.startWorkingTreeSubscription(target); + this.assertNotDisposed(); if (repoRoot === null) { this.startWorkingTreeWatchFallback(target, "not_a_git_checkout"); @@ -1191,6 +1275,95 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { return target; } + private async subscribeWithDeadline( + watchPath: string, + callback: parcelWatcher.SubscribeCallback, + options: parcelWatcher.Options, + onSubscribeSettled: () => void, + shouldAbandonSubscription: () => boolean, + ): Promise { + this.assertNotDisposed(); + const signal = this.disposeController.signal; + let outcome: "pending" | "accepted" | "expired" = "pending"; + let timeout: NodeJS.Timeout | null = null; + let removeAbortListener = () => {}; + let unsubscribePromise: Promise | null = null; + let subscriptionPromise: Promise; + try { + subscriptionPromise = this.deps + .subscribe(watchPath, callback, options) + .finally(onSubscribeSettled); + } catch (error) { + onSubscribeSettled(); + throw error; + } + void subscriptionPromise.then( + (subscription) => { + if ((outcome === "expired" || signal.aborted) && !shouldAbandonSubscription()) { + unsubscribePromise ??= this.unsubscribeWatcherSubscription(subscription, watchPath); + return unsubscribePromise; + } + return undefined; + }, + () => undefined, + ); + const timeoutPromise = new Promise((_resolve, reject) => { + timeout = setTimeout(() => { + outcome = "expired"; + reject(new WorkspaceGitWatcherSubscriptionTimeoutError(watchPath)); + }, WORKSPACE_GIT_WATCHER_SUBSCRIBE_TIMEOUT_MS); + }); + const disposalPromise = new Promise((_resolve, reject) => { + const rejectForDisposal = () => { + outcome = "expired"; + reject(signal.reason); + }; + if (signal.aborted) { + rejectForDisposal(); + return; + } + signal.addEventListener("abort", rejectForDisposal, { once: true }); + removeAbortListener = () => signal.removeEventListener("abort", rejectForDisposal); + }); + + try { + const subscription = await Promise.race([ + subscriptionPromise, + timeoutPromise, + disposalPromise, + ]); + if (signal.aborted) { + outcome = "expired"; + if (!shouldAbandonSubscription()) { + unsubscribePromise ??= this.unsubscribeWatcherSubscription(subscription, watchPath); + await unsubscribePromise; + } + throw signal.reason; + } + outcome = "accepted"; + return subscription; + } finally { + if (outcome === "pending") { + outcome = "expired"; + } + if (timeout) { + clearTimeout(timeout); + } + removeAbortListener(); + } + } + + private async unsubscribeWatcherSubscription( + subscription: parcelWatcher.AsyncSubscription, + watchPath: string, + ): Promise { + try { + await subscription.unsubscribe(); + } catch (error) { + this.logger.warn({ err: error, watchPath }, "Failed to stop watcher subscription"); + } + } + private async startWorkingTreeSubscription( target: WorkingTreeWatchTarget, options?: { replaceFallback?: boolean }, @@ -1198,8 +1371,14 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { const ignore = [join(target.watchPath, ".git"), ...target.ignoredDirectories]; let watcherErrored = false; let subscribeSettled = false; + const markSubscribeSettled = () => { + subscribeSettled = true; + if (watcherErrored) { + this.scheduleWorkingTreeWatchRecovery(target); + } + }; try { - const subscription = await this.deps.subscribe( + const subscription = await this.subscribeWithDeadline( target.watchPath, (error, events) => { if (error) { @@ -1227,10 +1406,10 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { this.notifyWorkingTreeChanged(target, "working-tree-watch"); }, { ignore }, + markSubscribeSettled, + () => watcherErrored, ); - subscribeSettled = true; if (watcherErrored) { - this.scheduleWorkingTreeWatchRecovery(target); return false; } if ( @@ -1238,7 +1417,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { (target.fallbackPolling && !options?.replaceFallback) || target.subscription ) { - await subscription.unsubscribe(); + await this.unsubscribeWatcherSubscription(subscription, target.watchPath); return false; } target.subscription = subscription; @@ -1251,11 +1430,12 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { } return true; } catch (error) { - subscribeSettled = true; if (watcherErrored) { - this.scheduleWorkingTreeWatchRecovery(target); return false; } + if (this.disposed || target.closed) { + throw error; + } this.logger.warn( { err: error, cwd: target.cwd }, "Failed to start working tree watcher; using degraded polling", @@ -1271,7 +1451,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { target: WorkingTreeWatchTarget, reason: WorkingTreeWatchFallbackReason, ): void { - if (target.fallbackPolling) { + if (this.disposed || target.closed || target.fallbackPolling) { return; } target.fallbackPolling = true; @@ -1606,8 +1786,14 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { const matchesRepoGitRoot = createRealpathAwarePathMatcher(target.repoGitRoot); let watcherErrored = false; let subscribeSettled = false; + const markSubscribeSettled = () => { + subscribeSettled = true; + if (watcherErrored) { + this.scheduleRepoMetadataWatchRecovery(target); + } + }; try { - const subscription = await this.deps.subscribe( + const subscription = await this.subscribeWithDeadline( target.repoGitRoot, (error, events) => { if (error) { @@ -1644,10 +1830,10 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { } }, { ignore }, + markSubscribeSettled, + () => watcherErrored, ); - subscribeSettled = true; if (watcherErrored) { - this.scheduleRepoMetadataWatchRecovery(target); return false; } if ( @@ -1655,7 +1841,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { (target.fallbackPolling && !options?.replaceFallback) || this.repoTargets.get(target.repoGitRoot) !== target ) { - await subscription.unsubscribe(); + await this.unsubscribeWatcherSubscription(subscription, target.repoGitRoot); return false; } target.subscription = subscription; @@ -1668,11 +1854,12 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { } return true; } catch (error) { - subscribeSettled = true; if (watcherErrored) { - this.scheduleRepoMetadataWatchRecovery(target); return false; } + if (this.disposed || target.closed) { + throw error; + } this.logger.warn( { err: error, repoGitRoot: target.repoGitRoot }, "Failed to start repository metadata watcher; using degraded polling", @@ -2280,6 +2467,9 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { try { await this.requestWorkspaceSnapshot(target, request); } catch (error) { + if (this.disposed || target.closed) { + return; + } this.logger.warn( { err: error, cwd: target.cwd, reason: request.reason }, "Failed to refresh workspace git snapshot", @@ -2419,7 +2609,16 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { }); } try { - snapshot = await this.refreshSnapshot(target, request); + const admittedSnapshot = await this.workspaceRefreshLimit(() => { + if (target.closed || this.workspaceTargets.get(target.cwd) !== target) { + return null; + } + return this.refreshSnapshot(target, request); + }); + if (!admittedSnapshot) { + break; + } + snapshot = admittedSnapshot; this.rememberSnapshot(target, snapshot, { notify: request.notify, forceEmit: request.force, @@ -2429,6 +2628,10 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { failure = { error }; } + if (this.disposed || target.closed || this.workspaceTargets.get(target.cwd) !== target) { + break; + } + const state = target.refreshState; if (state.status !== "in-flight" || !state.queued) { break; diff --git a/packages/server/src/utils/tree-kill.ts b/packages/server/src/utils/tree-kill.ts index 8101be987..5de58db22 100644 --- a/packages/server/src/utils/tree-kill.ts +++ b/packages/server/src/utils/tree-kill.ts @@ -38,13 +38,13 @@ export async function terminateWithTreeKill( } const exitPromise = waitForProcessExit(child); - await signalTreeOrChild(child, options.gracefulSignal ?? "SIGTERM"); + await signalProcessTree(child, options.gracefulSignal ?? "SIGTERM"); if (await waitForExitOrTimeout(exitPromise, options.gracefulTimeoutMs)) { return "terminated"; } options.onForceSignal?.(); - await signalTreeOrChild(child, options.forceSignal ?? "SIGKILL"); + await signalProcessTree(child, options.forceSignal ?? "SIGKILL"); if (options.forceTimeoutMs === undefined) { return "killed"; } @@ -53,7 +53,7 @@ export async function terminateWithTreeKill( : "kill-timeout"; } -function signalTreeOrChild(child: TreeKillTarget, signal: NodeJS.Signals): Promise { +export function signalProcessTree(child: TreeKillTarget, signal: NodeJS.Signals): Promise { if (isProcessExited(child)) { return Promise.resolve(); }