diff --git a/apps/server/src/vcs/VcsProcess.test.ts b/apps/server/src/vcs/VcsProcess.test.ts index 675d20cb82c..63aba076086 100644 --- a/apps/server/src/vcs/VcsProcess.test.ts +++ b/apps/server/src/vcs/VcsProcess.test.ts @@ -1,5 +1,6 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { describe, expect, it } from "@effect/vitest"; +import * as Deferred from "effect/Deferred"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; @@ -45,6 +46,49 @@ const captureProcessResult = ( ); describe("VcsProcess.run", () => { + it.effect("bounds how many processes it spawns at once", () => + Effect.gen(function* () { + const release = yield* Deferred.make(); + let inFlight = 0; + let peakInFlight = 0; + + const service = yield* VcsProcess.make.pipe( + Effect.provideService( + ProcessRunner.ProcessRunner, + ProcessRunner.ProcessRunner.of({ + run: () => + Effect.gen(function* () { + inFlight += 1; + peakInFlight = Math.max(peakInFlight, inFlight); + yield* Deferred.await(release); + inFlight -= 1; + return { + code: 0, + stdout: "", + stderr: "", + stdoutTruncated: false, + stderrTruncated: false, + } as ProcessRunner.ProcessRunOutput; + }), + }), + ), + ); + + const fibers = yield* Effect.forEach( + Array.from({ length: 40 }, (_, index) => index), + () => Effect.forkChild(service.run(baseInput), { startImmediately: true }), + ); + // Every fiber started eagerly and ran until it blocked, either on a + // permit or on the release latch. + expect(peakInFlight).toBeLessThanOrEqual(16); + expect(peakInFlight).toBeLessThan(40); + + yield* Deferred.succeed(release, undefined); + yield* Fiber.awaitAll(fibers); + expect(peakInFlight).toBeLessThanOrEqual(16); + }), + ); + it.effect("collects stdout", () => Effect.gen(function* () { const result = yield* run({ diff --git a/apps/server/src/vcs/VcsProcess.ts b/apps/server/src/vcs/VcsProcess.ts index 52db6f9b1fb..d904ceeda6b 100644 --- a/apps/server/src/vcs/VcsProcess.ts +++ b/apps/server/src/vcs/VcsProcess.ts @@ -1,7 +1,10 @@ +import * as NodeOS from "node:os"; + import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Match from "effect/Match"; +import * as Semaphore from "effect/Semaphore"; import { ChildProcessSpawner } from "effect/unstable/process"; import { @@ -86,8 +89,24 @@ const classifyNonZeroExit = (command: string, stderr: string): VcsProcessExitFai return "command-failed"; }; +// A single workspace refresh can ask for refs across every worktree at once, and +// each request spawns several git processes. Unbounded, that starves the event +// loop badly enough that unrelated work (provider health checks, HTTP requests) +// blows past its own timeouts. Bound the spawns instead; queued commands wait +// for a permit rather than competing. +// ponytail: one global limit, make it per-repository if a busy repo ever starves +// the others. +// The bound scales with the host: a single git command costs ~1s on a large +// repository, so too low a limit turns a refresh into a queue, while too high a +// limit is what caused the starvation in the first place. +const MAX_CONCURRENT_VCS_PROCESSES = Math.max( + 4, + Math.min(16, Math.floor(NodeOS.availableParallelism() / 4)), +); + export const make = Effect.gen(function* () { const processRunner = yield* ProcessRunner.ProcessRunner; + const spawnPermits = yield* Semaphore.make(MAX_CONCURRENT_VCS_PROCESSES); const run = Effect.fn("VcsProcess.run")(function* (input: VcsProcessInput) { const baseError = { @@ -112,6 +131,9 @@ export const make = Effect.gen(function* () { timeoutBehavior: "error", }) .pipe( + // Take the permit around the spawn only, so waiting in the queue never + // counts against the command's own timeout. + spawnPermits.withPermits(1), Effect.mapError( Match.valueTags({ ProcessSpawnError: (error) => diff --git a/packages/shared/src/shell.test.ts b/packages/shared/src/shell.test.ts index e8b2c41cb77..3b24581977a 100644 --- a/packages/shared/src/shell.test.ts +++ b/packages/shared/src/shell.test.ts @@ -2,9 +2,12 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { it as effectIt } from "@effect/vitest"; import { HostProcessEnvironment, HostProcessPlatform } from "@t3tools/shared/hostProcess"; import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Path from "effect/Path"; import { describe, expect, it, vi } from "vite-plus/test"; import { + clearCommandPathCache, extractPathFromShellOutput, CommandAvailability, type CommandAvailabilityChecker, @@ -364,6 +367,32 @@ effectIt.layer(NodeServices.layer)("resolveCommandPath", (it) => { expect(result._tag).toBe("Failure"); }), ); + + it.effect("reuses a resolution until the cache is cleared", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const directory = yield* fs.makeTempDirectoryScoped(); + const toolPath = path.join(directory, "tool.cmd"); + yield* fs.writeFileString(toolPath, "@echo off\n"); + const env = { PATH: directory, PATHEXT: ".CMD" }; + const resolve = () => + resolveCommandPath("tool", { env }).pipe( + Effect.provideService(HostProcessPlatform, "win32"), + Effect.result, + ); + + clearCommandPathCache(); + const resolved = yield* resolve(); + expect(resolved._tag).toBe("Success"); + + yield* fs.remove(toolPath); + expect(yield* resolve()).toStrictEqual(resolved); + + clearCommandPathCache(); + expect((yield* resolve())._tag).toBe("Failure"); + }).pipe(Effect.scoped), + ); }); effectIt.layer(NodeServices.layer)("resolveSpawnCommand", (it) => { diff --git a/packages/shared/src/shell.ts b/packages/shared/src/shell.ts index cf2f2417ff4..8f35ed3e8d9 100644 --- a/packages/shared/src/shell.ts +++ b/packages/shared/src/shell.ts @@ -3,6 +3,7 @@ import * as NodeOS from "node:os"; import * as NodePath from "node:path"; import * as NodeChildProcess from "node:child_process"; import * as NodeFS from "node:fs"; +import * as Clock from "effect/Clock"; import * as Data from "effect/Data"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; @@ -491,7 +492,9 @@ function resolveCommandCandidates( return Array.from(new Set(candidates)); } -const isExecutableFile = Effect.fn("shell.isExecutableFile")(function* ( +// Called once per PATH x PATHEXT candidate, so a single miss on Windows probes +// hundreds of paths. Keep it untraced: a span per probe floods the tracer. +const isExecutableFile = Effect.fnUntraced(function* ( filePath: string, platform: NodeJS.Platform, windowsPathExtensions: ReadonlyArray, @@ -510,7 +513,21 @@ const isExecutableFile = Effect.fn("shell.isExecutableFile")(function* ( return canExecuteFile(filePath); }); -const resolveCommandPathForPlatform = Effect.fn("shell.resolveCommandPathForPlatform")(function* ( +// A miss on Windows probes every PATH entry against every PATHEXT variant, and +// editor/provider refreshes resolve the same commands over and over. Cache the +// outcome briefly; the TTL keeps a newly installed CLI discoverable. +const COMMAND_PATH_CACHE_TTL_MS = 30_000; + +const commandPathCache = new Map< + string, + { readonly expiresAt: number; readonly resolved: string | null } +>(); + +export function clearCommandPathCache(): void { + commandPathCache.clear(); +} + +const resolveCommandPathUncached = Effect.fn("shell.resolveCommandPathForPlatform")(function* ( command: string, options: CommandAvailabilityOptions & { readonly platform: NodeJS.Platform }, ): Effect.fn.Return { @@ -557,6 +574,36 @@ const resolveCommandPathForPlatform = Effect.fn("shell.resolveCommandPathForPlat return yield* new CommandResolutionError({ command, reason: "not-found" }); }); +const resolveCommandPathForPlatform = Effect.fnUntraced(function* ( + command: string, + options: CommandAvailabilityOptions & { readonly platform: NodeJS.Platform }, +): Effect.fn.Return { + const env = options.env ?? process.env; + const cacheKey = [ + options.platform, + command, + resolvePathEnvironmentVariable(env), + env.PATHEXT ?? "", + ].join("\u0000"); + const now = yield* Clock.currentTimeMillis; + const cached = commandPathCache.get(cacheKey); + if (cached !== undefined && cached.expiresAt > now) { + if (cached.resolved === null) { + return yield* new CommandResolutionError({ command, reason: "not-found" }); + } + return cached.resolved; + } + + const resolved = yield* resolveCommandPathUncached(command, options).pipe( + Effect.catchTag("CommandResolutionError", () => Effect.succeed(null)), + ); + commandPathCache.set(cacheKey, { expiresAt: now + COMMAND_PATH_CACHE_TTL_MS, resolved }); + if (resolved === null) { + return yield* new CommandResolutionError({ command, reason: "not-found" }); + } + return resolved; +}); + export const resolveCommandPath = Effect.fn("shell.resolveCommandPath")(function* ( command: string, options: CommandAvailabilityOptions = {},