From fb67d2cdc87c12a1057fae8602087bf9a80b6f22 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Fri, 28 Aug 2026 10:25:20 +0200 Subject: [PATCH] Keep the daemon responsive during Git queue pressure (#3945) * fix(server): keep Git pressure from stalling daemon Git process starts were refilled synchronously, so slow process spawning could monopolize the main loop long enough to trip connection liveness checks. Admit one process per event-loop turn and expose refresh/fetch provenance in runtime queue metrics. * perf(server): avoid delaying isolated Git commands Start the first command immediately when the scheduler is idle. Backlog admission and completion-driven refills still yield between spawns, preserving daemon liveness without adding a turn to normal sequential Git traffic. * fix(server): keep Git admission outside fake clocks Subprocess admission is event-loop ordering rather than simulated wall-clock behavior. Use the native Node immediate so service tests can fake debounce timers without deadlocking queued real Git processes. * refactor(server): bind Git command provenance explicitly * test(server): expect bound Git fetch runner --- .../src/server/workspace-git-fetch.test.ts | 33 ++++ .../server/src/server/workspace-git-fetch.ts | 13 +- .../workspace-git-service.observation.test.ts | 6 +- .../workspace-git-service.primitive.test.ts | 47 ++++- .../src/server/workspace-git-service.ts | 76 ++++++-- packages/server/src/utils/checkout-git.ts | 184 +++++++++++------- .../utils/git-command-runtime-metrics.test.ts | 13 +- .../src/utils/git-command-runtime-metrics.ts | 31 ++- .../src/utils/git-process-scheduler.test.ts | 74 ++++++- .../server/src/utils/git-process-scheduler.ts | 44 +++-- .../server/src/utils/run-git-command.test.ts | 14 ++ packages/server/src/utils/run-git-command.ts | 16 +- 12 files changed, 429 insertions(+), 122 deletions(-) diff --git a/packages/server/src/server/workspace-git-fetch.test.ts b/packages/server/src/server/workspace-git-fetch.test.ts index c6d1c943d1..29a95b64d8 100644 --- a/packages/server/src/server/workspace-git-fetch.test.ts +++ b/packages/server/src/server/workspace-git-fetch.test.ts @@ -3,10 +3,43 @@ import { describe, expect, test } from "vitest"; import { diffWorkspaceGitRefs, diffWorkspaceGitRemoteRefs, + fetchWorkspaceGitRemote, parseWorkspaceGitRefs, } from "./workspace-git-fetch.js"; +import type { RunGitCommand } from "../utils/run-git-command.js"; describe("workspace Git remote refs", () => { + test("runs the complete fetch through the supplied Git runner", async () => { + const commands: string[][] = []; + let refRead = 0; + const runGitCommand: RunGitCommand = async (args) => { + commands.push(args); + const stdout = + args[0] === "for-each-ref" + ? `refs/remotes/origin/main\0${refRead++ === 0 ? "aaaaaaaa" : "bbbbbbbb"}` + : ""; + return { stdout, stderr: "", truncated: false, exitCode: 0, signal: null }; + }; + + await expect( + fetchWorkspaceGitRemote("/repo", { onRefSnapshot() {} }, runGitCommand), + ).resolves.toMatchObject({ + changes: [ + { + kind: "moved", + ref: "origin/main", + beforeOid: "aaaaaaaa", + afterOid: "bbbbbbbb", + }, + ], + }); + expect(commands).toEqual([ + ["for-each-ref", "--format=%(refname)%00%(objectname)"], + ["fetch", "origin", "--prune"], + ["for-each-ref", "--format=%(refname)%00%(objectname)"], + ]); + }); + test("parses refs for semantic before/after comparison", () => { expect( parseWorkspaceGitRefs( diff --git a/packages/server/src/server/workspace-git-fetch.ts b/packages/server/src/server/workspace-git-fetch.ts index 795066c386..72f5d407da 100644 --- a/packages/server/src/server/workspace-git-fetch.ts +++ b/packages/server/src/server/workspace-git-fetch.ts @@ -1,4 +1,4 @@ -import { runGitCommand } from "../utils/run-git-command.js"; +import { runGitCommand, type RunGitCommand } from "../utils/run-git-command.js"; export interface WorkspaceGitFetchResult { changes: WorkspaceGitRemoteRefChange[] | null; @@ -19,18 +19,19 @@ export type WorkspaceGitRemoteRefChange = export async function fetchWorkspaceGitRemote( cwd: string, observer: WorkspaceGitFetchObserver, + run: RunGitCommand = runGitCommand, ): Promise { let before: Map | null = null; let after: Map | null = null; let error: unknown | null = null; try { - before = await readWorkspaceGitRefs(cwd); + before = await readWorkspaceGitRefs(cwd, run); observer.onRefSnapshot("before"); } catch (caught) { error = caught; } try { - await runGitCommand(["fetch", "origin", "--prune"], { + await run(["fetch", "origin", "--prune"], { cwd, envOverlay: { GIT_TERMINAL_PROMPT: "0" }, timeout: 120_000, @@ -39,7 +40,7 @@ export async function fetchWorkspaceGitRemote( error ??= caught; } try { - after = await readWorkspaceGitRefs(cwd); + after = await readWorkspaceGitRefs(cwd, run); } catch (caught) { error ??= caught; } @@ -66,8 +67,8 @@ function collectWorkspaceGitRemoteRefs(refs: ReadonlyMap): Set> { - const { stdout } = await runGitCommand(["for-each-ref", "--format=%(refname)%00%(objectname)"], { +async function readWorkspaceGitRefs(cwd: string, run: RunGitCommand): Promise> { + const { stdout } = await run(["for-each-ref", "--format=%(refname)%00%(objectname)"], { cwd, }); return parseWorkspaceGitRefs(stdout); 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 e863ebb4a0..ad80acb71f 100644 --- a/packages/server/src/server/workspace-git-service.observation.test.ts +++ b/packages/server/src/server/workspace-git-service.observation.test.ts @@ -1389,7 +1389,11 @@ describe("WorkspaceGitService checkout observation", () => { await vi.waitFor(() => { expect(runGitFetch).toHaveBeenCalledTimes(2); }); - expect(runGitFetch).toHaveBeenLastCalledWith(worktrees[1], expect.anything()); + expect(runGitFetch).toHaveBeenLastCalledWith( + worktrees[1], + expect.anything(), + expect.anything(), + ); for (const subscription of subscriptions.slice(1)) { subscription.unsubscribe(); diff --git a/packages/server/src/server/workspace-git-service.primitive.test.ts b/packages/server/src/server/workspace-git-service.primitive.test.ts index 4cccaa6e55..00f84f2aa5 100644 --- a/packages/server/src/server/workspace-git-service.primitive.test.ts +++ b/packages/server/src/server/workspace-git-service.primitive.test.ts @@ -17,7 +17,10 @@ import { type CheckoutStatusGit, type PullRequestStatusResult, } from "../utils/checkout-git.js"; -import { runGitCommand as runGitCommandReal } from "../utils/run-git-command.js"; +import { + runGitCommand as runGitCommandReal, + snapshotGitCommandRuntimeMetrics, +} from "../utils/run-git-command.js"; import { getWorkspaceGitSelfHealPhaseMs, WorkspaceGitServiceImpl, @@ -448,6 +451,48 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => { service.dispose(); }); + test("attributes Git commands submitted by a workspace refresh", async () => { + const tempDir = realpathSync(mkdtempSync(join(tmpdir(), "workspace-git-provenance-"))); + const repoDir = join(tempDir, "repo"); + mkdirSync(repoDir, { recursive: true }); + execFileSync("git", ["init", "-b", "main"], { cwd: repoDir, stdio: "pipe" }); + writeFileSync(join(repoDir, "tracked.txt"), "tracked\n"); + execFileSync("git", ["add", "tracked.txt"], { cwd: repoDir, stdio: "pipe" }); + execFileSync( + "git", + [ + "-c", + "commit.gpgsign=false", + "-c", + "user.name=Paseo Test", + "-c", + "user.email=paseo@example.test", + "commit", + "-m", + "initial", + ], + { cwd: repoDir, stdio: "pipe" }, + ); + snapshotGitCommandRuntimeMetrics(); + const service = createService({ + getCheckoutSnapshotFacts: getCheckoutSnapshotFactsUncached as never, + getCheckoutStatus: getCheckoutStatusUncached as never, + }); + + try { + await service.getSnapshot(repoDir, { force: true, reason: "provenance-test" }); + + const metrics = snapshotGitCommandRuntimeMetrics(); + expect(metrics.submitted).toBeGreaterThan(0); + expect(metrics.provenanceTop).toEqual([ + ["workspace-refresh:provenance-test", metrics.submitted], + ]); + } finally { + service.dispose(); + rmSync(tempDir, { recursive: true, force: true }); + } + }); + test("getSnapshot cold-loads when no snapshot exists yet with one shell burst", async () => { const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd)); const getCheckoutShortstat = vi.fn(async () => ({ additions: 1, deletions: 0 })); diff --git a/packages/server/src/server/workspace-git-service.ts b/packages/server/src/server/workspace-git-service.ts index bcc550d64a..5304de646d 100644 --- a/packages/server/src/server/workspace-git-service.ts +++ b/packages/server/src/server/workspace-git-service.ts @@ -46,7 +46,11 @@ import { getRealpathAwareRelativePath, isRealpathInsideRoot, } from "../utils/path.js"; -import { runGitCommand } from "../utils/run-git-command.js"; +import { + createRunGitCommand, + runGitCommand, + type RunGitCommand, +} from "../utils/run-git-command.js"; import { branchNameFromRef } from "../utils/worktree-metadata.js"; import { listPaseoWorktrees, type PaseoWorktreeInfo } from "../utils/worktree.js"; import { READ_ONLY_GIT_ENV } from "./checkout-git-utils.js"; @@ -351,6 +355,7 @@ interface WorkspaceGitServiceDependencies { runGitFetch: ( cwd: string, observer: WorkspaceGitFetchObserver, + runGitCommand: RunGitCommand, ) => Promise; runGitCommand: typeof runGitCommand; getWorkspaceGitSelfHealPhaseMs: typeof getWorkspaceGitObservationReensurePhaseMs; @@ -2776,11 +2781,12 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { }); } try { + const runRefreshGitCommand = createRunGitCommand(`workspace-refresh:${request.reason}`); const admittedSnapshot = await this.workspaceRefreshLimit(() => { if (target.closed || this.workspaceTargets.get(target.cwd) !== target) { return null; } - return this.refreshSnapshot(target, request); + return this.refreshSnapshot(target, request, runRefreshGitCommand); }); if (!admittedSnapshot) { break; @@ -2819,12 +2825,18 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { private async refreshSnapshot( target: WorkspaceGitTarget, request: WorkspaceGitRefreshRequest, + runRefreshGitCommand: RunGitCommand, ): Promise { let facts = target.latestFacts; if (request.movedRemoteRefs.size > 0 && !request.refreshStructure && !request.refreshWorktree) { if (facts?.isGit && target.latestGit?.isGit) { try { - await this.refreshRefDerivedSnapshot(target, facts, request.movedRemoteRefs); + await this.refreshRefDerivedSnapshot( + target, + facts, + request.movedRemoteRefs, + runRefreshGitCommand, + ); return this.combineSnapshot(target); } catch (error) { this.logger.debug( @@ -2833,23 +2845,27 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { ); } } - facts = await this.refreshGitSnapshot(target, { - ...request, - refreshStructure: true, - refreshWorktree: true, - }); + facts = await this.refreshGitSnapshot( + target, + { + ...request, + refreshStructure: true, + refreshWorktree: true, + }, + runRefreshGitCommand, + ); return this.combineSnapshot(target); } if (request.refreshStructure || !facts || !target.latestGit) { - facts = await this.refreshGitSnapshot(target, request); + facts = await this.refreshGitSnapshot(target, request, runRefreshGitCommand); } else if (request.refreshWorktree) { - await this.refreshWorktreeSnapshot(target, facts); + await this.refreshWorktreeSnapshot(target, facts, runRefreshGitCommand); } if (!facts) { - facts = await this.refreshGitSnapshot(target, request); + facts = await this.refreshGitSnapshot(target, request, runRefreshGitCommand); } if (request.includeForge) { - await this.refreshForgeSnapshot(target, request, facts); + await this.refreshForgeSnapshot(target, request, facts, runRefreshGitCommand); } const snapshot = this.combineSnapshot(target); @@ -2861,6 +2877,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { target: WorkspaceGitTarget, facts: Extract, movedRemoteRefs: ReadonlySet, + runRefreshGitCommand: RunGitCommand, ): Promise { const latestGit = target.latestGit; if (!latestGit?.isGit) { @@ -2872,7 +2889,13 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { facts, { aheadBehind: latestGit.aheadBehind, diffStat: latestGit.diffStat }, movedRemoteRefs, - { paseoHome: this.paseoHome, worktreesRoot: this.worktreesRoot, logger: this.logger, facts }, + { + paseoHome: this.paseoHome, + worktreesRoot: this.worktreesRoot, + logger: this.logger, + facts, + runGitCommand: runRefreshGitCommand, + }, ); target.latestFacts = { ...facts, upstreamStatus: derived.upstreamStatus }; target.latestGit = { @@ -2890,6 +2913,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { private async refreshWorktreeSnapshot( target: WorkspaceGitTarget, facts: CheckoutSnapshotFacts, + runRefreshGitCommand: RunGitCommand, ): Promise { const latestGit = target.latestGit; if (!latestGit || !facts.isGit) { @@ -2902,6 +2926,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { worktreesRoot: this.worktreesRoot, logger: this.logger, facts, + runGitCommand: runRefreshGitCommand, }; const worktree = await this.deps.getCheckoutWorktreeState(target.cwd, context); target.latestGit = { @@ -2916,6 +2941,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { private async refreshGitSnapshot( target: WorkspaceGitTarget, request: WorkspaceGitRefreshRequest, + runRefreshGitCommand: RunGitCommand, ): Promise { const now = this.deps.now(); target.lastShellOutAtMs = now.getTime(); @@ -2926,6 +2952,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { paseoHome: this.paseoHome, worktreesRoot: this.worktreesRoot, logger: this.logger, + runGitCommand: runRefreshGitCommand, }; const facts = await this.loadCheckoutFacts(target, baseContext); const context: CheckoutContext = { ...baseContext, facts }; @@ -2980,6 +3007,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { target: WorkspaceGitTarget, request: WorkspaceGitRefreshRequest, facts: CheckoutSnapshotFacts, + runRefreshGitCommand: RunGitCommand, ): Promise { const remoteUrl = target.latestGit?.remoteUrl ?? null; const resolution = await this.forgeResolver.resolveFromRemoteUrlAsync(remoteUrl); @@ -3006,6 +3034,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { force: forceForge, reason: request.reason, facts, + runGitCommand: runRefreshGitCommand, }); // Carry the resolved forge (probe-aware) so the wire projection labels // self-managed GitLab hosts correctly instead of falling back to "github". @@ -3112,14 +3141,18 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { let result: WorkspaceGitFetchResult | null = null; const eventsBeforeFetchSnapshot: FileChange[] = []; try { - result = await this.deps.runGitFetch(target.cwd, { - onRefSnapshot: (phase) => { - const events = target.bufferedFetchMetadataEvents.splice(0); - if (phase === "before") { - eventsBeforeFetchSnapshot.push(...events); - } + result = await this.deps.runGitFetch( + target.cwd, + { + onRefSnapshot: (phase) => { + const events = target.bufferedFetchMetadataEvents.splice(0); + if (phase === "before") { + eventsBeforeFetchSnapshot.push(...events); + } + }, }, - }); + createRunGitCommand("background-fetch"), + ); } catch (error) { this.logger.warn( { err: error, repoGitRoot: target.repoGitRoot, cwd: target.cwd }, @@ -3353,6 +3386,7 @@ async function loadForgeSnapshot(options: { force?: boolean; reason?: string; facts?: CheckoutSnapshotFacts; + runGitCommand: RunGitCommand; }): Promise { const forgeService = options.forgeService; if (!forgeService) { @@ -3379,7 +3413,7 @@ async function loadForgeSnapshot(options: { force: options.force, reason: options.reason, }, - { facts: options.facts }, + { facts: options.facts, runGitCommand: options.runGitCommand }, ); return buildForgeSnapshot(result.authState, result.status, null); } catch (error) { diff --git a/packages/server/src/utils/checkout-git.ts b/packages/server/src/utils/checkout-git.ts index b65e5ca5e4..c0ae12f8f0 100644 --- a/packages/server/src/utils/checkout-git.ts +++ b/packages/server/src/utils/checkout-git.ts @@ -28,7 +28,7 @@ import { ForgeCommandError, } from "../services/forge-cli-command.js"; import { parseGitRevParsePath, resolveGitRevParsePath } from "./git-rev-parse-path.js"; -import { runGitCommand } from "./run-git-command.js"; +import { runGitCommand, type RunGitCommand } from "./run-git-command.js"; import { isPaseoOwnedWorktreeCwd, resolvePaseoWorktreesBaseRoot } from "./worktree.js"; import { branchNameFromRef, @@ -848,6 +848,7 @@ export interface CheckoutContext { worktreesRoot?: string; logger?: Pick; facts?: CheckoutSnapshotFacts | null; + runGitCommand?: RunGitCommand; } export type CheckoutSnapshotFacts = @@ -876,9 +877,16 @@ function isNotGitRepositoryError(error: unknown): boolean { return error instanceof Error && /not a git repository/i.test(error.message); } -async function requireGitRepo(cwd: string): Promise { +function getRunGitCommand(context?: CheckoutContext): RunGitCommand { + return context?.runGitCommand ?? runGitCommand; +} + +async function requireGitRepo(cwd: string, context?: CheckoutContext): Promise { try { - await runGitCommand(["rev-parse", "--git-dir"], { cwd, envOverlay: READ_ONLY_GIT_ENV }); + await getRunGitCommand(context)(["rev-parse", "--git-dir"], { + cwd, + envOverlay: READ_ONLY_GIT_ENV, + }); } catch { throw new NotGitRepoError(cwd); } @@ -900,15 +908,18 @@ async function requireGitWorktreeRoot(cwd: string): Promise { } } -export async function getCurrentBranch(cwd: string): Promise { +export async function getCurrentBranch( + cwd: string, + context?: CheckoutContext, +): Promise { try { - const { stdout } = await runGitCommand(["rev-parse", "--abbrev-ref", "HEAD"], { + const { stdout } = await getRunGitCommand(context)(["rev-parse", "--abbrev-ref", "HEAD"], { cwd, envOverlay: READ_ONLY_GIT_ENV, }); const branch = stdout.trim(); if (branch === "HEAD") { - return await getRebaseHeadBranch(cwd); + return await getRebaseHeadBranch(cwd, context); } return branch.length > 0 ? branch : null; } catch { @@ -924,7 +935,7 @@ async function getCurrentHeadSha(cwd: string, context?: CheckoutContext): Promis return knownSha; } try { - const { stdout } = await runGitCommand(["rev-parse", "HEAD"], { + const { stdout } = await getRunGitCommand(context)(["rev-parse", "HEAD"], { cwd, envOverlay: READ_ONLY_GIT_ENV, logger: context?.logger, @@ -948,12 +959,12 @@ async function addHeadShaToPullRequestLookupTarget( return headSha ? { ...target, headSha } : target; } -async function getRebaseHeadBranch(cwd: string): Promise { +async function getRebaseHeadBranch(cwd: string, context?: CheckoutContext): Promise { const paths = ["rebase-merge/head-name", "rebase-apply/head-name"]; const results = await Promise.all( paths.map(async (path): Promise => { try { - const { stdout } = await runGitCommand(["rev-parse", "--git-path", path], { + const { stdout } = await getRunGitCommand(context)(["rev-parse", "--git-path", path], { cwd, envOverlay: READ_ONLY_GIT_ENV, }); @@ -972,7 +983,7 @@ async function getRebaseHeadBranch(cwd: string): Promise { async function getWorktreeRoot(cwd: string, context?: CheckoutContext): Promise { try { - const { stdout } = await runGitCommand(["rev-parse", "--show-toplevel"], { + const { stdout } = await getRunGitCommand(context)(["rev-parse", "--show-toplevel"], { cwd, envOverlay: READ_ONLY_GIT_ENV, logger: context?.logger, @@ -989,12 +1000,15 @@ async function getWorktreeRoot(cwd: string, context?: CheckoutContext): Promise< } } -export async function getMainRepoRoot(cwd: string): Promise { - const { stdout: commonDirOut } = await runGitCommand(["rev-parse", "--git-common-dir"], { - cwd, - envOverlay: READ_ONLY_GIT_ENV, - }); - return getMainRepoRootFromCommonDir(cwd, resolveGitRevParsePath(cwd, commonDirOut)); +export async function getMainRepoRoot(cwd: string, context?: CheckoutContext): Promise { + const { stdout: commonDirOut } = await getRunGitCommand(context)( + ["rev-parse", "--git-common-dir"], + { + cwd, + envOverlay: READ_ONLY_GIT_ENV, + }, + ); + return getMainRepoRootFromCommonDir(cwd, resolveGitRevParsePath(cwd, commonDirOut), context); } async function getMainRepoRootFromCommonDir( @@ -1011,10 +1025,13 @@ async function getMainRepoRootFromCommonDir( return dirname(normalized); } - const { stdout: worktreeOut } = await runGitCommand(["worktree", "list", "--porcelain"], { - cwd, - envOverlay: READ_ONLY_GIT_ENV, - }); + const { stdout: worktreeOut } = await getRunGitCommand(context)( + ["worktree", "list", "--porcelain"], + { + cwd, + envOverlay: READ_ONLY_GIT_ENV, + }, + ); const worktrees = parseWorktreeList(worktreeOut); const nonBareNonPaseo = worktrees.filter( (wt) => @@ -1217,7 +1234,7 @@ async function resolveBaseRefForCwd( const storedBaseRef = await getStoredBaseRefForCwd(cwd, context); return { storedBaseRef, - resolvedBaseRef: storedBaseRef ?? (await resolveBaseRef(cwd)), + resolvedBaseRef: storedBaseRef ?? (await resolveBaseRef(cwd, context)), }; } @@ -1253,7 +1270,7 @@ function resolveOperationBaseRef(input: { } async function isWorkingTreeDirty(cwd: string, context?: CheckoutContext): Promise { - const { stdout } = await runGitCommand(["status", "--porcelain"], { + const { stdout } = await getRunGitCommand(context)(["status", "--porcelain"], { cwd, envOverlay: READ_ONLY_GIT_ENV, logger: context?.logger, @@ -1261,9 +1278,12 @@ async function isWorkingTreeDirty(cwd: string, context?: CheckoutContext): Promi return stdout.trim().length > 0; } -export async function getOriginRemoteUrl(cwd: string): Promise { +export async function getOriginRemoteUrl( + cwd: string, + context?: CheckoutContext, +): Promise { try { - const { stdout } = await runGitCommand(["config", "--get", "remote.origin.url"], { + const { stdout } = await getRunGitCommand(context)(["config", "--get", "remote.origin.url"], { cwd, envOverlay: READ_ONLY_GIT_ENV, }); @@ -1285,7 +1305,7 @@ async function getGitConfigValue( context?: CheckoutContext, ): Promise { try { - const { stdout } = await runGitCommand(["config", "--get", key], { + const { stdout } = await getRunGitCommand(context)(["config", "--get", key], { cwd, envOverlay: READ_ONLY_GIT_ENV, logger: context?.logger, @@ -1303,11 +1323,14 @@ async function getGitRemotePushUrl( context?: CheckoutContext, ): Promise { try { - const { stdout } = await runGitCommand(["remote", "get-url", "--push", remoteName], { - cwd, - envOverlay: READ_ONLY_GIT_ENV, - logger: context?.logger, - }); + const { stdout } = await getRunGitCommand(context)( + ["remote", "get-url", "--push", remoteName], + { + cwd, + envOverlay: READ_ONLY_GIT_ENV, + logger: context?.logger, + }, + ); const value = stdout.trim(); return value.length > 0 ? value : null; } catch { @@ -1383,9 +1406,12 @@ async function resolvePullRequestStatusLookupTarget( return pushTarget ?? branchTarget; } -export async function resolveAbsoluteGitDir(cwd: string): Promise { +export async function resolveAbsoluteGitDir( + cwd: string, + context?: CheckoutContext, +): Promise { try { - const { stdout } = await runGitCommand(["rev-parse", "--absolute-git-dir"], { + const { stdout } = await getRunGitCommand(context)(["rev-parse", "--absolute-git-dir"], { cwd, envOverlay: READ_ONLY_GIT_ENV, }); @@ -1396,9 +1422,9 @@ export async function resolveAbsoluteGitDir(cwd: string): Promise } } -async function resolveGitCommonDir(cwd: string): Promise { +async function resolveGitCommonDir(cwd: string, context?: CheckoutContext): Promise { try { - const { stdout } = await runGitCommand(["rev-parse", "--git-common-dir"], { + const { stdout } = await getRunGitCommand(context)(["rev-parse", "--git-common-dir"], { cwd, envOverlay: READ_ONLY_GIT_ENV, }); @@ -1435,9 +1461,12 @@ async function abortGitPullConflictState(cwd: string): Promise { } } -export async function resolveRepositoryDefaultBranch(repoRoot: string): Promise { +export async function resolveRepositoryDefaultBranch( + repoRoot: string, + context?: CheckoutContext, +): Promise { try { - const { stdout } = await runGitCommand( + const { stdout } = await getRunGitCommand(context)( ["symbolic-ref", "--quiet", "refs/remotes/origin/HEAD"], { cwd: repoRoot, @@ -1453,10 +1482,13 @@ export async function resolveRepositoryDefaultBranch(repoRoot: string): Promise< ? remoteShort.slice("origin/".length) : remoteShort; try { - await runGitCommand(["show-ref", "--verify", "--quiet", `refs/heads/${localName}`], { - cwd: repoRoot, - envOverlay: READ_ONLY_GIT_ENV, - }); + await getRunGitCommand(context)( + ["show-ref", "--verify", "--quiet", `refs/heads/${localName}`], + { + cwd: repoRoot, + envOverlay: READ_ONLY_GIT_ENV, + }, + ); return localName; } catch { return remoteShort; @@ -1466,7 +1498,7 @@ export async function resolveRepositoryDefaultBranch(repoRoot: string): Promise< // ignore } - const { stdout } = await runGitCommand(["branch", "--format=%(refname:short)"], { + const { stdout } = await getRunGitCommand(context)(["branch", "--format=%(refname:short)"], { cwd: repoRoot, envOverlay: READ_ONLY_GIT_ENV, }); @@ -1487,8 +1519,8 @@ export async function resolveRepositoryDefaultBranch(repoRoot: string): Promise< return null; } -async function resolveBaseRef(repoRoot: string): Promise { - return resolveRepositoryDefaultBranch(repoRoot); +async function resolveBaseRef(repoRoot: string, context?: CheckoutContext): Promise { + return resolveRepositoryDefaultBranch(repoRoot, context); } interface ComparisonBaseRefName { @@ -1515,7 +1547,7 @@ async function doesGitRefExist( fullRef: string, context?: CheckoutContext, ): Promise { - const result = await runGitCommand(["show-ref", "--verify", "--quiet", fullRef], { + const result = await getRunGitCommand(context)(["show-ref", "--verify", "--quiet", fullRef], { cwd, envOverlay: READ_ONLY_GIT_ENV, acceptExitCodes: [0, 1], @@ -1632,7 +1664,7 @@ async function getAheadBehindForComparisonRef( currentBranch: string, context?: CheckoutContext, ): Promise { - const { stdout } = await runGitCommand( + const { stdout } = await getRunGitCommand(context)( ["rev-list", "--left-right", "--count", `${comparisonRef}...${currentBranch}`], { cwd, envOverlay: READ_ONLY_GIT_ENV, logger: context?.logger }, ); @@ -1651,7 +1683,7 @@ async function getUpstreamStatus( context?: CheckoutContext, ): Promise { try { - const { stdout } = await runGitCommand( + const { stdout } = await getRunGitCommand(context)( [ "for-each-ref", "--format=%(upstream)%00%(upstream:track,nobracket)", @@ -1690,10 +1722,10 @@ async function inspectCheckoutContext( } const [currentBranch, remoteUrl, absoluteGitDir, gitCommonDir] = await Promise.all([ - getCurrentBranch(cwd), - getOriginRemoteUrl(cwd), - resolveAbsoluteGitDir(cwd), - resolveGitCommonDir(cwd), + getCurrentBranch(cwd, context), + getOriginRemoteUrl(cwd, context), + resolveAbsoluteGitDir(cwd, context), + resolveGitCommonDir(cwd, context), ]); const paseoWorktree = await getPaseoWorktreeForCwd(cwd, { context, @@ -1940,7 +1972,7 @@ export async function getCheckoutSnapshotFacts( ? readPaseoWorktreeMetadata(inspected.paseoWorktree.worktreeRoot) : null; const storedBaseRef = storedBaseRefFromMetadata(paseoWorktreeMetadata); - const resolvedBaseRef = storedBaseRef ?? (await resolveBaseRef(cwd)); + const resolvedBaseRef = storedBaseRef ?? (await resolveBaseRef(cwd, context)); const mainRepoRoot = await getMainRepoRootFromCommonDir( cwd, inspected.gitCommonDir, @@ -2593,12 +2625,19 @@ function parseCheckoutShortstat(text: string): CheckoutShortstat | null { const UNTRACKED_SHORTSTAT_MAX_FILES = 500; -async function countUntrackedAdditions(cwd: string, throwOnGitError = false): Promise { +async function countUntrackedAdditions( + cwd: string, + context?: CheckoutContext, + throwOnGitError = false, +): Promise { try { - const { stdout } = await runGitCommand(["ls-files", "--others", "--exclude-standard"], { - cwd, - envOverlay: READ_ONLY_GIT_ENV, - }); + const { stdout } = await getRunGitCommand(context)( + ["ls-files", "--others", "--exclude-standard"], + { + cwd, + envOverlay: READ_ONLY_GIT_ENV, + }, + ); const files = stdout .split("\n") .map((l) => l.trim()) @@ -2646,7 +2685,7 @@ async function getCheckoutShortstatUncached( } if (!context?.facts?.isGit) { try { - await requireGitRepo(cwd); + await requireGitRepo(cwd, context); } catch { return null; } @@ -2656,33 +2695,37 @@ async function getCheckoutShortstatUncached( const localBaseRef = facts?.isGit ? facts.resolvedBaseRef : await getResolvedBaseRefForCwd(cwd, context); - const currentBranch = facts?.isGit ? facts.currentBranch : await getCurrentBranch(cwd); + const currentBranch = facts?.isGit ? facts.currentBranch : await getCurrentBranch(cwd, context); const comparisonRef = await resolveShortstatComparisonRef({ cwd, currentBranch, localBaseRef, facts, + context, }); if (!comparisonRef) { return null; } try { - const { stdout: mergeBaseOut } = await runGitCommand(["merge-base", "HEAD", comparisonRef], { - cwd, - envOverlay: READ_ONLY_GIT_ENV, - }); + const { stdout: mergeBaseOut } = await getRunGitCommand(context)( + ["merge-base", "HEAD", comparisonRef], + { + cwd, + envOverlay: READ_ONLY_GIT_ENV, + }, + ); const mergeBase = mergeBaseOut.trim(); if (!mergeBase) { return null; } const [{ stdout }, untrackedAdditions] = await Promise.all([ - runGitCommand(["diff", "--shortstat", mergeBase], { + getRunGitCommand(context)(["diff", "--shortstat", mergeBase], { cwd, envOverlay: READ_ONLY_GIT_ENV, }), - countUntrackedAdditions(cwd, options?.throwOnGitError), + countUntrackedAdditions(cwd, context, options?.throwOnGitError), ]); const tracked = parseCheckoutShortstat(stdout); @@ -2704,8 +2747,9 @@ async function resolveShortstatComparisonRef(input: { currentBranch: string | null; localBaseRef: string | null; facts?: CheckoutSnapshotFacts | null; + context?: CheckoutContext; }): Promise { - const { cwd, currentBranch, localBaseRef, facts } = input; + const { cwd, currentBranch, localBaseRef, facts, context } = input; if (!currentBranch) { return null; } @@ -2714,13 +2758,13 @@ async function resolveShortstatComparisonRef(input: { try { return facts?.isGit && facts.resolvedBaseRef === localBaseRef && facts.comparisonBaseRef ? facts.comparisonBaseRef - : await resolveBestComparisonBaseRef(cwd, localBaseRef); + : await resolveBestComparisonBaseRef(cwd, localBaseRef, context); } catch { return null; } } - const hasOrigin = await doesGitRefExist(cwd, `refs/remotes/origin/${currentBranch}`); + const hasOrigin = await doesGitRefExist(cwd, `refs/remotes/origin/${currentBranch}`, context); return hasOrigin ? `origin/${currentBranch}` : null; } @@ -3976,9 +4020,11 @@ async function getPullRequestStatusUncached( const unavailable = getUnavailablePullRequestStatus(context?.facts); if (unavailable) return unavailable; if (!context?.facts?.isGit) { - await requireGitRepo(cwd); + await requireGitRepo(cwd, context); } - const head = context?.facts?.isGit ? context.facts.currentBranch : await getCurrentBranch(cwd); + const head = context?.facts?.isGit + ? context.facts.currentBranch + : await getCurrentBranch(cwd, context); if (!head) { return buildPullRequestStatusResult(null, "no_remote"); } diff --git a/packages/server/src/utils/git-command-runtime-metrics.test.ts b/packages/server/src/utils/git-command-runtime-metrics.test.ts index 7f882ceb77..c0ad7b8d9b 100644 --- a/packages/server/src/utils/git-command-runtime-metrics.test.ts +++ b/packages/server/src/utils/git-command-runtime-metrics.test.ts @@ -37,9 +37,9 @@ describe("GitCommandRuntimeMetricsWindow", () => { test("reports live queue pressure across window resets", () => { const { metrics, advance } = createMetricsWindow(1); - const active = metrics.submit("fetch"); + const active = metrics.submit("fetch", "background-fetch"); metrics.start(active); - const pending = metrics.submit("rev-parse"); + const pending = metrics.submit("rev-parse", "workspace-refresh:watch"); metrics.observeLimiter(1, 1); advance(25); @@ -53,6 +53,12 @@ describe("GitCommandRuntimeMetricsWindow", () => { oldestPendingMs: 25, submitted: 2, started: 1, + provenanceTop: [ + ["background-fetch", 1], + ["workspace-refresh:watch", 1], + ], + pendingProvenanceTop: [["workspace-refresh:watch", 1]], + activeProvenanceTop: [["background-fetch", 1]], }); advance(10); @@ -73,6 +79,9 @@ describe("GitCommandRuntimeMetricsWindow", () => { failed: 1, timedOut: 1, queueWaitMs: { count: 1, p50Ms: 35, p95Ms: 35, maxMs: 35 }, + provenanceTop: [], + pendingProvenanceTop: [], + activeProvenanceTop: [], }); }); }); diff --git a/packages/server/src/utils/git-command-runtime-metrics.ts b/packages/server/src/utils/git-command-runtime-metrics.ts index debced34bb..a8f32cdccc 100644 --- a/packages/server/src/utils/git-command-runtime-metrics.ts +++ b/packages/server/src/utils/git-command-runtime-metrics.ts @@ -21,17 +21,22 @@ export interface GitCommandRuntimeMetricsSnapshot { queueWaitMs: GitCommandDurationStats; executionMs: GitCommandDurationStats; operationsTop: Array<[string, number]>; + provenanceTop: Array<[string, number]>; + pendingProvenanceTop: Array<[string, number]>; + activeProvenanceTop: Array<[string, number]>; } interface GitCommandRuntimeMetric { queuedAtMs: number; startedAtMs: number | null; + provenance: string; } type Clock = () => number; export class GitCommandRuntimeMetricsWindow { private readonly pendingCommands = new Set(); + private readonly activeCommands = new Set(); private active = 0; private peakActive = 0; private peakPending = 0; @@ -43,6 +48,7 @@ export class GitCommandRuntimeMetricsWindow { private readonly queueWaitSamples: number[] = []; private readonly executionSamples: number[] = []; private readonly operationCounts = new Map(); + private readonly provenanceCounts = new Map(); constructor( private readonly limits: { @@ -52,11 +58,12 @@ export class GitCommandRuntimeMetricsWindow { private readonly clock: Clock = Date.now, ) {} - submit(operation: string): GitCommandRuntimeMetric { - const metric = { queuedAtMs: this.clock(), startedAtMs: null }; + submit(operation: string, provenance = "unattributed"): GitCommandRuntimeMetric { + const metric = { queuedAtMs: this.clock(), startedAtMs: null, provenance }; this.pendingCommands.add(metric); this.submittedCount += 1; this.operationCounts.set(operation, (this.operationCounts.get(operation) ?? 0) + 1); + this.provenanceCounts.set(provenance, (this.provenanceCounts.get(provenance) ?? 0) + 1); return metric; } @@ -71,6 +78,7 @@ export class GitCommandRuntimeMetricsWindow { } const now = this.clock(); metric.startedAtMs = now; + this.activeCommands.add(metric); this.active += 1; this.startedCount += 1; this.peakActive = Math.max(this.peakActive, this.active); @@ -83,6 +91,7 @@ export class GitCommandRuntimeMetricsWindow { } const startedAtMs = metric.startedAtMs; metric.startedAtMs = null; + this.activeCommands.delete(metric); this.active = Math.max(0, this.active - 1); this.completedCount += 1; if (!outcome.success) { @@ -122,6 +131,9 @@ export class GitCommandRuntimeMetricsWindow { operationsTop: [...this.operationCounts.entries()] .sort((left, right) => right[1] - left[1] || left[0].localeCompare(right[0])) .slice(0, 12), + provenanceTop: topCounts(this.provenanceCounts), + pendingProvenanceTop: topProvenance(this.pendingCommands), + activeProvenanceTop: topProvenance(this.activeCommands), }; this.peakActive = limiter.active; @@ -134,10 +146,25 @@ export class GitCommandRuntimeMetricsWindow { this.queueWaitSamples.length = 0; this.executionSamples.length = 0; this.operationCounts.clear(); + this.provenanceCounts.clear(); return snapshot; } } +function topProvenance(commands: ReadonlySet): Array<[string, number]> { + const counts = new Map(); + for (const command of commands) { + counts.set(command.provenance, (counts.get(command.provenance) ?? 0) + 1); + } + return topCounts(counts); +} + +function topCounts(counts: ReadonlyMap): Array<[string, number]> { + return [...counts.entries()] + .sort((left, right) => right[1] - left[1] || left[0].localeCompare(right[0])) + .slice(0, 12); +} + function summarizeDurations(samples: number[]): GitCommandDurationStats { if (samples.length === 0) { return { count: 0, p50Ms: 0, p95Ms: 0, maxMs: 0 }; diff --git a/packages/server/src/utils/git-process-scheduler.test.ts b/packages/server/src/utils/git-process-scheduler.test.ts index 4c879267b9..c8a72cd199 100644 --- a/packages/server/src/utils/git-process-scheduler.test.ts +++ b/packages/server/src/utils/git-process-scheduler.test.ts @@ -1,3 +1,4 @@ +import { setImmediate as scheduleImmediate } from "node:timers"; import { afterEach, describe, expect, test, vi } from "vitest"; import { GitProcessScheduler, resolveGitProcessPolicy } from "./git-process-scheduler.js"; @@ -55,18 +56,76 @@ describe("GitProcessScheduler", () => { const started: number[] = []; const activeObservation = scheduler.run(createConcurrencyTask(0, started, releases)); + await vi.waitFor(() => expect(started).toEqual([0])); const queuedObservation = scheduler.run(createConcurrencyTask(2, started, releases)); const userOperation = scheduler.run(createConcurrencyTask(3, started, releases), { priority: "high", }); - await vi.waitFor(() => expect(started).toEqual([0])); releases.shift()?.(); await Promise.all([activeObservation, queuedObservation, userOperation]); expect(started).toEqual([0, 3, 2]); }); + test("yields to daemon liveness work between process starts", async () => { + const scheduler = new GitProcessScheduler({ + maxProcessesPerSecond: 100, + maxProcessConcurrency: 8, + }); + const started: number[] = []; + const start = () => { + started.push(started.length); + const completed = Promise.resolve(); + return { result: completed, exited: completed }; + }; + + const livenessTurn = new Promise((resolve) => setImmediate(resolve)); + const tasks = Array.from({ length: 8 }, () => scheduler.run(start)); + await livenessTurn; + + expect(started).toHaveLength(1); + await Promise.all(tasks); + }); + + test("starts an isolated process without an extra event-loop turn", async () => { + const scheduler = new GitProcessScheduler({ + maxProcessesPerSecond: 100, + maxProcessConcurrency: 8, + }); + let started = false; + + const result = scheduler.run(() => { + started = true; + const completed = Promise.resolve(); + return { result: completed, exited: completed }; + }); + + expect(started).toBe(true); + await result; + }); + + test("queued processes keep moving while application timers are faked", async () => { + vi.useFakeTimers(); + const scheduler = new GitProcessScheduler({ + maxProcessesPerSecond: 100, + maxProcessConcurrency: 1, + }); + const releases: Array<() => void> = []; + const started: number[] = []; + + const first = scheduler.run(createConcurrencyTask(0, started, releases)); + const second = scheduler.run(createConcurrencyTask(2, started, releases)); + expect(started).toEqual([0]); + + releases.shift()?.(); + await Promise.resolve(); + await new Promise((resolve) => scheduleImmediate(resolve)); + + expect(started).toEqual([0, 2]); + await Promise.all([first, second]); + }); + test("limits process starts in each strict one-second interval", async () => { vi.useFakeTimers(); const scheduler = new GitProcessScheduler({ @@ -81,13 +140,14 @@ describe("GitProcessScheduler", () => { }; const tasks = Array.from({ length: 4 }, () => scheduler.run(recordStart)); - await vi.advanceTimersByTimeAsync(0); - expect(startedAt).toHaveLength(2); await vi.advanceTimersByTimeAsync(999); - expect(startedAt).toHaveLength(2); - await vi.advanceTimersByTimeAsync(1); - expect(startedAt).toHaveLength(3); - await vi.advanceTimersByTimeAsync(500); + expect(startedAt.length).toBeLessThanOrEqual(2); + while (startedAt.length < 3) { + await vi.advanceTimersToNextTimerAsync(); + } + while (startedAt.length < 4) { + await vi.advanceTimersToNextTimerAsync(); + } await Promise.all(tasks); expect(startedAt).toHaveLength(4); diff --git a/packages/server/src/utils/git-process-scheduler.ts b/packages/server/src/utils/git-process-scheduler.ts index 366029c366..cd5575d79e 100644 --- a/packages/server/src/utils/git-process-scheduler.ts +++ b/packages/server/src/utils/git-process-scheduler.ts @@ -1,3 +1,4 @@ +import { setImmediate as scheduleImmediate } from "node:timers"; import pThrottle from "p-throttle"; export interface GitProcessPolicy { @@ -50,6 +51,7 @@ export class GitProcessScheduler { private admitted = 0; private running = 0; private waiting = 0; + private drainScheduled = false; constructor(readonly policy: GitProcessPolicy) { const throttle = pThrottle({ @@ -79,25 +81,45 @@ export class GitProcessScheduler { this.waiting = Math.max(0, this.waiting - 1); this.admitted = Math.max(0, this.admitted - 1); reject(error); - this.drain(); + this.scheduleDrain(); }); }; const queue = options?.priority === "high" ? this.highPriorityQueue : this.normalPriorityQueue; queue.push(admit); - this.drain(); + if (this.admitted === 0 && !this.drainScheduled) { + this.drainOne(); + } else { + this.scheduleDrain(); + } }); } - private drain(): void { - while (this.admitted < this.policy.maxProcessConcurrency) { - const admit = this.highPriorityQueue.shift() ?? this.normalPriorityQueue.shift(); - if (!admit) { - return; - } - this.admitted += 1; - admit(); + private scheduleDrain(): void { + if (this.drainScheduled || this.admitted >= this.policy.maxProcessConcurrency) { + return; } + if (this.highPriorityQueue.length === 0 && this.normalPriorityQueue.length === 0) { + return; + } + this.drainScheduled = true; + scheduleImmediate(() => this.drainOne()); + } + + private drainOne(): void { + this.drainScheduled = false; + if (this.admitted >= this.policy.maxProcessConcurrency) { + return; + } + const admit = this.highPriorityQueue.shift() ?? this.normalPriorityQueue.shift(); + if (!admit) { + return; + } + this.admitted += 1; + admit(); + // Spawning can be synchronous and expensive on a throttled host. Let timers, + // sockets, and child exits run before admitting another Git process. + this.scheduleDrain(); } private async execute( @@ -116,7 +138,7 @@ export class GitProcessScheduler { } finally { this.running = Math.max(0, this.running - 1); this.admitted = Math.max(0, this.admitted - 1); - this.drain(); + this.scheduleDrain(); } } } diff --git a/packages/server/src/utils/run-git-command.test.ts b/packages/server/src/utils/run-git-command.test.ts index 1c33aff9d9..32844d0de8 100644 --- a/packages/server/src/utils/run-git-command.test.ts +++ b/packages/server/src/utils/run-git-command.test.ts @@ -409,6 +409,20 @@ describe("runGitCommand", () => { ]); }); + it("binds queue provenance to a Git command runner", async () => { + const { createRunGitCommand, snapshotGitCommandRuntimeMetrics } = await loadRunGitCommand(1); + enqueueSpawnBehaviors({ stdoutData: "ok" }); + + const runWorkspaceRefreshGitCommand = createRunGitCommand("workspace-refresh:watch"); + await runWorkspaceRefreshGitCommand(["status"], { cwd: process.cwd() }); + + expect(snapshotGitCommandRuntimeMetrics()).toMatchObject({ + provenanceTop: [["workspace-refresh:watch", 1]], + pendingProvenanceTop: [], + activeProvenanceTop: [], + }); + }); + it("resolves truncated stdout, caps output, and kills the child process", async () => { const { runGitCommand } = await loadRunGitCommand(1); diff --git a/packages/server/src/utils/run-git-command.ts b/packages/server/src/utils/run-git-command.ts index 71c9e95f60..fb8cc32c76 100644 --- a/packages/server/src/utils/run-git-command.ts +++ b/packages/server/src/utils/run-git-command.ts @@ -62,6 +62,15 @@ export interface GitCommandResult { signal: NodeJS.Signals | null; } +export type RunGitCommand = ( + args: string[], + options: GitCommandOptions, +) => Promise; + +export function createRunGitCommand(provenance: string): RunGitCommand { + return (args, options) => runGitCommandWithProvenance(args, options, provenance); +} + export interface GitCommandMetric { args: string[]; cwd: string; @@ -249,16 +258,17 @@ function getEnvOverlayKeys(envOverlay: ProcessEnvRecord | undefined): string[] { return Object.keys(envOverlay ?? {}).sort(); } -export function runGitCommand( +function runGitCommandWithProvenance( args: string[], options: GitCommandOptions, + provenance?: string, ): Promise { const metricsState = submitGitCommandMetric(args, options.cwd); const commandTrace = submitGitCommandTrace(args, options.cwd, { active: gitProcessScheduler.activeCount, pending: gitProcessScheduler.pendingCount, }); - const runtimeMetric = gitRuntimeMetrics.submit(getGitOperation(args)); + const runtimeMetric = gitRuntimeMetrics.submit(getGitOperation(args), provenance); const startCommand = () => { let releaseProcessSlot!: () => void; const exited = new Promise((resolve) => { @@ -526,6 +536,8 @@ export function runGitCommand( return promise; } +export const runGitCommand: RunGitCommand = runGitCommandWithProvenance; + function formatGitCommand(args: string[]): string { return ["git", ...args].join(" "); }