From 7d093967efc5a509b7f2d0bec4e39ec957a5fa48 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Tue, 10 Nov 2026 02:58:52 +0100 Subject: [PATCH] fix(server): pick the git job when its start token fires Background jobs that already reserved a start-rate token no longer start ahead of normal work submitted after them. --- .../src/utils/git-process-scheduler.test.ts | 32 +++++++++++ .../server/src/utils/git-process-scheduler.ts | 57 +++++++++---------- 2 files changed, 59 insertions(+), 30 deletions(-) diff --git a/packages/server/src/utils/git-process-scheduler.test.ts b/packages/server/src/utils/git-process-scheduler.test.ts index dce2cdf8d8..74ea99f273 100644 --- a/packages/server/src/utils/git-process-scheduler.test.ts +++ b/packages/server/src/utils/git-process-scheduler.test.ts @@ -94,6 +94,38 @@ describe("GitProcessScheduler", () => { expect(started).toEqual([0, 6, 5, 2, 3, 4]); }); + test("starts normal work at the next start slot when background work saturates the rate", async () => { + vi.useFakeTimers(); + const scheduler = new GitProcessScheduler({ + maxProcessesPerSecond: 2, + maxProcessConcurrency: 8, + }); + const started: string[] = []; + const task = (name: string) => () => { + started.push(name); + const completed = Promise.resolve(); + return { result: completed, exited: completed }; + }; + const background = Array.from({ length: 8 }, (_, index) => + scheduler.run(task(`background-${index}`), { priority: "background" }), + ); + // Admission yields a real event-loop turn per process; let every + // background job reach the start-rate limiter. + for (let turn = 0; turn < 10; turn += 1) { + await new Promise((resolve) => scheduleImmediate(resolve)); + } + expect(started).toEqual(["background-0", "background-1"]); + + const normal = scheduler.run(task("normal")); + while (started.length < 3) { + await vi.advanceTimersToNextTimerAsync(); + } + + expect(started[2]).toBe("normal"); + await vi.runAllTimersAsync(); + await Promise.all([...background, normal]); + }); + test("yields to daemon liveness work between process starts", async () => { const scheduler = new GitProcessScheduler({ maxProcessesPerSecond: 100, diff --git a/packages/server/src/utils/git-process-scheduler.ts b/packages/server/src/utils/git-process-scheduler.ts index 8f3287d78d..a1897b22af 100644 --- a/packages/server/src/utils/git-process-scheduler.ts +++ b/packages/server/src/utils/git-process-scheduler.ts @@ -47,14 +47,19 @@ export function resolveGitProcessPolicy(input: { }; } +// Admission reserves a process slot and a start-rate token; the job runs only +// once its token fires. The job is chosen when the token fires, not when the +// slot is reserved, so work submitted while background jobs hold every reserved +// slot still takes the next start. export class GitProcessScheduler { private readonly startThrottled; - private readonly queues: Record void>> = { + private readonly queues: Record Promise>> = { high: [], normal: [], background: [], }; private admitted = 0; + private reservedStarts = 0; private running = 0; private waiting = 0; private drainScheduled = false; @@ -65,7 +70,7 @@ export class GitProcessScheduler { interval: 1_000, strict: true, }); - this.startThrottled = throttle((start: () => Promise) => start()); + this.startThrottled = throttle(() => this.startNext()); } get activeCount(): number { @@ -82,15 +87,7 @@ export class GitProcessScheduler { ): Promise { this.waiting += 1; return new Promise((resolve, reject) => { - const admit = () => { - void this.startThrottled(() => this.execute(start, resolve, reject)).catch((error) => { - this.waiting = Math.max(0, this.waiting - 1); - this.admitted = Math.max(0, this.admitted - 1); - reject(error); - this.scheduleDrain(); - }); - }; - this.queues[options?.priority ?? "normal"].push(admit); + this.queues[options?.priority ?? "normal"].push(() => this.execute(start, resolve, reject)); if (this.admitted === 0 && !this.drainScheduled) { this.drainOne(); } else { @@ -99,42 +96,42 @@ export class GitProcessScheduler { }); } + private canReserveStart(): boolean { + return this.admitted < this.policy.maxProcessConcurrency && this.waiting > this.reservedStarts; + } + private scheduleDrain(): void { - if (this.drainScheduled || this.admitted >= this.policy.maxProcessConcurrency) { - return; - } - if (!this.hasQueuedWork()) { + if (this.drainScheduled || !this.canReserveStart()) { 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.queues.high.shift() ?? this.queues.normal.shift() ?? this.queues.background.shift(); - if (!admit) { + if (!this.canReserveStart()) { return; } this.admitted += 1; - admit(); + this.reservedStarts += 1; + void this.startThrottled(); // 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 startNext(): Promise { + this.reservedStarts -= 1; + // Every reservation is backed by a queued job: jobs leave the queues only here. + const next = + this.queues.high.shift() ?? this.queues.normal.shift() ?? this.queues.background.shift(); + if (!next) { + throw new Error("Git process start fired without a queued process"); + } + return next(); + } + private async execute( start: () => ScheduledGitProcess, resolve: (value: T | PromiseLike) => void,