Keep workspace Git refreshes responsive at scale (#2979)

* fix(server): prevent Git watcher fanout stalls Repository metadata events were refreshing every observed worktree, multiplying Git process submissions and starving workspace create and archive requests behind observation work. Route metadata to the workspaces that depend on it, preserve conservative shared-ref refreshes, and admit interactive lifecycle operations ahead of queued background reads. * fix(server): preserve Git metadata dependencies
2026-08-07 18:50:43 +08:00
parent a5184fd08a
commit a06e4d60e9
14 changed files with 1569 additions and 57 deletions
@@ -46,6 +46,27 @@ describe("GitProcessScheduler", () => {
    expect(scheduler.pendingCount).toBe(0);
  });

+  test("admits a high-priority user operation before queued observation work", async () => {
+    const scheduler = new GitProcessScheduler({
+      maxProcessesPerSecond: 100,
+      maxProcessConcurrency: 1,
+    });
+    const releases: Array<() => void> = [];
+    const started: number[] = [];
+
+    const activeObservation = scheduler.run(createConcurrencyTask(0, started, releases));
+    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("limits process starts in each strict one-second interval", async () => {
    vi.useFakeTimers();
    const scheduler = new GitProcessScheduler({
@@ -1,4 +1,3 @@
-import pLimit from "p-limit";
 import pThrottle from "p-throttle";

 export interface GitProcessPolicy {
@@ -16,6 +15,8 @@ export interface ScheduledGitProcess<T> {
  exited: Promise<void>;
 }

+export type GitProcessPriority = "normal" | "high";
+
 function parsePositiveInteger(value: string | undefined): number | undefined {
  if (value === undefined) {
    return undefined;
@@ -43,18 +44,20 @@ export function resolveGitProcessPolicy(input: {
 }

 export class GitProcessScheduler {
-  private readonly limit;
-  private readonly throttle;
+  private readonly startThrottled;
+  private readonly highPriorityQueue: Array<() => void> = [];
+  private readonly normalPriorityQueue: Array<() => void> = [];
+  private admitted = 0;
  private running = 0;
  private waiting = 0;

  constructor(readonly policy: GitProcessPolicy) {
-    this.limit = pLimit(policy.maxProcessConcurrency);
-    this.throttle = pThrottle({
+    const throttle = pThrottle({
      limit: policy.maxProcessesPerSecond,
      interval: 1_000,
      strict: true,
    });
+    this.startThrottled = throttle((start: () => Promise<void>) => start());
  }

  get activeCount(): number {
@@ -65,24 +68,55 @@ export class GitProcessScheduler {
    return this.waiting;
  }

-  run<T>(start: () => ScheduledGitProcess<T>): Promise<T> {
+  run<T>(
+    start: () => ScheduledGitProcess<T>,
+    options?: { priority?: GitProcessPriority },
+  ): Promise<T> {
    this.waiting += 1;
    return new Promise<T>((resolve, reject) => {
-      void this.limit(
-        this.throttle(async () => {
+      const admit = () => {
+        void this.startThrottled(() => this.execute(start, resolve, reject)).catch((error) => {
          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);
+          this.admitted = Math.max(0, this.admitted - 1);
+          reject(error);
+          this.drain();
+        });
+      };
+      const queue =
+        options?.priority === "high" ? this.highPriorityQueue : this.normalPriorityQueue;
+      queue.push(admit);
+      this.drain();
    });
  }
+
+  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 async execute<T>(
+    start: () => ScheduledGitProcess<T>,
+    resolve: (value: T | PromiseLike<T>) => void,
+    reject: (reason?: unknown) => void,
+  ): Promise<void> {
+    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);
+      this.admitted = Math.max(0, this.admitted - 1);
+      this.drain();
+    }
+  }
 }
@@ -42,6 +42,13 @@ describe("path equivalence", () => {
    );
  });

+  test("preserves the casing of Windows relative suffixes", () => {
+    expect(getRealpathAwareRelativePath("C:\\Repo\\.git", "c:\\repo\\.git\\HEAD")).toBe("HEAD");
+    expect(
+      getRealpathAwareRelativePath("C:\\Repo\\.git", "c:\\repo\\.git\\refs\\heads\\FeatureCase"),
+    ).toBe("refs\\heads\\FeatureCase");
+  });
+
  test.skipIf(process.platform === "win32")(
    "derives the contained suffix from a realpath-equivalent root",
    () => {
@@ -97,11 +97,18 @@ function getRelativePathInsideRoot(root: string, candidate: string): string | nu
  const platformPath = compareAsWindows ? nodePath.win32 : nodePath.posix;
  const normalizedRoot = normalizePathForComparison(root, compareAsWindows);
  const normalizedCandidate = normalizePathForComparison(candidate, compareAsWindows);
-  const relative = platformPath.relative(normalizedRoot, normalizedCandidate);
+  const comparableRelative = platformPath.relative(normalizedRoot, normalizedCandidate);

-  return relative === "" || (!relative.startsWith("..") && !platformPath.isAbsolute(relative))
-    ? relative
-    : null;
+  if (
+    comparableRelative !== "" &&
+    (comparableRelative.startsWith("..") || platformPath.isAbsolute(comparableRelative))
+  ) {
+    return null;
+  }
+
+  const casePreservingRoot = normalizePathPreservingCase(root, compareAsWindows);
+  const casePreservingCandidate = normalizePathPreservingCase(candidate, compareAsWindows);
+  return platformPath.relative(casePreservingRoot, casePreservingCandidate);
 }

 function collectPathVariants(value: string): string[] {
@@ -140,15 +147,19 @@ function looksLikeDefiniteWindowsPath(value: string): boolean {
 }

 function normalizePathForComparison(value: string, compareAsWindows: boolean): string {
+  const normalized = normalizePathPreservingCase(value, compareAsWindows);
+  return compareAsWindows ? normalized.toLowerCase() : normalized;
+}
+
+function normalizePathPreservingCase(value: string, compareAsWindows: boolean): string {
  const platformPath = compareAsWindows ? nodePath.win32 : nodePath.posix;
  const comparableValue = compareAsWindows ? stripWindowsNamespacePrefix(value) : value;
  const platformNormalized = platformPath.normalize(comparableValue);
-  const normalized = stripTrailingSeparators(
+  return stripTrailingSeparators(
    platformNormalized,
    platformPath.parse(platformNormalized).root,
    compareAsWindows,
  );
-  return compareAsWindows ? normalized.toLowerCase() : normalized;
 }

 function stripWindowsNamespacePrefix(value: string): string {
@@ -342,6 +342,73 @@ describe("runGitCommand", () => {
    });
  });

+  it("records submitted Git commands before they start and refuses an incomplete measurement", async () => {
+    const {
+      getGitCommandMetrics,
+      runGitCommand,
+      startGitCommandMetrics,
+      stopGitCommandMetrics,
+      waitForGitCommandMetricsIdle,
+    } = await loadRunGitCommand(1);
+
+    enqueueSpawnBehaviors({ delayMs: 100 }, { delayMs: 0 });
+    startGitCommandMetrics();
+
+    const first = runGitCommand(["status", "--short"], { cwd: process.cwd() });
+    const second = runGitCommand(["rev-parse", "--show-toplevel"], { cwd: process.cwd() });
+
+    await vi.waitFor(() => {
+      expect(fakeSpawnController.processes).toHaveLength(1);
+    });
+    expect(getGitCommandMetrics()).toMatchObject({
+      submitted: 2,
+      started: 1,
+      completed: 0,
+      active: 1,
+      pending: 1,
+      submissions: [
+        { args: ["status", "--short"], cwd: process.cwd() },
+        { args: ["rev-parse", "--show-toplevel"], cwd: process.cwd() },
+      ],
+    });
+    expect(() => stopGitCommandMetrics()).toThrow(
+      "Cannot stop Git command metrics while 2 submitted commands are unfinished",
+    );
+
+    await Promise.all([first, second]);
+    await waitForGitCommandMetricsIdle({ quietMs: 0, timeoutMs: 1_000 });
+    expect(stopGitCommandMetrics()).toMatchObject({
+      submitted: 2,
+      started: 2,
+      completed: 2,
+      active: 0,
+      pending: 0,
+      total: 2,
+    });
+  });
+
+  it("carries high-priority admission across an async user operation", async () => {
+    const { runGitCommand, runWithGitCommandPriority } = await loadRunGitCommand(1);
+    enqueueSpawnBehaviors(
+      { delayMs: 100, stdoutData: "active" },
+      { stdoutData: "high" },
+      { stdoutData: "normal" },
+    );
+
+    const active = runGitCommand(["status"], { cwd: process.cwd() });
+    await vi.waitFor(() => expect(fakeSpawnController.processes).toHaveLength(1));
+    const normal = runGitCommand(["rev-parse", "normal"], { cwd: process.cwd() });
+    const high = runWithGitCommandPriority("high", () =>
+      runGitCommand(["rev-parse", "high"], { cwd: process.cwd() }),
+    );
+
+    await expect(Promise.all([active, high, normal])).resolves.toEqual([
+      expect.objectContaining({ stdout: "active" }),
+      expect.objectContaining({ stdout: "high" }),
+      expect.objectContaining({ stdout: "normal" }),
+    ]);
+  });
+
  it("resolves truncated stdout, caps output, and kills the child process", async () => {
    const { runGitCommand } = await loadRunGitCommand(1);

@@ -1,3 +1,4 @@
+import { AsyncLocalStorage } from "node:async_hooks";
 import { existsSync } from "node:fs";
 import type { Logger } from "pino";
 import type { ProcessEnvRecord } from "../server/paseo-env.js";
@@ -14,6 +15,7 @@ import {
 import { spawnProcess } from "./spawn.js";
 import {
  GitProcessScheduler,
+  type GitProcessPriority,
  resolveGitProcessPolicy,
  type GitProcessPolicy,
 } from "./git-process-scheduler.js";
@@ -24,6 +26,11 @@ const DEFAULT_STDERR_LIMIT = 2048;

 let gitProcessScheduler = new GitProcessScheduler(resolveGitProcessPolicy({ env: process.env }));
 let gitRuntimeMetrics = createGitCommandRuntimeMetricsWindow(gitProcessScheduler.policy);
+const gitCommandPriority = new AsyncLocalStorage<GitProcessPriority>();
+
+export function runWithGitCommandPriority<T>(priority: GitProcessPriority, operation: () => T): T {
+  return gitCommandPriority.run(priority, operation);
+}

 function createGitCommandRuntimeMetricsWindow(policy: GitProcessPolicy) {
  return new GitCommandRuntimeMetricsWindow({
@@ -67,15 +74,31 @@ export interface GitCommandMetric {

 export interface GitCommandMetricsSnapshot {
  commands: GitCommandMetric[];
+  submissions: GitCommandSubmissionMetric[];
+  submitted: number;
+  started: number;
+  completed: number;
+  active: number;
+  pending: number;
  total: number;
  failed: number;
  maxConcurrent: number;
 }

+export interface GitCommandSubmissionMetric {
+  args: string[];
+  cwd: string;
+}
+
 interface GitCommandMetricsState {
  commands: GitCommandMetric[];
+  submissions: GitCommandSubmissionMetric[];
+  submitted: number;
+  started: number;
+  completed: number;
  active: number;
  maxConcurrent: number;
+  lastSubmittedAtMs: number;
 }

 let gitCommandMetricsState: GitCommandMetricsState | null = null;
@@ -83,24 +106,87 @@ let gitCommandMetricsState: GitCommandMetricsState | null = null;
 export function startGitCommandMetrics(): void {
  gitCommandMetricsState = {
    commands: [],
+    submissions: [],
+    submitted: 0,
+    started: 0,
+    completed: 0,
    active: 0,
    maxConcurrent: 0,
+    lastSubmittedAtMs: Date.now(),
  };
 }

 export function stopGitCommandMetrics(): GitCommandMetricsSnapshot {
  const state = gitCommandMetricsState;
-  gitCommandMetricsState = null;
  if (!state) {
    return {
      commands: [],
+      submissions: [],
+      submitted: 0,
+      started: 0,
+      completed: 0,
+      active: 0,
+      pending: 0,
      total: 0,
      failed: 0,
      maxConcurrent: 0,
    };
  }
+  const unfinished = state.submitted - state.completed;
+  if (unfinished > 0) {
+    throw new Error(
+      `Cannot stop Git command metrics while ${unfinished} submitted commands are unfinished`,
+    );
+  }
+  gitCommandMetricsState = null;
+  return snapshotGitCommandMetricsState(state);
+}
+
+export function getGitCommandMetrics(): GitCommandMetricsSnapshot {
+  const state = gitCommandMetricsState;
+  if (!state) {
+    return stopGitCommandMetrics();
+  }
+  return snapshotGitCommandMetricsState(state);
+}
+
+export async function waitForGitCommandMetricsIdle(options: {
+  quietMs: number;
+  timeoutMs: number;
+}): Promise<void> {
+  const startedAtMs = Date.now();
+  while (true) {
+    const state = gitCommandMetricsState;
+    if (!state) {
+      throw new Error("Git command metrics are not running");
+    }
+    const isComplete = state.completed === state.submitted;
+    const hasBeenQuiet = Date.now() - state.lastSubmittedAtMs >= options.quietMs;
+    if (isComplete && hasBeenQuiet) {
+      return;
+    }
+    if (Date.now() - startedAtMs >= options.timeoutMs) {
+      const unfinished = state.submitted - state.completed;
+      throw new Error(
+        `Timed out waiting for Git command metrics to become idle (${unfinished} unfinished)`,
+      );
+    }
+    await new Promise((resolve) => setTimeout(resolve, 10));
+  }
+}
+
+function snapshotGitCommandMetricsState(state: GitCommandMetricsState): GitCommandMetricsSnapshot {
  return {
    commands: [...state.commands],
+    submissions: state.submissions.map((submission) => ({
+      args: [...submission.args],
+      cwd: submission.cwd,
+    })),
+    submitted: state.submitted,
+    started: state.started,
+    completed: state.completed,
+    active: state.active,
+    pending: state.submitted - state.started,
    total: state.commands.length,
    failed: state.commands.filter((command) => !command.success).length,
    maxConcurrent: state.maxConcurrent,
@@ -114,14 +200,24 @@ export function snapshotGitCommandRuntimeMetrics(): GitCommandRuntimeMetricsSnap
  });
 }

-function beginGitCommandMetric(): GitCommandMetricsState | null {
+function submitGitCommandMetric(args: string[], cwd: string): GitCommandMetricsState | null {
  const state = gitCommandMetricsState;
  if (!state) {
    return null;
  }
+  state.submissions.push({ args: [...args], cwd });
+  state.submitted += 1;
+  state.lastSubmittedAtMs = Date.now();
+  return state;
+}
+
+function beginGitCommandMetric(state: GitCommandMetricsState | null): void {
+  if (!state) {
+    return;
+  }
+  state.started += 1;
  state.active += 1;
  state.maxConcurrent = Math.max(state.maxConcurrent, state.active);
-  return state;
 }

 function finishGitCommandMetric(
@@ -132,6 +228,7 @@ function finishGitCommandMetric(
    return;
  }
  state.active = Math.max(0, state.active - 1);
+  state.completed += 1;
  state.commands.push(metric);
 }

@@ -156,12 +253,13 @@ export function runGitCommand(
  args: string[],
  options: GitCommandOptions,
 ): Promise<GitCommandResult> {
+  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 promise = gitProcessScheduler.run(() => {
+  const startCommand = () => {
    let releaseProcessSlot!: () => void;
    const exited = new Promise<void>((resolve) => {
      releaseProcessSlot = resolve;
@@ -178,7 +276,7 @@ export function runGitCommand(
      const command = formatGitCommand(args);
      const envOverlay = mergeEnvOverlays(options.env, options.envOverlay);
      const startedAt = Date.now();
-      const metricsState = beginGitCommandMetric();
+      beginGitCommandMetric(metricsState);
      const logger = typeof options.logger?.trace === "function" ? options.logger : undefined;
      const traceContext = logger
        ? {
@@ -417,6 +515,9 @@ export function runGitCommand(
      });
    });
    return { result: resultPromise, exited };
+  };
+  const promise = gitProcessScheduler.run(startCommand, {
+    priority: gitCommandPriority.getStore() ?? "normal",
  });
  gitRuntimeMetrics.observeLimiter(
    gitProcessScheduler.activeCount,