fix(server): keep cold diffs out from behind background git fan-out

Add a background lane to the git process scheduler and run reconciliation
checkout reads, observation-driven workspace refreshes and background fetches
in it. Bound reconciliation checkout reads to four at a time. Stop the Linux
file observer's startup scan from reporting existing files as creates, which
made every cold diff compute and discard its first revision.
This commit is contained in:
Mohamed Boudra
2026-11-10 02:49:54 +01:00
parent f9779a14a5
commit 1d4b436bf6
12 changed files with 423 additions and 48 deletions
@@ -68,6 +68,32 @@ describe("GitProcessScheduler", () => {
expect(started).toEqual([0, 3, 2]);
});
test("admits normal work before a queued background burst", async () => {
const scheduler = new GitProcessScheduler({
maxProcessesPerSecond: 100,
maxProcessConcurrency: 1,
});
const releases: Array<() => void> = [];
const started: number[] = [];
const activeBackground = scheduler.run(createConcurrencyTask(0, started, releases), {
priority: "background",
});
await vi.waitFor(() => expect(started).toEqual([0]));
const queuedBackground = [2, 3, 4].map((index) =>
scheduler.run(createConcurrencyTask(index, started, releases), { priority: "background" }),
);
const normalWork = scheduler.run(createConcurrencyTask(5, started, releases));
const highWork = scheduler.run(createConcurrencyTask(6, started, releases), {
priority: "high",
});
releases.shift()?.();
await Promise.all([activeBackground, ...queuedBackground, normalWork, highWork]);
expect(started).toEqual([0, 6, 5, 2, 3, 4]);
});
test("yields to daemon liveness work between process starts", async () => {
const scheduler = new GitProcessScheduler({
maxProcessesPerSecond: 100,
@@ -16,7 +16,10 @@ export interface ScheduledGitProcess<T> {
exited: Promise<void>;
}
export type GitProcessPriority = "normal" | "high";
// Lanes drain in this order. "background" is for daemon-initiated fan-out
// (reconciliation, observation-driven refreshes, fetches) so it never sits in
// front of work a user is waiting for.
export type GitProcessPriority = "high" | "normal" | "background";
function parsePositiveInteger(value: string | undefined): number | undefined {
if (value === undefined) {
@@ -46,8 +49,11 @@ export function resolveGitProcessPolicy(input: {
export class GitProcessScheduler {
private readonly startThrottled;
private readonly highPriorityQueue: Array<() => void> = [];
private readonly normalPriorityQueue: Array<() => void> = [];
private readonly queues: Record<GitProcessPriority, Array<() => void>> = {
high: [],
normal: [],
background: [],
};
private admitted = 0;
private running = 0;
private waiting = 0;
@@ -84,9 +90,7 @@ export class GitProcessScheduler {
this.scheduleDrain();
});
};
const queue =
options?.priority === "high" ? this.highPriorityQueue : this.normalPriorityQueue;
queue.push(admit);
this.queues[options?.priority ?? "normal"].push(admit);
if (this.admitted === 0 && !this.drainScheduled) {
this.drainOne();
} else {
@@ -99,19 +103,28 @@ export class GitProcessScheduler {
if (this.drainScheduled || this.admitted >= this.policy.maxProcessConcurrency) {
return;
}
if (this.highPriorityQueue.length === 0 && this.normalPriorityQueue.length === 0) {
if (!this.hasQueuedWork()) {
return;
}
this.drainScheduled = true;
scheduleImmediate(() => this.drainOne());
}
private hasQueuedWork(): boolean {
return (
this.queues.high.length > 0 ||
this.queues.normal.length > 0 ||
this.queues.background.length > 0
);
}
private drainOne(): void {
this.drainScheduled = false;
if (this.admitted >= this.policy.maxProcessConcurrency) {
return;
}
const admit = this.highPriorityQueue.shift() ?? this.normalPriorityQueue.shift();
const admit =
this.queues.high.shift() ?? this.queues.normal.shift() ?? this.queues.background.shift();
if (!admit) {
return;
}
+34 -19
View File
@@ -67,8 +67,17 @@ export type RunGitCommand = (
options: GitCommandOptions,
) => Promise<GitCommandResult>;
export function createRunGitCommand(provenance: string): RunGitCommand {
return (args, options) => runGitCommandWithProvenance(args, options, provenance);
export interface GitCommandRunnerOptions {
/** Overrides the ambient `runWithGitCommandPriority` lane for every command this runner starts. */
priority?: GitProcessPriority;
}
export function createRunGitCommand(
provenance: string,
runnerOptions: GitCommandRunnerOptions = {},
): RunGitCommand {
return (args, options) =>
executeGitCommand(args, options, decodeUtf8, { provenance, priority: runnerOptions.priority });
}
export interface GitCommandMetric {
@@ -97,6 +106,7 @@ export interface GitCommandMetricsSnapshot {
export interface GitCommandSubmissionMetric {
args: string[];
cwd: string;
priority: GitProcessPriority;
}
interface GitCommandMetricsState {
@@ -188,8 +198,8 @@ function snapshotGitCommandMetricsState(state: GitCommandMetricsState): GitComma
return {
commands: [...state.commands],
submissions: state.submissions.map((submission) => ({
...submission,
args: [...submission.args],
cwd: submission.cwd,
})),
submitted: state.submitted,
started: state.started,
@@ -209,12 +219,16 @@ export function snapshotGitCommandRuntimeMetrics(): GitCommandRuntimeMetricsSnap
});
}
function submitGitCommandMetric(args: string[], cwd: string): GitCommandMetricsState | null {
function submitGitCommandMetric(
args: string[],
cwd: string,
priority: GitProcessPriority,
): GitCommandMetricsState | null {
const state = gitCommandMetricsState;
if (!state) {
return null;
}
state.submissions.push({ args: [...args], cwd });
state.submissions.push({ args: [...args], cwd, priority });
state.submitted += 1;
state.lastSubmittedAtMs = Date.now();
return state;
@@ -258,12 +272,13 @@ function getEnvOverlayKeys(envOverlay: ProcessEnvRecord | undefined): string[] {
return Object.keys(envOverlay ?? {}).sort();
}
function runGitCommandWithProvenance(
args: string[],
options: GitCommandOptions,
provenance?: string,
): Promise<GitCommandResult> {
return executeGitCommand(args, options, (output) => output.toString("utf8"), provenance);
function decodeUtf8(output: Buffer): string {
return output.toString("utf8");
}
interface GitCommandSubmission {
provenance?: string;
priority?: GitProcessPriority;
}
/** Binary stdout preserves cat-file byte framing before text decoding. */
@@ -271,21 +286,22 @@ export function runGitCommandBytes(
args: string[],
options: GitCommandOptions,
): Promise<GitCommandResult<Buffer>> {
return executeGitCommand(args, options, (output) => output);
return executeGitCommand(args, options, (output) => output, {});
}
function executeGitCommand<Output>(
args: string[],
options: GitCommandOptions,
decode: (output: Buffer) => Output,
provenance?: string,
submission: GitCommandSubmission,
): Promise<GitCommandResult<Output>> {
const metricsState = submitGitCommandMetric(args, options.cwd);
const priority = submission.priority ?? gitCommandPriority.getStore() ?? "normal";
const metricsState = submitGitCommandMetric(args, options.cwd, priority);
const commandTrace = submitGitCommandTrace(args, options.cwd, {
active: gitProcessScheduler.activeCount,
pending: gitProcessScheduler.pendingCount,
});
const runtimeMetric = gitRuntimeMetrics.submit(getGitOperation(args), provenance);
const runtimeMetric = gitRuntimeMetrics.submit(getGitOperation(args), submission.provenance);
const startCommand = () => {
let releaseProcessSlot!: () => void;
const exited = new Promise<void>((resolve) => {
@@ -558,9 +574,7 @@ function executeGitCommand<Output>(
});
return { result: resultPromise, exited };
};
const promise = gitProcessScheduler.run(startCommand, {
priority: gitCommandPriority.getStore() ?? "normal",
});
const promise = gitProcessScheduler.run(startCommand, { priority });
gitRuntimeMetrics.observeLimiter(
gitProcessScheduler.activeCount,
gitProcessScheduler.pendingCount,
@@ -568,7 +582,8 @@ function executeGitCommand<Output>(
return promise;
}
export const runGitCommand: RunGitCommand = runGitCommandWithProvenance;
export const runGitCommand: RunGitCommand = (args, options) =>
executeGitCommand(args, options, decodeUtf8, {});
function formatGitCommand(args: string[]): string {
return ["git", ...args].join(" ");