jb/paseo
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Limit daemon-wide Git process pressure (#2797)

Browse Source 
* perf(server): bound Git process starts

Git refresh bursts could keep creating short-lived processes at the concurrency ceiling, causing sustained process churn on constrained hosts. Apply one daemon-global start-rate and concurrency policy to every Git command while preserving the legacy concurrency environment variable.

* build(nix): update npm dependency hash

* fix(server): hold Git slots until process exit

* fix(server): enforce Git limits across process lifetime

* fix(server): align Git metrics with process exit

* Fix Git limiter test deadlocks

* Tune default Git process rate
This commit is contained in:
Mohamed Boudra
2026-08-02 19:59:43 +02:00
committed by GitHub
parent f682156302
commit 2e188787ff
23 changed files with 822 additions and 225 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/bootstrap.ts
+7
View File
@@ -198,6 +198,8 @@ import { createWebUiMiddleware } from "./web-ui.js";
import { WorkspaceAutoName } from "./workspace-auto-name.js";
import { createGitMutationService } from "./session/git-mutation/git-mutation-service.js";
import { workspaceIdsOnCheckout } from "./workspace-directory.js";
import { configureGitProcessPolicy } from "../utils/run-git-command.js";
import { resolveGitProcessPolicy } from "../utils/git-process-scheduler.js";
import { resolveFirstAgentPromptTitle } from "./agent/create-agent-title.js";
import {
createAgentCommand,
@@ -388,6 +390,10 @@ export interface PaseoDaemonConfig {
mcpEnabled?: boolean;
mcpInjectIntoAgents?: boolean;
browserToolsEnabled?: boolean;
git?: {
maxProcessesPerSecond: number;
maxProcessConcurrency: number;
};
autoArchiveAfterMerge?: boolean;
enableTerminalAgentHooks?: boolean;
appendSystemPrompt?: string;
@@ -539,6 +545,7 @@ export async function createPaseoDaemon(
rootLogger: Logger,
dependencies: PaseoDaemonDependencies = {},
): Promise<PaseoDaemon> {
configureGitProcessPolicy(config.git ?? resolveGitProcessPolicy({ env: process.env }));
const logger = rootLogger.child({ module: "bootstrap" });
const bootstrapStart = performance.now();
const elapsed = () => `${(performance.now() - bootstrapStart).toFixed(0)}ms`;
packages/server/src/server/config-git.test.ts
+89
View File
@@ -0,0 +1,89 @@
import { mkdtemp, rm, writeFile } from "node:fs/promises";
import os from "node:os";
import path from "node:path";
import { afterEach, describe, expect, test } from "vitest";
import { loadConfig } from "./config.js";
const roots: string[] = [];
async function createHome(config: object = {}): Promise<string> {
const home = await mkdtemp(path.join(os.tmpdir(), "paseo-config-git-"));
roots.push(home);
await writeFile(path.join(home, "config.json"), JSON.stringify(config));
return home;
}
describe("daemon Git process config", () => {
afterEach(async () => {
await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true })));
});
test("defaults the global process limits", async () => {
const home = await createHome();
expect(loadConfig(home, { env: {} }).git).toEqual({
maxProcessesPerSecond: 64,
maxProcessConcurrency: 8,
});
});
test("loads both limits from daemon.git", async () => {
const home = await createHome({
daemon: {
git: {
maxProcessesPerSecond: 5,
maxProcessConcurrency: 4,
},
},
});
expect(loadConfig(home, { env: {} }).git).toEqual({
maxProcessesPerSecond: 5,
maxProcessConcurrency: 4,
});
});
test("new environment variables override config.json", async () => {
const home = await createHome({
daemon: {
git: {
maxProcessesPerSecond: 5,
maxProcessConcurrency: 4,
},
},
});
expect(
loadConfig(home, {
env: {
PASEO_GIT_MAX_PROCESSES_PER_SECOND: "12",
PASEO_GIT_MAX_PROCESS_CONCURRENCY: "6",
},
}).git,
).toEqual({
maxProcessesPerSecond: 12,
maxProcessConcurrency: 6,
});
});
test("accepts legacy PASEO_GIT_CONCURRENCY below the renamed variable", async () => {
const home = await createHome();
expect(
loadConfig(home, {
env: {
PASEO_GIT_CONCURRENCY: "3",
},
}).git?.maxProcessConcurrency,
).toBe(3);
expect(
loadConfig(home, {
env: {
PASEO_GIT_CONCURRENCY: "3",
PASEO_GIT_MAX_PROCESS_CONCURRENCY: "7",
},
}).git?.maxProcessConcurrency,
).toBe(7);
});
});
packages/server/src/server/config.ts
+12
View File
@@ -22,6 +22,7 @@ import { AgentProviderSchema } from "@getpaseo/protocol/provider-manifest";
import { hashDaemonPassword } from "./auth.js";
import { resolveSpeechConfig } from "./speech/speech-config-resolver.js";
import { mergeHostnames, parseHostnamesEnv, type HostnamesConfig } from "./hostnames.js";
import { resolveGitProcessPolicy } from "../utils/git-process-scheduler.js";
const DEFAULT_PORT = 6767;
const DEFAULT_RELAY_ENDPOINT = "relay.paseo.sh:443";
@@ -87,6 +88,16 @@ function normalizeLogEnv(value: string | undefined): string | undefined {
return value.trim().toLowerCase();
}
function resolveGitProcessConfig(
env: NodeJS.ProcessEnv,
persisted: ReturnType<typeof loadPersistedConfig>,
): NonNullable<PaseoDaemonConfig["git"]> {
return resolveGitProcessPolicy({
env,
persisted: persisted.daemon?.git,
});
}
export type CliConfigOverrides = Partial<{
listen: string;
relayEnabled: boolean;
@@ -504,6 +515,7 @@ export function loadConfig(
mcpEnabled,
mcpInjectIntoAgents,
browserToolsEnabled,
git: resolveGitProcessConfig(env, persisted),
autoArchiveAfterMerge,
enableTerminalAgentHooks: persisted.daemon?.enableTerminalAgentHooks ?? false,
appendSystemPrompt,
packages/server/src/server/persisted-config.ts
+7
View File
@@ -248,6 +248,13 @@ export const PersistedConfigSchema = z
})
.passthrough()
.optional(),
git: z
.object({
maxProcessesPerSecond: z.number().int().positive().optional(),
maxProcessConcurrency: z.number().int().positive().optional(),
})
.strict()
.optional(),
autoArchiveAfterMerge: z.boolean().optional(),
enableTerminalAgentHooks: z.boolean().optional(),
appendSystemPrompt: z.string().optional(),
packages/server/src/server/session.test.ts
+17 -6
View File
@@ -201,10 +201,13 @@ const agentResponseMocks = vi.hoisted(() => ({
}));
const spawnMocks = vi.hoisted(() => ({
execCommand: vi.fn(),
spawnWorkspaceScript: vi.fn(),
}));
const gitCommandMocks = vi.hoisted(() => ({
runGitCommand: vi.fn(),
}));
const paseoWorktreeServiceMocks = vi.hoisted(() => ({
createPaseoWorktree: vi.fn(),
}));
@@ -253,11 +256,11 @@ vi.mock("./paseo-worktree-service.js", async (importOriginal) => {
};
});
vi.mock("../utils/spawn.js", async (importOriginal) => {
const actual = await importOriginal<typeof import("../utils/spawn.js")>();
vi.mock("../utils/run-git-command.js", async (importOriginal) => {
const actual = await importOriginal<typeof import("../utils/run-git-command.js")>();
return {
...actual,
execCommand: spawnMocks.execCommand,
runGitCommand: gitCommandMocks.runGitCommand,
};
});
@@ -4045,7 +4048,7 @@ describe("session stash mutation handling", () => {
const messages: unknown[] = [];
const workspaceGitService = { getSnapshot: vi.fn().mockResolvedValue({}) };
const session = createSessionForTest({ workspaceGitService, messages });
spawnMocks.execCommand.mockResolvedValue({
gitCommandMocks.runGitCommand.mockResolvedValue({
stdout: "",
stderr: "",
exitCode: 0,
@@ -4060,6 +4063,10 @@ describe("session stash mutation handling", () => {
requestId: "request-stash-push",
});
expect(gitCommandMocks.runGitCommand).toHaveBeenCalledWith(
["stash", "push", "--include-untracked", "-m", "paseo-auto-stash: feature"],
{ cwd: "/tmp/repo", timeout: 120_000 },
);
expect(workspaceGitService.getSnapshot).toHaveBeenCalledWith("/tmp/repo", {
force: true,
reason: "stash-push",
@@ -4079,7 +4086,7 @@ describe("session stash mutation handling", () => {
const messages: unknown[] = [];
const workspaceGitService = { getSnapshot: vi.fn().mockResolvedValue({}) };
const session = createSessionForTest({ workspaceGitService, messages });
spawnMocks.execCommand.mockResolvedValue({
gitCommandMocks.runGitCommand.mockResolvedValue({
stdout: "",
stderr: "",
exitCode: 0,
@@ -4094,6 +4101,10 @@ describe("session stash mutation handling", () => {
requestId: "request-stash-pop",
});
expect(gitCommandMocks.runGitCommand).toHaveBeenCalledWith(["stash", "pop", "stash@{0}"], {
cwd: "/tmp/repo",
timeout: 120_000,
});
expect(workspaceGitService.getSnapshot).toHaveBeenCalledWith("/tmp/repo", {
force: true,
reason: "stash-pop",
packages/server/src/server/session/checkout/checkout-session.ts
+5 -3
View File
@@ -51,7 +51,7 @@ import {
listCheckoutCommits,
getCommitFileDiff,
} from "../../../utils/checkout-git.js";
import { execCommand } from "../../../utils/spawn.js";
import { runGitCommand } from "../../../utils/run-git-command.js";
import { expandTilde } from "../../../utils/path.js";
import type { GitMetadataGenerator } from "./git-metadata-generator.js";
@@ -614,8 +614,9 @@ export class CheckoutSession {
const message = branchLabel
? `${CheckoutSession.PASEO_STASH_PREFIX} ${branchLabel}`
: `${CheckoutSession.PASEO_STASH_PREFIX} unnamed`;
await execCommand("git", ["stash", "push", "--include-untracked", "-m", message], {
await runGitCommand(["stash", "push", "--include-untracked", "-m", message], {
cwd,
timeout: 120_000,
});
await this.gitMutation.notifyGitMutation(cwd, "stash-push");
this.scheduleDiffRefresh(cwd);
@@ -636,8 +637,9 @@ export class CheckoutSession {
): Promise<void> {
const { cwd, stashIndex, requestId } = msg;
try {
await execCommand("git", ["stash", "pop", `stash@{${stashIndex}}`], {
await runGitCommand(["stash", "pop", `stash@{${stashIndex}}`], {
cwd,
timeout: 120_000,
});
await this.gitMutation.notifyGitMutation(cwd, "stash-pop");
this.scheduleDiffRefresh(cwd);
packages/server/src/server/session/git-mutation/git-mutation-service.ts
+5 -2
View File
@@ -5,7 +5,7 @@ import {
type CheckoutExistingBranchResult,
type GitMutationRefreshReason,
} from "../../../utils/checkout-git.js";
import { execCommand } from "../../../utils/spawn.js";
import { runGitCommand } from "../../../utils/run-git-command.js";
import type { WorkspaceGitService } from "../../workspace-git-service.js";
import { assertSafeGitRef as assertWorktreeSafeGitRef } from "../../worktree-session.js";
@@ -118,7 +118,10 @@ export function createGitMutationService(deps: {
}
await ensureCleanWorkingTree(cwd);
await execCommand("git", ["checkout", "-b", newBranchName, baseBranch], { cwd });
await runGitCommand(["checkout", "-b", newBranchName, baseBranch], {
cwd,
timeout: 120_000,
});
await notifyGitMutation(cwd, "create-branch");
},
packages/server/src/server/workspace-git-service.primitive.test.ts
+8 -4
View File
@@ -1153,12 +1153,13 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => {
test("settled GitHub self-heal reads stay on the slow poll window without refreshing git", async () => {
let nowMs = 0;
const githubReadCalls: Array<{ reason: string | undefined; tickMs: number }> = [];
const runner = vi.fn(async () => ({
stdout: currentPullRequestJson(),
stderr: "",
}));
const github = createGitHubService({
ttlMs: 0,
runner: vi.fn(async () => ({
stdout: currentPullRequestJson(),
stderr: "",
})),
runner,
resolveGhPath: async () => "/usr/bin/gh",
now: () => nowMs,
});
@@ -1180,6 +1181,9 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => {
const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn());
await flushPromises();
await vi.advanceTimersByTimeAsync(0);
await vi.waitFor(() => {
expect(runner).toHaveBeenCalledTimes(1);
});
await flushPromises();
const gitReadsAfterInitialSnapshot = getCheckoutStatus.mock.calls.length;
packages/server/src/services/github-service.test.ts
+2
View File
@@ -820,6 +820,7 @@ describe("ForgeService", () => {
reason: "self-heal-github",
}),
]);
await vi.advanceTimersByTimeAsync(0);
expect(currentPullRequestStatusCalls(runner.calls)).toHaveLength(1);
subscription?.unsubscribe();
@@ -856,6 +857,7 @@ describe("ForgeService", () => {
now = EXPECTED_GITHUB_SLOW_POLL_MS;
await vi.advanceTimersByTimeAsync(EXPECTED_GITHUB_SLOW_POLL_MS - EXPECTED_GITHUB_FAST_POLL_MS);
await vi.advanceTimersByTimeAsync(0);
expect(currentPullRequestStatusCalls(runner.calls)).toHaveLength(2);
expect(reads.map((read) => read.reason)).toEqual([undefined, "self-heal-github"]);
packages/server/src/utils/git-command-runtime-metrics.test.ts
+6 -2
View File
@@ -1,10 +1,13 @@
import { describe, expect, test } from "vitest";
import { GitCommandRuntimeMetricsWindow } from "./git-command-runtime-metrics.js";
function createMetricsWindow(concurrencyLimit = 2) {
function createMetricsWindow(concurrencyLimit = 2, maxProcessesPerSecond = 8) {
let now = 1_000;
return {
metrics: new GitCommandRuntimeMetricsWindow(concurrencyLimit, () => now),
metrics: new GitCommandRuntimeMetricsWindow(
{ concurrencyLimit, maxProcessesPerSecond },
() => now,
),
advance(ms: number) {
now += ms;
},
@@ -42,6 +45,7 @@ describe("GitCommandRuntimeMetricsWindow", () => {
expect(metrics.snapshotAndReset()).toMatchObject({
concurrencyLimit: 1,
maxProcessesPerSecond: 8,
active: 1,
pending: 1,
peakActive: 1,
packages/server/src/utils/git-command-runtime-metrics.ts
+7 -2
View File
@@ -7,6 +7,7 @@ export interface GitCommandDurationStats {
export interface GitCommandRuntimeMetricsSnapshot {
concurrencyLimit: number;
maxProcessesPerSecond: number;
active: number;
pending: number;
peakActive: number;
@@ -44,7 +45,10 @@ export class GitCommandRuntimeMetricsWindow {
private readonly operationCounts = new Map<string, number>();
constructor(
private readonly concurrencyLimit: number,
private readonly limits: {
concurrencyLimit: number;
maxProcessesPerSecond: number;
},
private readonly clock: Clock = Date.now,
) {}
@@ -98,7 +102,8 @@ export class GitCommandRuntimeMetricsWindow {
...Array.from(this.pendingCommands, (metric) => metric.queuedAtMs),
);
const snapshot: GitCommandRuntimeMetricsSnapshot = {
concurrencyLimit: this.concurrencyLimit,
concurrencyLimit: this.limits.concurrencyLimit,
maxProcessesPerSecond: this.limits.maxProcessesPerSecond,
active: limiter.active,
pending: limiter.pending,
peakActive: this.peakActive,
packages/server/src/utils/git-process-scheduler.test.ts
+90
View File
@@ -0,0 +1,90 @@
import { afterEach, describe, expect, test, vi } from "vitest";
import { GitProcessScheduler, resolveGitProcessPolicy } from "./git-process-scheduler.js";
function createConcurrencyTask(
index: number,
started: number[],
releases: Array<() => void>,
): () => { result: Promise<void>; exited: Promise<void> } {
return () => {
started.push(index);
if (index >= 2) {
const completed = Promise.resolve();
return { result: completed, exited: completed };
}
const completed = new Promise<void>((resolve) => {
releases.push(resolve);
});
return { result: completed, exited: completed };
};
}
describe("GitProcessScheduler", () => {
afterEach(() => {
vi.useRealTimers();
});
test("limits unresolved processes independently of their start rate", async () => {
const scheduler = new GitProcessScheduler({
maxProcessesPerSecond: 100,
maxProcessConcurrency: 2,
});
const releases: Array<() => void> = [];
const started: number[] = [];
const tasks = Array.from({ length: 4 }, (_, index) =>
scheduler.run(createConcurrencyTask(index, started, releases)),
);
await vi.waitFor(() => expect(started).toEqual([0, 1]));
releases.shift()?.();
await vi.waitFor(() => expect(started).toContain(2));
releases.shift()?.();
await Promise.all(tasks);
expect(scheduler.activeCount).toBe(0);
expect(scheduler.pendingCount).toBe(0);
});
test("limits process starts in each strict one-second interval", async () => {
vi.useFakeTimers();
const scheduler = new GitProcessScheduler({
maxProcessesPerSecond: 2,
maxProcessConcurrency: 4,
});
const startedAt: number[] = [];
const recordStart = () => {
startedAt.push(Date.now());
const completed = Promise.resolve();
return { result: completed, exited: completed };
};
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);
await Promise.all(tasks);
expect(startedAt).toHaveLength(4);
expect(startedAt[2]! - startedAt[0]!).toBeGreaterThanOrEqual(1_000);
});
});
describe("resolveGitProcessPolicy", () => {
test("prefers renamed environment variables over legacy and persisted values", () => {
expect(
resolveGitProcessPolicy({
env: {
PASEO_GIT_MAX_PROCESSES_PER_SECOND: "12",
PASEO_GIT_MAX_PROCESS_CONCURRENCY: "7",
PASEO_GIT_CONCURRENCY: "3",
},
persisted: { maxProcessesPerSecond: 5, maxProcessConcurrency: 4 },
}),
).toEqual({ maxProcessesPerSecond: 12, maxProcessConcurrency: 7 });
});
});
packages/server/src/utils/git-process-scheduler.ts
+88
View File
@@ -0,0 +1,88 @@
import pLimit from "p-limit";
import pThrottle from "p-throttle";
export interface GitProcessPolicy {
maxProcessesPerSecond: number;
maxProcessConcurrency: number;
}
export const DEFAULT_GIT_PROCESS_POLICY: GitProcessPolicy = {
maxProcessesPerSecond: 64,
maxProcessConcurrency: 8,
};
export interface ScheduledGitProcess<T> {
result: Promise<T>;
exited: Promise<void>;
}
function parsePositiveInteger(value: string | undefined): number | undefined {
if (value === undefined) {
return undefined;
}
const parsed = Number(value);
return Number.isInteger(parsed) && parsed > 0 ? parsed : undefined;
}
export function resolveGitProcessPolicy(input: {
env: NodeJS.ProcessEnv;
persisted?: Partial<GitProcessPolicy>;
}): GitProcessPolicy {
return {
maxProcessesPerSecond:
parsePositiveInteger(input.env.PASEO_GIT_MAX_PROCESSES_PER_SECOND) ??
input.persisted?.maxProcessesPerSecond ??
DEFAULT_GIT_PROCESS_POLICY.maxProcessesPerSecond,
maxProcessConcurrency:
parsePositiveInteger(input.env.PASEO_GIT_MAX_PROCESS_CONCURRENCY) ??
// COMPAT(gitConcurrencyEnv): renamed in v0.2.6; remove after 2027-02-02.
parsePositiveInteger(input.env.PASEO_GIT_CONCURRENCY) ??
input.persisted?.maxProcessConcurrency ??
DEFAULT_GIT_PROCESS_POLICY.maxProcessConcurrency,
};
}
export class GitProcessScheduler {
private readonly limit;
private readonly throttle;
private running = 0;
private waiting = 0;
constructor(readonly policy: GitProcessPolicy) {
this.limit = pLimit(policy.maxProcessConcurrency);
this.throttle = pThrottle({
limit: policy.maxProcessesPerSecond,
interval: 1_000,
strict: true,
});
}
get activeCount(): number {
return this.running;
}
get pendingCount(): number {
return this.waiting;
}
run<T>(start: () => ScheduledGitProcess<T>): Promise<T> {
this.waiting += 1;
return new Promise<T>((resolve, reject) => {
void this.limit(
this.throttle(async () => {
this.waiting = Math.max(0, this.waiting - 1);
this.running += 1;
try {
const process = start();
void process.result.then(resolve, reject);
await process.exited;
} catch (error) {
reject(error);
} finally {
this.running = Math.max(0, this.running - 1);
}
}),
).catch(reject);
});
}
}
packages/server/src/utils/run-git-command.test.ts
+135 -8
View File
@@ -2,9 +2,14 @@ import { EventEmitter } from "node:events";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
interface FakeSpawnBehavior {
closeDelayMs?: number;
delayMs?: number;
emitError?: Error;
exitDelayMs?: number;
exitCode?: number | null;
killCloseDelayMs?: number;
killExitDelayMs?: number;
spawnError?: Error;
stderrData?: Buffer | string;
stdoutData?: Buffer | string;
}
@@ -43,10 +48,11 @@ class FakeChildProcess extends EventEmitter {
public readonly stdout = new EventEmitter();
public killed = false;
public killSignals: NodeJS.Signals[] = [];
public closed = false;
public exited = false;
private readonly behavior: FakeSpawnBehavior;
private readonly timers: NodeJS.Timeout[] = [];
private closed = false;
public constructor(behavior: FakeSpawnBehavior) {
super();
@@ -69,17 +75,31 @@ class FakeChildProcess extends EventEmitter {
this.killed = true;
this.killSignals.push(signal);
this.clearTimers();
const exitDelayMs = this.behavior.killExitDelayMs ?? 0;
this.schedule(() => {
this.finishClose({
this.finishExit({
exitCode: null,
signal,
});
}, 0);
}, exitDelayMs);
this.schedule(
() => {
this.finishClose({
exitCode: null,
signal,
});
},
Math.max(exitDelayMs, this.behavior.killCloseDelayMs ?? exitDelayMs),
);
return true;
}
public dispose(): void {
this.clearTimers();
if (!this.exited) {
fakeSpawnController.activeCount -= 1;
this.exited = true;
}
this.closed = true;
}
@@ -108,12 +128,22 @@ class FakeChildProcess extends EventEmitter {
return;
}
const exitDelayMs = this.behavior.exitDelayMs ?? this.behavior.delayMs ?? 0;
this.schedule(() => {
this.finishClose({
this.finishExit({
exitCode: this.behavior.exitCode ?? 0,
signal: null,
});
}, this.behavior.delayMs ?? 0);
}, exitDelayMs);
this.schedule(
() => {
this.finishClose({
exitCode: this.behavior.exitCode ?? 0,
signal: null,
});
},
Math.max(exitDelayMs, this.behavior.closeDelayMs ?? this.behavior.delayMs ?? exitDelayMs),
);
}
private finishClose({
@@ -125,19 +155,32 @@ class FakeChildProcess extends EventEmitter {
}): void {
if (this.closed) return;
this.finishExit({ exitCode, signal });
this.closed = true;
this.clearTimers();
fakeSpawnController.activeCount -= 1;
this.emit("close", exitCode, signal);
}
private finishExit({
exitCode,
signal,
}: {
exitCode: number | null;
signal: NodeJS.Signals | null;
}): void {
if (this.exited) return;
this.exited = true;
fakeSpawnController.activeCount -= 1;
this.emit("exit", exitCode, signal);
}
private finishError(error: Error): void {
if (this.closed) return;
this.closed = true;
this.clearTimers();
fakeSpawnController.activeCount -= 1;
this.emit("error", error);
this.finishClose({ exitCode: null, signal: null });
}
private schedule(callback: () => void, delayMs: number): void {
@@ -160,6 +203,9 @@ vi.mock("node:child_process", async () => {
...actual,
spawn: vi.fn(() => {
const behavior = fakeSpawnController.queue.shift() ?? {};
if (behavior.spawnError) {
throw behavior.spawnError;
}
const child = new FakeChildProcess(behavior);
fakeSpawnController.processes.push(child);
return child as unknown as ReturnType<typeof actual.spawn>;
@@ -230,6 +276,72 @@ describe("runGitCommand", () => {
});
});
it("holds the limiter slot until a timed out process exits without waiting for close", async () => {
const { runGitCommand } = await loadRunGitCommand(1);
enqueueSpawnBehaviors(
{ delayMs: 5_000, killExitDelayMs: 100, killCloseDelayMs: 500 },
{ delayMs: 0, stdoutData: "next" },
);
const timedOut = runGitCommand(["status"], {
cwd: process.cwd(),
timeout: 20,
}).then(
() => null,
(error: unknown) => error,
);
const next = runGitCommand(["rev-parse", "--show-toplevel"], {
cwd: process.cwd(),
});
await vi.waitFor(() => {
expect(fakeSpawnController.processes[0]?.killed).toBe(true);
});
expect(fakeSpawnController.processes).toHaveLength(1);
expect(fakeSpawnController.activeCount).toBe(1);
await expect(timedOut).resolves.toEqual(
expect.objectContaining({
message: "Git command timed out after 20ms: git status",
}),
);
await vi.waitFor(() => {
expect(fakeSpawnController.processes).toHaveLength(2);
});
expect(fakeSpawnController.processes[0]?.exited).toBe(true);
expect(fakeSpawnController.processes[0]?.closed).toBe(false);
await expect(next).resolves.toMatchObject({
exitCode: 0,
stdout: "next",
});
expect(fakeSpawnController.peakActiveCount).toBe(1);
});
it("finishes process metrics at exit without waiting for close", async () => {
const { runGitCommand, startGitCommandMetrics, stopGitCommandMetrics } =
await loadRunGitCommand(1);
enqueueSpawnBehaviors(
{ exitDelayMs: 0, closeDelayMs: 500 },
{ delayMs: 0, stdoutData: "next" },
);
startGitCommandMetrics();
void runGitCommand(["status"], { cwd: process.cwd() });
await expect(
runGitCommand(["rev-parse", "--show-toplevel"], { cwd: process.cwd() }),
).resolves.toMatchObject({ stdout: "next" });
expect(fakeSpawnController.processes[0]?.exited).toBe(true);
expect(fakeSpawnController.processes[0]?.closed).toBe(false);
expect(stopGitCommandMetrics()).toMatchObject({
total: 2,
failed: 0,
maxConcurrent: 1,
});
});
it("resolves truncated stdout, caps output, and kills the child process", async () => {
const { runGitCommand } = await loadRunGitCommand(1);
@@ -275,6 +387,21 @@ describe("runGitCommand", () => {
});
});
it("releases the limiter after a synchronous spawn failure", async () => {
const { runGitCommand } = await loadRunGitCommand(1);
enqueueSpawnBehaviors(
{ spawnError: new Error("spawn threw") },
{ delayMs: 0, stdoutData: "ok" },
);
await expect(runGitCommand(["status"], { cwd: process.cwd() })).rejects.toThrow("spawn threw");
await expect(runGitCommand(["status"], { cwd: process.cwd() })).resolves.toMatchObject({
exitCode: 0,
stdout: "ok",
});
}, 1_000);
it("traces git command spawn and close metadata when a logger is provided", async () => {
const { runGitCommand } = await loadRunGitCommand(1);
const trace = vi.fn();
packages/server/src/utils/run-git-command.ts
+251 -197
View File
@@ -1,5 +1,4 @@
import { existsSync } from "node:fs";
import pLimit from "p-limit";
import type { Logger } from "pino";
import type { ProcessEnvRecord } from "../server/paseo-env.js";
import {
@@ -13,14 +12,30 @@ import {
submitGitCommandTrace,
} from "./git-command-trace.js";
import { spawnProcess } from "./spawn.js";
import {
GitProcessScheduler,
resolveGitProcessPolicy,
type GitProcessPolicy,
} from "./git-process-scheduler.js";
const DEFAULT_TIMEOUT_MS = 30_000;
const DEFAULT_MAX_OUTPUT_BYTES = 20 * 1024 * 1024; // 20MB
const DEFAULT_STDERR_LIMIT = 2048;
const gitConcurrency = parseInt(process.env.PASEO_GIT_CONCURRENCY ?? "8", 10) || 8;
const gitLimit = pLimit(gitConcurrency);
const gitRuntimeMetrics = new GitCommandRuntimeMetricsWindow(gitConcurrency);
let gitProcessScheduler = new GitProcessScheduler(resolveGitProcessPolicy({ env: process.env }));
let gitRuntimeMetrics = createGitCommandRuntimeMetricsWindow(gitProcessScheduler.policy);
function createGitCommandRuntimeMetricsWindow(policy: GitProcessPolicy) {
return new GitCommandRuntimeMetricsWindow({
concurrencyLimit: policy.maxProcessConcurrency,
maxProcessesPerSecond: policy.maxProcessesPerSecond,
});
}
export function configureGitProcessPolicy(policy: GitProcessPolicy): void {
gitProcessScheduler = new GitProcessScheduler(policy);
gitRuntimeMetrics = createGitCommandRuntimeMetricsWindow(policy);
}
export interface GitCommandOptions {
cwd: string;
@@ -94,8 +109,8 @@ export function stopGitCommandMetrics(): GitCommandMetricsSnapshot {
export function snapshotGitCommandRuntimeMetrics(): GitCommandRuntimeMetricsSnapshot {
return gitRuntimeMetrics.snapshotAndReset({
active: gitLimit.activeCount,
pending: gitLimit.pendingCount,
active: gitProcessScheduler.activeCount,
pending: gitProcessScheduler.pendingCount,
});
}
@@ -142,232 +157,271 @@ export function runGitCommand(
options: GitCommandOptions,
): Promise<GitCommandResult> {
const commandTrace = submitGitCommandTrace(args, options.cwd, {
active: gitLimit.activeCount,
pending: gitLimit.pendingCount,
active: gitProcessScheduler.activeCount,
pending: gitProcessScheduler.pendingCount,
});
const runtimeMetric = gitRuntimeMetrics.submit(getGitOperation(args));
const promise = gitLimit(
() =>
new Promise<GitCommandResult>((resolve, reject) => {
startGitCommandTrace(commandTrace, {
active: gitLimit.activeCount,
pending: gitLimit.pendingCount,
const promise = gitProcessScheduler.run(() => {
let releaseProcessSlot!: () => void;
const exited = new Promise<void>((resolve) => {
releaseProcessSlot = resolve;
});
const resultPromise = new Promise<GitCommandResult>((resolve, reject) => {
startGitCommandTrace(commandTrace, {
active: gitProcessScheduler.activeCount,
pending: gitProcessScheduler.pendingCount,
});
gitRuntimeMetrics.start(runtimeMetric);
const timeout = options.timeout ?? DEFAULT_TIMEOUT_MS;
const maxOutputBytes = options.maxOutputBytes ?? DEFAULT_MAX_OUTPUT_BYTES;
const acceptExitCodes = options.acceptExitCodes ?? [0];
const command = formatGitCommand(args);
const envOverlay = mergeEnvOverlays(options.env, options.envOverlay);
const startedAt = Date.now();
const metricsState = beginGitCommandMetric();
const logger = typeof options.logger?.trace === "function" ? options.logger : undefined;
const traceContext = logger
? {
command: "git",
args,
cwd: options.cwd,
cwdExists: existsSync(options.cwd),
timeout,
maxOutputBytes,
acceptExitCodes,
envOverlayKeys: getEnvOverlayKeys(envOverlay),
}
: null;
if (logger && traceContext) {
logger.trace(traceContext, "Spawning git command");
}
let settled = false;
let metricFinished = false;
let processError: Error | null = null;
let processExit: { exitCode: number | null; signal: NodeJS.Signals | null } | null = null;
let timeoutError: Error | null = null;
let truncated = false;
let stdoutBytes = 0;
let stderrBytes = 0;
const stdoutChunks: Buffer[] = [];
const stderrChunks: Buffer[] = [];
let timer: NodeJS.Timeout | undefined;
const settle = (callback: () => void) => {
if (settled) return;
settled = true;
if (timer) clearTimeout(timer);
callback();
};
const finishMetricOnce = (metric: GitCommandMetric, timedOut = false) => {
if (metricFinished) return;
metricFinished = true;
finishGitCommandMetric(metricsState, metric);
gitRuntimeMetrics.finish(runtimeMetric, { success: metric.success, timedOut });
};
const settleTimeoutTrace = (exitCode: number | null, signal: NodeJS.Signals | null) => {
settleGitCommandTrace(commandTrace, {
outcome: "timed_out",
exitCode,
signal,
});
gitRuntimeMetrics.start(runtimeMetric);
const timeout = options.timeout ?? DEFAULT_TIMEOUT_MS;
const maxOutputBytes = options.maxOutputBytes ?? DEFAULT_MAX_OUTPUT_BYTES;
const acceptExitCodes = options.acceptExitCodes ?? [0];
const command = formatGitCommand(args);
const envOverlay = mergeEnvOverlays(options.env, options.envOverlay);
const startedAt = Date.now();
const metricsState = beginGitCommandMetric();
const logger = typeof options.logger?.trace === "function" ? options.logger : undefined;
const traceContext = logger
? {
command: "git",
args,
cwd: options.cwd,
cwdExists: existsSync(options.cwd),
timeout,
maxOutputBytes,
acceptExitCodes,
envOverlayKeys: getEnvOverlayKeys(envOverlay),
}
: null;
};
if (logger && traceContext) {
logger.trace(traceContext, "Spawning git command");
const markProcessExited = (exitCode: number | null, signal: NodeJS.Signals | null) => {
if (processExit) return;
processExit = { exitCode, signal };
const timedOut = timeoutError !== null;
finishMetricOnce(
{
args,
cwd: options.cwd,
startedAtMs: startedAt,
durationMs: Date.now() - startedAt,
exitCode,
signal,
success:
!timedOut && !processError && (truncated || acceptExitCodes.includes(exitCode ?? -1)),
},
timedOut,
);
releaseProcessSlot();
if (timedOut) {
settleTimeoutTrace(exitCode, signal);
}
};
const rejectSpawnFailure = (error: unknown) => {
processError = error instanceof Error ? error : new Error(String(error));
markProcessExited(null, null);
settleGitCommandTrace(commandTrace, {
outcome: "spawn_error",
exitCode: null,
signal: null,
});
settle(() => reject(processError));
};
let child: ReturnType<typeof spawnProcess>;
try {
// `core.quotepath=false` makes git emit raw UTF-8 paths instead of
// octal-escaping non-ASCII bytes (e.g. `测试文件.txt` vs `"\346\265\213..."`).
const child = spawnProcess("git", ["-c", "core.quotepath=false", ...args], {
child = spawnProcess("git", ["-c", "core.quotepath=false", ...args], {
cwd: options.cwd,
envOverlay,
shell: false,
stdio: ["ignore", "pipe", "pipe"],
});
spawnGitCommandTrace(commandTrace, child.pid);
} catch (error) {
rejectSpawnFailure(error);
return;
}
let settled = false;
let metricFinished = false;
let truncated = false;
let stdoutBytes = 0;
let stderrBytes = 0;
const stdoutChunks: Buffer[] = [];
const stderrChunks: Buffer[] = [];
const stdout = child.stdout;
const stderr = child.stderr;
if (!stdout || !stderr) {
child.kill("SIGKILL");
rejectSpawnFailure(new Error("Git process did not expose piped stdout and stderr"));
return;
}
const settle = (callback: () => void) => {
if (settled) return;
settled = true;
clearTimeout(timer);
callback();
};
timer = setTimeout(() => {
timeoutError = new Error(`Git command timed out after ${timeout}ms: ${command}`);
child.kill("SIGKILL");
settle(() => reject(timeoutError));
if (processExit) {
settleTimeoutTrace(processExit.exitCode, processExit.signal);
}
}, timeout);
const finishMetricOnce = (metric: GitCommandMetric, timedOut = false) => {
if (metricFinished) return;
metricFinished = true;
finishGitCommandMetric(metricsState, metric);
gitRuntimeMetrics.finish(runtimeMetric, { success: metric.success, timedOut });
};
stdout.on("data", (chunk: Buffer | string) => {
if (settled || truncated) return;
const timer = setTimeout(() => {
const error = new Error(`Git command timed out after ${timeout}ms: ${command}`);
const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
const remainingBytes = maxOutputBytes - stdoutBytes;
if (remainingBytes <= 0) {
truncated = true;
child.kill("SIGKILL");
settleGitCommandTrace(commandTrace, {
outcome: "timed_out",
exitCode: null,
signal: "SIGKILL",
});
finishMetricOnce(
return;
}
if (buffer.length > remainingBytes) {
stdoutChunks.push(buffer.subarray(0, remainingBytes));
stdoutBytes += remainingBytes;
truncated = true;
child.kill("SIGKILL");
return;
}
stdoutChunks.push(buffer);
stdoutBytes += buffer.length;
});
stderr.on("data", (chunk: Buffer | string) => {
if (settled || stderrBytes >= DEFAULT_STDERR_LIMIT) return;
const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
const remainingBytes = DEFAULT_STDERR_LIMIT - stderrBytes;
if (buffer.length > remainingBytes) {
stderrChunks.push(buffer.subarray(0, remainingBytes));
stderrBytes += remainingBytes;
return;
}
stderrChunks.push(buffer);
stderrBytes += buffer.length;
});
child.on("error", (error) => {
processError = error;
if (logger && traceContext) {
logger.trace(
{
args,
cwd: options.cwd,
startedAtMs: startedAt,
...traceContext,
err: error,
durationMs: Date.now() - startedAt,
exitCode: null,
signal: "SIGKILL",
success: false,
},
true,
"Git command process error",
);
settle(() => reject(error));
}, timeout);
}
});
child.stdout!.on("data", (chunk: Buffer | string) => {
if (settled || truncated) return;
child.on("exit", markProcessExited);
const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
const remainingBytes = maxOutputBytes - stdoutBytes;
if (remainingBytes <= 0) {
truncated = true;
child.kill("SIGKILL");
return;
}
if (buffer.length > remainingBytes) {
stdoutChunks.push(buffer.subarray(0, remainingBytes));
stdoutBytes += remainingBytes;
truncated = true;
child.kill("SIGKILL");
return;
}
stdoutChunks.push(buffer);
stdoutBytes += buffer.length;
});
child.stderr!.on("data", (chunk: Buffer | string) => {
if (settled || stderrBytes >= DEFAULT_STDERR_LIMIT) return;
const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
const remainingBytes = DEFAULT_STDERR_LIMIT - stderrBytes;
if (buffer.length > remainingBytes) {
stderrChunks.push(buffer.subarray(0, remainingBytes));
stderrBytes += remainingBytes;
return;
}
stderrChunks.push(buffer);
stderrBytes += buffer.length;
});
child.on("error", (error) => {
settleGitCommandTrace(commandTrace, {
outcome: "spawn_error",
exitCode: null,
signal: null,
});
finishMetricOnce({
args,
cwd: options.cwd,
startedAtMs: startedAt,
durationMs: Date.now() - startedAt,
exitCode: null,
signal: null,
success: false,
});
if (logger && traceContext) {
logger.trace(
{
...traceContext,
err: error,
durationMs: Date.now() - startedAt,
},
"Git command process error",
);
}
settle(() => reject(error));
});
child.on("close", (exitCode, signal) => {
settleGitCommandTrace(commandTrace, {
outcome: "closed",
exitCode,
signal,
truncated,
});
const result: GitCommandResult = {
stdout: Buffer.concat(stdoutChunks).toString("utf8"),
stderr: Buffer.concat(stderrChunks).toString("utf8"),
truncated,
exitCode,
signal,
};
if (logger && traceContext) {
logger.trace(
{
...traceContext,
durationMs: Date.now() - startedAt,
exitCode,
signal,
truncated,
stdoutBytes,
stderrBytes,
},
"Git command closed",
);
}
if (!truncated && !acceptExitCodes.includes(exitCode ?? -1)) {
finishMetricOnce({
args,
cwd: options.cwd,
startedAtMs: startedAt,
child.on("close", (exitCode, signal) => {
markProcessExited(exitCode, signal);
const result: GitCommandResult = {
stdout: Buffer.concat(stdoutChunks).toString("utf8"),
stderr: Buffer.concat(stderrChunks).toString("utf8"),
truncated,
exitCode,
signal,
};
if (logger && traceContext) {
logger.trace(
{
...traceContext,
durationMs: Date.now() - startedAt,
exitCode,
signal,
success: false,
});
const stderrPreview = result.stderr.trim() || "(no stderr)";
const truncationNote = result.truncated ? " (stdout truncated)" : "";
truncated,
stdoutBytes,
stderrBytes,
},
"Git command closed",
);
}
settle(() =>
reject(
new Error(
`Git command failed: ${command}${truncationNote} (exit code: ${String(exitCode)}, signal: ${signal ?? "none"})\n${stderrPreview}`,
),
),
);
return;
}
if (timeoutError) {
return;
}
finishMetricOnce({
args,
cwd: options.cwd,
startedAtMs: startedAt,
durationMs: Date.now() - startedAt,
if (processError) {
settleGitCommandTrace(commandTrace, {
outcome: "spawn_error",
exitCode,
signal,
success: true,
});
settle(() => resolve(result));
settle(() => reject(processError));
return;
}
settleGitCommandTrace(commandTrace, {
outcome: "closed",
exitCode,
signal,
truncated,
});
}),
if (!truncated && !acceptExitCodes.includes(exitCode ?? -1)) {
const stderrPreview = result.stderr.trim() || "(no stderr)";
const truncationNote = result.truncated ? " (stdout truncated)" : "";
settle(() =>
reject(
new Error(
`Git command failed: ${command}${truncationNote} (exit code: ${String(exitCode)}, signal: ${signal ?? "none"})\n${stderrPreview}`,
),
),
);
return;
}
settle(() => resolve(result));
});
});
return { result: resultPromise, exited };
});
gitRuntimeMetrics.observeLimiter(
gitProcessScheduler.activeCount,
gitProcessScheduler.pendingCount,
);
gitRuntimeMetrics.observeLimiter(gitLimit.activeCount, gitLimit.pendingCount);
return promise;
}