From 48a1497fc0abb654dde264850d6bf4088ac34ac1 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Tue, 10 Nov 2026 10:21:13 +0700 Subject: [PATCH] Load the Changes diff quickly while the daemon reconciles worktrees (#9243) * 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. * 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. * fix(server): re-read the working tree once its watcher covers it The Linux observer no longer reports pre-existing files at startup, so an edit made between a workspace's first snapshot and its watcher subscribing reached no consumer. WorkspaceGit now runs one background working-tree read once observation setup completes with a live watcher. Demand refreshes keep the caller's Git lane instead of forcing normal, so user actions running at high priority are not demoted. Scope the populated-tree observer test to Linux (FSEvents replays recent writes) and update the observer benchmark test, which expected the removed startup batch. * fix(server): re-read checkout metadata once observation covers it A branch switch made before the Git metadata watcher subscribed left the cached branch stale. The post-coverage read now refreshes structure as well as the working tree, matching the refresh the Linux startup events used to trigger. --- docs/data-model.md | 8 + docs/file-observation.md | 2 +- .../scripts/measure-file-observer.test.ts | 8 +- .../src/server/diff/subscriptions.test.ts | 109 ++++++++++++ .../src/server/file-observer/index.test.ts | 28 +++ .../server/file-observer/internal/linux.ts | 29 ++-- ...it-service.observation.integration.test.ts | 163 ++++++++++++++++++ .../workspace-git-service.observation.test.ts | 44 +++-- .../workspace-git-service.primitive.test.ts | 4 +- .../src/server/workspace-git-service.test.ts | 20 ++- .../src/server/workspace-git-service.ts | 84 ++++++++- .../workspace-reconciliation-service.test.ts | 56 ++++++ .../workspace-reconciliation-service.ts | 11 +- .../src/utils/git-process-scheduler.test.ts | 58 +++++++ .../server/src/utils/git-process-scheduler.ts | 60 ++++--- packages/server/src/utils/run-git-command.ts | 53 ++++-- 16 files changed, 641 insertions(+), 96 deletions(-) diff --git a/docs/data-model.md b/docs/data-model.md index 518d3e03d1..aa0f201ce6 100644 --- a/docs/data-model.md +++ b/docs/data-model.md @@ -340,6 +340,14 @@ the number of Git processes that have started but not exited. Every Git command including initial workspace reads, filesystem-triggered refreshes, background checks, and explicit requests. +Commands wait in one of three lanes, drained in order: `high` (worktree creation and archive), +`normal` (everything a caller waits for, including diffs and explicit refreshes), and `background` +(workspace reconciliation, refreshes triggered by watchers, polls and fetches, and the fetches +themselves). Daemon-initiated fan-out must run in the background lane: one rescan over ~90 worktrees +queues over 1,000 commands, and in the normal lane every command a cold diff runs waited behind +them. Set the lane on the runner with `createRunGitCommand(provenance, { priority })`, or around a +whole operation with `runWithGitCommandPriority`. + Environment variables override `config.json`: | Environment variable | Setting | diff --git a/docs/file-observation.md b/docs/file-observation.md index d7d1d8f2f0..c2e68d83a9 100644 --- a/docs/file-observation.md +++ b/docs/file-observation.md @@ -8,7 +8,7 @@ Closing an observation must remain safe after its root is renamed or removed. Wo `unsubscribe()` is an awaited barrier. Once it resolves, no scan, reconciliation, queued event, native handle, or callback may remain. `updateIgnore()` is the equivalent barrier for excluded paths: events queued below a newly excluded root cannot be delivered after it resolves. Treat path disappearance while scanning, attaching, or receiving a watcher error as rename/archive churn. Surface permission and resource errors so WorkspaceGit can enter bounded polling. -Linux topology repair is trailing-debounced and scoped to the renamed path. A file rename needs one classification stat; a new or moved directory scans only its subtree. Full-tree scans are reserved for startup and ignore-set replacement. The initial Linux scan conservatively emits existing files discovered before their directory watcher was installed; WorkspaceGit batches these into one refresh. Benchmark this startup cost with pre-populated trees. +Linux topology repair is trailing-debounced and scoped to the renamed path. A file rename needs one classification stat; a new or moved directory scans only its subtree. Full-tree scans are reserved for startup and ignore-set replacement. A subscription reports changes after `subscribe()` resolves, not the files that already existed. macOS FSEvents can still replay writes made moments before the stream started. Consumers read their initial state after `subscribe()` resolves; WorkspaceGit re-reads its checkout once observation setup completes, because its first refresh runs in parallel with setup. The Linux startup scan therefore emits nothing; only a scoped reconciliation of new topology reports files found before their directory's watcher was installed. Startup creates made every cold diff compute and discard a first revision. Benchmark startup cost with pre-populated trees. Windows and macOS use one native recursive watcher plus an ignored-pruned, directory-indexed inventory. Native events deliver immediately. Named events trigger serialized shallow scans of their parent directories; known directory events also reconcile that subtree. This recovers concrete create and delete paths omitted by coalescing without walking the whole root for each edit. Classification is admission-controlled to 32 concurrent `fs.stat` calls, since `fs.stat` runs on the libuv threadpool the whole daemon shares. Startup and exclusion changes use an awaited full inventory barrier. Activity arms a full safety audit 30 seconds after it settles, with a 30-second minimum interval and a five-minute starvation bound; pathless native events also trigger an immediate shallow root scan. Observation fails into polling rather than retaining more than 250,000 file and directory entries per root. A repository that large is also expensive to poll, so degraded polling doubles its gap from 5 seconds to a 60-second ceiling while the workspace snapshot comes back unchanged, and resets to 5 seconds on the first change. Explicit reads still force a refresh; the ceiling only bounds how long an edit made outside Paseo stays undiscovered. diff --git a/packages/server/scripts/measure-file-observer.test.ts b/packages/server/scripts/measure-file-observer.test.ts index 345da462bb..a784a9fbdd 100644 --- a/packages/server/scripts/measure-file-observer.test.ts +++ b/packages/server/scripts/measure-file-observer.test.ts @@ -8,10 +8,10 @@ import { expect, test } from "vitest"; const execFileAsync = promisify(execFile); -// Linux delivers the existing inventory as one batch. Other backends have -// different startup semantics, so they do not exercise this regression. +// The Linux backend scans the existing inventory at startup without reporting +// it. Other backends have different startup semantics. test.skipIf(process.platform !== "linux").each(["healthy", "directory-churn"])( - "%s benchmark collects a startup batch larger than the argument limit", + "%s benchmark over a large pre-populated tree reports only later edits", async (mode) => { const temporaryRoot = await mkdtemp(join(tmpdir(), "paseo-watch-benchmark-test-")); try { @@ -47,7 +47,7 @@ test.skipIf(process.platform !== "linux").each(["healthy", "directory-churn"])( }, ); const measurement = JSON.parse(stdout); - expect(measurement.startupEvents).toBe(40_000); + expect(measurement.startupEvents).toBe(0); expect(measurement.deliveredEvents).toBeGreaterThan(measurement.startupEvents); expect(measurement.missedTrackedPaths).toBe(0); expect(measurement.ignoredEvents).toBe(0); diff --git a/packages/server/src/server/diff/subscriptions.test.ts b/packages/server/src/server/diff/subscriptions.test.ts index 69457d2be7..6f792a3311 100644 --- a/packages/server/src/server/diff/subscriptions.test.ts +++ b/packages/server/src/server/diff/subscriptions.test.ts @@ -6,6 +6,115 @@ import { expect, test } from "vitest"; import { DiffOpcode, type DiffFrame } from "@getpaseo/protocol/binary-frames/index"; import { DiffSubscriptions } from "./index.js"; import { createFileObserver } from "../file-observer/index.js"; +import { + configureGitProcessPolicy, + runGitCommand, + runWithGitCommandPriority, +} from "../../utils/run-git-command.js"; +import { + DEFAULT_GIT_PROCESS_POLICY, + resolveGitProcessPolicy, +} from "../../utils/git-process-scheduler.js"; + +async function createObservedWorktreeDiff(prefix: string) { + const cwd = await mkdtemp(join(tmpdir(), prefix)); + const git = (...args: string[]) => execFileSync("git", args, { cwd, encoding: "utf8" }).trim(); + const observer = createFileObserver(); + const subscriptions = new DiffSubscriptions({ + async requestWorkingTreeWatch(root, onChange) { + const watch = await observer.subscribe(root, () => onChange(), { + ignore: [join(root, ".git")], + }); + return { + repoRoot: root, + unsubscribe: async () => { + await watch.unsubscribe(); + }, + }; + }, + }); + git("init", "-b", "main"); + git("config", "user.email", "test@example.com"); + git("config", "user.name", "Test"); + for (const directory of ["src", "src/nested", "docs"]) { + await mkdir(join(cwd, directory), { recursive: true }); + for (let index = 0; index < 40; index++) + await writeFile(join(cwd, directory, `${index}.ts`), `export const value = ${index};\n`); + } + git("add", "."); + git("-c", "commit.gpgsign=false", "commit", "-m", "base"); + for (let index = 0; index < 10; index++) + await writeFile(join(cwd, "src", `${index}.ts`), `export const value = ${index + 100};\n`); + return { + cwd, + subscriptions, + compare: { + cwd, + base: { kind: "rev" as const, rev: "HEAD" }, + head: { kind: "worktree" as const }, + highlight: true, + }, + async close() { + await subscriptions.close(); + await observer.close(); + await rm(cwd, { recursive: true, force: true }); + }, + }; +} + +test("a cold worktree diff over existing files is complete at its first revision", async () => { + const fixture = await createObservedWorktreeDiff("diff-cold-revision-"); + const frames: DiffFrame[] = []; + try { + const subscription = await fixture.subscriptions.subscribe(fixture.compare, async (frame) => { + frames.push(frame); + }); + await subscription.start(); + await writeFile(join(fixture.cwd, "docs", "0.ts"), "export const value = 999;\n"); + await expect + .poll(() => frames.filter((frame) => frame.opcode === DiffOpcode.End).length, { + timeout: 10_000, + }) + .toBe(2); + expect( + frames.filter((frame) => frame.opcode === DiffOpcode.Begin).map((frame) => frame.revision), + ).toEqual([1, 2]); + await subscription.release(); + } finally { + await fixture.close(); + } +}, 30_000); + +test("a cold diff finishes ahead of a queued background Git burst", async () => { + const fixture = await createObservedWorktreeDiff("diff-cold-burst-"); + // The suite lifts the start-rate cap; the burst needs the daemon's real one. + configureGitProcessPolicy(DEFAULT_GIT_PROCESS_POLICY); + try { + let burstSettledAt = Number.POSITIVE_INFINITY; + // ~3 s of queued work at 64 starts/s, the shape of a reconciliation rescan. + const burst = runWithGitCommandPriority("background", () => + Promise.all( + Array.from({ length: 200 }, () => + runGitCommand(["rev-parse", "HEAD"], { cwd: fixture.cwd }), + ), + ), + ).then(() => { + burstSettledAt = performance.now(); + return undefined; + }); + let readyAt = Number.POSITIVE_INFINITY; + const subscription = await fixture.subscriptions.subscribe(fixture.compare, async (frame) => { + if (frame.opcode === DiffOpcode.End && frame.error === null) readyAt = performance.now(); + }); + await subscription.start(); + await burst; + expect(readyAt).toBeLessThan(burstSettledAt); + await subscription.release(); + } finally { + configureGitProcessPolicy(resolveGitProcessPolicy({ env: process.env })); + await fixture.close(); + } +}, 30_000); test("shares computation, highlights only the changed file, recovers refs and releases observations", async () => { const cwd = await mkdtemp(join(tmpdir(), "diff-subscriptions-")); diff --git a/packages/server/src/server/file-observer/index.test.ts b/packages/server/src/server/file-observer/index.test.ts index ccb451632a..c30e294817 100644 --- a/packages/server/src/server/file-observer/index.test.ts +++ b/packages/server/src/server/file-observer/index.test.ts @@ -47,6 +47,34 @@ test("observes nested files while pruning excluded directories", async () => { await subscription.unsubscribe(); }); +// macOS FSEvents can replay writes made just before the stream starts, so the +// startup contract is only exact on the Linux backend. +test.skipIf(process.platform !== "linux")( + "subscribing to a populated tree reports only later changes", + async () => { + const root = await createRoot(); + for (const directory of ["src", join("src", "nested"), "docs"]) { + await mkdir(join(root, directory), { recursive: true }); + await Promise.all( + Array.from({ length: 20 }, (_, index) => + writeFile(join(root, directory, `existing-${index}.txt`), `${index}`), + ), + ); + } + const events: FileChange[] = []; + const subscription = await subscribeToFileChanges(root, (error, batch) => { + expect(error).toBeNull(); + events.push(...batch); + }); + + const laterPath = join(root, "src", "nested", "later.txt"); + await writeFile(laterPath, "later"); + await expect.poll(() => events.map((event) => event.path)).toContain(laterPath); + expect(new Set(events.map((event) => event.path))).toEqual(new Set([laterPath])); + await subscription.unsubscribe(); + }, +); + test("covers files populated immediately inside a newly created directory", async () => { const root = await createRoot(); const events: FileChange[] = []; diff --git a/packages/server/src/server/file-observer/internal/linux.ts b/packages/server/src/server/file-observer/internal/linux.ts index d67517a006..dc5138430f 100644 --- a/packages/server/src/server/file-observer/internal/linux.ts +++ b/packages/server/src/server/file-observer/internal/linux.ts @@ -23,6 +23,11 @@ const RECONCILIATION_DELAY_MS = 50; const MAX_CONCURRENT_CLASSIFICATIONS = 32; const MAX_QUEUED_CLASSIFICATIONS = 16_384; +// "baseline" establishes coverage of what already exists when the subscription +// starts, so it reports nothing. "changes" scans topology that appeared while +// observed, so files found before their directory was covered are creates. +type ReconcileMode = "baseline" | "changes"; + interface DirectorySnapshot { directories: Set; filesDiscoveredBeforeCoverage: Set; @@ -62,7 +67,7 @@ class LinuxBackend implements ObservationBackend { async start(): Promise { this.watchDirectory(this.host.root); - await this.enqueue([this.host.root]); + await this.enqueue([this.host.root], "baseline"); } async updateIgnore(): Promise { @@ -75,7 +80,7 @@ class LinuxBackend implements ObservationBackend { } } this.cancelQueued(); - await this.enqueue([this.host.root]); + await this.enqueue([this.host.root], "changes"); await Promise.allSettled(inFlight); } @@ -159,15 +164,17 @@ class LinuxBackend implements ObservationBackend { this.timer = null; const scopes = [...this.pendingScopes]; this.pendingScopes.clear(); - void this.enqueue(scopes).catch((error: unknown) => this.host.fail(toError(error))); + void this.enqueue(scopes, "changes").catch((error: unknown) => + this.host.fail(toError(error)), + ); }, RECONCILIATION_DELAY_MS); this.timer.unref(); } - private enqueue(scopes: string[]): Promise { + private enqueue(scopes: string[], mode: ReconcileMode): Promise { this.queueDepth += 1; const result = this.reconcileTail - .then(() => this.run(scopes)) + .then(() => this.run(scopes, mode)) .finally(() => { this.queueDepth -= 1; }); @@ -175,12 +182,12 @@ class LinuxBackend implements ObservationBackend { return result; } - private async run(scopes: string[]): Promise { + private async run(scopes: string[], mode: ReconcileMode): Promise { if (!this.host.isActive()) return; this.inFlight = true; const startedAt = performance.now(); try { - await this.reconcile(this.paths.collapse(scopes)); + await this.reconcile(this.paths.collapse(scopes), mode); } catch (error) { this.host.metrics.reconciliationFailureCount += 1; throw error; @@ -197,15 +204,17 @@ class LinuxBackend implements ObservationBackend { } } - private async reconcile(scopes: string[]): Promise { + private async reconcile(scopes: string[], mode: ReconcileMode): Promise { for (const scope of scopes) { const snapshot = await this.scan(scope); if (!this.host.isActive()) return; for (const directory of snapshot.directories) { if (!this.watchSnapshotDirectory(directory)) snapshot.directories.delete(directory); } - for (const path of snapshot.filesDiscoveredBeforeCoverage) { - this.host.queueEvent("create", path); + if (mode === "changes") { + for (const path of snapshot.filesDiscoveredBeforeCoverage) { + this.host.queueEvent("create", path); + } } for (const [directory, watcher] of this.watchers) { const disappeared = diff --git a/packages/server/src/server/workspace-git-service.observation.integration.test.ts b/packages/server/src/server/workspace-git-service.observation.integration.test.ts index 0a5152e5d4..8480d0604d 100644 --- a/packages/server/src/server/workspace-git-service.observation.integration.test.ts +++ b/packages/server/src/server/workspace-git-service.observation.integration.test.ts @@ -1,4 +1,10 @@ import { flushGitCommandTrace } from "../utils/git-command-trace.js"; +import { + runWithGitCommandPriority, + startGitCommandMetrics, + stopGitCommandMetrics, + waitForGitCommandMetricsIdle, +} from "../utils/run-git-command.js"; import { createWorkspaceRuntimeStore } from "./workspace-runtime-store/index.js"; import { mkdirSync, mkdtempSync, readFileSync, realpathSync, rmSync, writeFileSync } from "node:fs"; import { execFileSync } from "node:child_process"; @@ -274,6 +280,8 @@ test("recursive observation updates tracked state and prunes ignored storms", as }), ).toBe(2); expect(workspaceRuntimeStore.get(repoDir)).not.toBeNull(); + // The initial read, then the checkout read that follows watcher coverage. + expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(2); expect(service.getMetrics()).toMatchObject({ workspaceObservationSetupInFlightCount: 0, workspaceRefreshInFlightCount: 0, @@ -856,3 +864,158 @@ test.skipIf(process.platform !== "linux")( }, 60_000, ); + +test("observation-driven refreshes run their Git work in the background lane", async () => { + const root = realpathSync(mkdtempSync(path.join(tmpdir(), "paseo-git-priority-"))); + const repo = path.join(root, "repo"); + mkdirSync(repo); + const git = (...args: string[]) => + execFileSync("git", ["-c", "commit.gpgsign=false", ...args], { cwd: repo, stdio: "pipe" }); + git("init", "-b", "main"); + git("config", "user.email", "test@example.com"); + git("config", "user.name", "Test"); + writeFileSync(path.join(repo, "README.md"), "base\n"); + git("add", "."); + git("commit", "-m", "base"); + const service = new WorkspaceGitServiceImpl({ + logger: pino({ enabled: false }), + paseoHome: path.join(root, "home"), + }); + const heads: Array = []; + const subscription = service.registerWorkspace({ cwd: repo }, (snapshot) => { + heads.push(snapshot.git.currentBranch); + }); + cleanup.push(async () => { + subscription.unsubscribe(); + await service.dispose(); + rmSync(root, { recursive: true, force: true }); + }); + // Registration, observation setup and the first refresh settle before the + // measured change. + startGitCommandMetrics(); + await service.getSnapshot(repo, { includeForge: false }); + await waitForGitCommandMetricsIdle({ quietMs: 500, timeoutMs: 10_000 }); + stopGitCommandMetrics(); + + startGitCommandMetrics(); + const deliveredBefore = heads.length; + git("switch", "-c", "feature"); + await expect.poll(() => heads.slice(deliveredBefore), { timeout: 10_000 }).toContain("feature"); + await waitForGitCommandMetricsIdle({ quietMs: 300, timeoutMs: 10_000 }); + const observed = stopGitCommandMetrics().submissions; + expect(observed.length).toBeGreaterThan(0); + expect(new Set(observed.map((command) => command.priority))).toEqual(new Set(["background"])); + + startGitCommandMetrics(); + await service.getSnapshot(repo, { force: true, reason: "test-demand", includeForge: false }); + await service.getCheckout(repo, { priority: "background" }); + await waitForGitCommandMetricsIdle({ quietMs: 0, timeoutMs: 10_000 }); + const demand = stopGitCommandMetrics().submissions; + const checkoutStart = demand.findIndex((command) => command.priority === "background"); + expect(checkoutStart).toBeGreaterThan(0); + expect(new Set(demand.slice(0, checkoutStart).map((command) => command.priority))).toEqual( + new Set(["normal"]), + ); + expect(new Set(demand.slice(checkoutStart).map((command) => command.priority))).toEqual( + new Set(["background"]), + ); + + startGitCommandMetrics(); + await runWithGitCommandPriority("high", () => + service.getSnapshot(repo, { force: true, reason: "test-user-action", includeForge: false }), + ); + await waitForGitCommandMetricsIdle({ quietMs: 0, timeoutMs: 10_000 }); + const userAction = stopGitCommandMetrics().submissions; + expect(userAction.length).toBeGreaterThan(0); + expect(new Set(userAction.map((command) => command.priority))).toEqual(new Set(["high"])); +}, 30_000); + +test("an edit made before the working-tree watcher starts still reaches the cached snapshot", async () => { + const root = realpathSync(mkdtempSync(path.join(tmpdir(), "paseo-git-coverage-gap-"))); + const repo = path.join(root, "repo"); + const trackedPath = path.join(repo, "src", "tracked.txt"); + mkdirSync(path.dirname(trackedPath), { recursive: true }); + const git = (...args: string[]) => + execFileSync("git", ["-c", "commit.gpgsign=false", ...args], { cwd: repo, stdio: "pipe" }); + git("init", "-b", "main"); + git("config", "user.email", "test@example.com"); + git("config", "user.name", "Test"); + writeFileSync(trackedPath, "base\n"); + git("add", "."); + git("commit", "-m", "base"); + const observer = createFileObserver(); + const workingTreeWatchGate = Promise.withResolvers(); + const service = new WorkspaceGitServiceImpl({ + logger: pino({ enabled: false }), + paseoHome: path.join(root, "home"), + deps: { + subscribe: async (directory, callback, options) => { + if (directory === repo) await workingTreeWatchGate.promise; + return observer.subscribe(directory, callback, options); + }, + }, + }); + const subscription = service.registerWorkspace({ cwd: repo }, () => {}); + cleanup.push(async () => { + workingTreeWatchGate.resolve(); + subscription.unsubscribe(); + await service.dispose(); + await observer.close(); + rmSync(root, { recursive: true, force: true }); + }); + + expect((await service.getSnapshot(repo, { includeForge: false })).git.isDirty).toBe(false); + writeFileSync(trackedPath, "edited before the watcher started\n"); + workingTreeWatchGate.resolve(); + + await expect + .poll(async () => (await service.getSnapshot(repo, { includeForge: false })).git.isDirty, { + timeout: 10_000, + }) + .toBe(true); +}, 30_000); + +test("a branch switch made before the Git metadata watcher starts still reaches the cached snapshot", async () => { + const root = realpathSync(mkdtempSync(path.join(tmpdir(), "paseo-git-metadata-gap-"))); + const repo = path.join(root, "repo"); + mkdirSync(repo); + const git = (...args: string[]) => + execFileSync("git", ["-c", "commit.gpgsign=false", ...args], { cwd: repo, stdio: "pipe" }); + git("init", "-b", "main"); + git("config", "user.email", "test@example.com"); + git("config", "user.name", "Test"); + writeFileSync(path.join(repo, "README.md"), "base\n"); + git("add", "."); + git("commit", "-m", "base"); + const observer = createFileObserver(); + const metadataWatchGate = Promise.withResolvers(); + const service = new WorkspaceGitServiceImpl({ + logger: pino({ enabled: false }), + paseoHome: path.join(root, "home"), + deps: { + subscribe: async (directory, callback, options) => { + if (directory !== repo) await metadataWatchGate.promise; + return observer.subscribe(directory, callback, options); + }, + }, + }); + const subscription = service.registerWorkspace({ cwd: repo }, () => {}); + cleanup.push(async () => { + metadataWatchGate.resolve(); + subscription.unsubscribe(); + await service.dispose(); + await observer.close(); + rmSync(root, { recursive: true, force: true }); + }); + + expect((await service.getSnapshot(repo, { includeForge: false })).git.currentBranch).toBe("main"); + git("switch", "-c", "feature"); + metadataWatchGate.resolve(); + + await expect + .poll( + async () => (await service.getSnapshot(repo, { includeForge: false })).git.currentBranch, + { timeout: 10_000 }, + ) + .toBe("feature"); +}, 30_000); 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 f866aa8dc6..c73e2de271 100644 --- a/packages/server/src/server/workspace-git-service.observation.test.ts +++ b/packages/server/src/server/workspace-git-service.observation.test.ts @@ -192,6 +192,8 @@ async function driveWatchRecoveryLadder( } } +// Registering a workspace reads the checkout twice: once at start and once +// when observation setup establishes watcher coverage. function createService( watcher: ReturnType, overrides?: Record, @@ -391,8 +393,9 @@ describe("WorkspaceGitService checkout observation", () => { expect(service.getMetrics().workspaceRefreshInFlightCount).toBe(0); }); - expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(1); - expect(getCheckoutStatus).toHaveBeenCalledTimes(1); + expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(2); + // The initial read, then one checkout read once the watcher covers the tree. + expect(getCheckoutStatus).toHaveBeenCalledTimes(2); expect(runGitCommand).toHaveBeenCalledTimes(2); expect(watcher.records.filter((record) => record.directory === REPO_CWD)).toHaveLength(1); expect(watcher.records.filter((record) => record.directory === GIT_DIR)).toHaveLength(1); @@ -480,7 +483,7 @@ describe("WorkspaceGitService checkout observation", () => { }); await flushPromises(); - expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(1); + expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(2); expect(getPullRequestStatus).not.toHaveBeenCalled(); diffSubscription.unsubscribe(); @@ -552,7 +555,7 @@ describe("WorkspaceGitService checkout observation", () => { ?.callback(null, [{ path: path.join(GIT_DIR, "packed-refs"), type: "update" }]); await vi.advanceTimersByTimeAsync(1_000); await vi.waitFor(() => { - expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(2); + expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(3); }); expect(listener).toHaveBeenCalledTimes(1); @@ -622,7 +625,7 @@ describe("WorkspaceGitService checkout observation", () => { expect(getCheckoutRefDerivedState).toHaveBeenCalledTimes(1); }); - expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(1); + expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(2); expect(listener).toHaveBeenCalledTimes(1); subscription.unsubscribe(); @@ -678,7 +681,7 @@ describe("WorkspaceGitService checkout observation", () => { await flushPromises(); await vi.advanceTimersByTimeAsync(1_000); - expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(1); + expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(2); expect(getCheckoutRefDerivedState).not.toHaveBeenCalled(); subscription.unsubscribe(); @@ -734,7 +737,7 @@ describe("WorkspaceGitService checkout observation", () => { }); await vi.advanceTimersByTimeAsync(1_000); await vi.waitFor(() => { - expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(2); + expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(3); }); subscription.unsubscribe(); @@ -785,7 +788,7 @@ describe("WorkspaceGitService checkout observation", () => { }); await vi.advanceTimersByTimeAsync(1_000); await vi.waitFor(() => { - expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(2); + expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(3); }); expect(getCheckoutRefDerivedState).not.toHaveBeenCalled(); @@ -836,7 +839,7 @@ describe("WorkspaceGitService checkout observation", () => { // An idle fetch count also holds before observation setup starts. expect(fetchStarted).toBe(true); expect(service.getMetrics().fetchInFlightCount).toBe(0); - expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(1); + expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(2); }); const repoWatcher = watcher.records.find((record) => record.directory === GIT_DIR)!; @@ -849,7 +852,7 @@ describe("WorkspaceGitService checkout observation", () => { }, ]); await vi.advanceTimersByTimeAsync(1_000); - expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(1); + expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(2); expect(getCheckoutRefDerivedState).not.toHaveBeenCalled(); repoWatcher.callback(null, [ @@ -859,7 +862,7 @@ describe("WorkspaceGitService checkout observation", () => { await vi.waitFor(() => { expect(getCheckoutRefDerivedState).toHaveBeenCalledTimes(1); }); - expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(1); + expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(2); subscription.unsubscribe(); await service.dispose(); @@ -994,7 +997,7 @@ describe("WorkspaceGitService checkout observation", () => { }); await vi.advanceTimersByTimeAsync(1_000); await vi.waitFor(() => { - expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(2); + expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(3); }); subscription.unsubscribe(); @@ -1036,14 +1039,14 @@ describe("WorkspaceGitService checkout observation", () => { expect(service.getMetrics().fetchInFlightCount).toBe(0); expect(runGitFetch).toHaveBeenCalledTimes(1); }); - expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(1); + expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(2); const repoWatcher = getWatcherRecordsForDirectory(watcher, GIT_DIR)[0]!; repoWatcher.callback(null, [ { path: path.join(GIT_DIR, "refs", "remotes", "origin", "main"), type: "update" }, ]); await vi.advanceTimersByTimeAsync(1_000); await vi.waitFor(() => { - expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(2); + expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(3); }); subscription.unsubscribe(); @@ -1607,7 +1610,8 @@ describe("WorkspaceGitService checkout observation", () => { expect.any(Function), expect.any(Object), ); - expect(getCheckoutStatus).toHaveBeenCalledTimes(1); + // The initial read, then one checkout read once the watcher covers the tree. + expect(getCheckoutStatus).toHaveBeenCalledTimes(2); expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0); expect(service.getMetrics().workspaceRefreshInFlightCount).toBe(0); }); @@ -1742,7 +1746,9 @@ describe("WorkspaceGitService checkout observation", () => { await vi.waitFor(() => { expect(workspaceRuntimeStore.get(REPO_CWD)).not.toBeNull(); expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0); + expect(service.getMetrics().workspaceRefreshInFlightCount).toBe(0); }); + const statusCallsBeforeError = getCheckoutStatus.mock.calls.length; const checkoutWatcher = watcher.records.find((record) => record.directory === REPO_CWD); expect(checkoutWatcher).toBeDefined(); @@ -1751,10 +1757,10 @@ describe("WorkspaceGitService checkout observation", () => { expect(service.getMetrics().watcherErrorCallbackCount).toBe(1); await vi.advanceTimersByTimeAsync(1_000); await vi.waitFor(() => { - expect(getCheckoutStatus).toHaveBeenCalledTimes(2); + expect(getCheckoutStatus).toHaveBeenCalledTimes(statusCallsBeforeError + 1); }); - expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(1); + expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(2); expect(service.getMetrics().workspaceRefreshQueuedCount).toBe(0); ignoredDirectories = "node_modules/\nbuild/\n"; @@ -2757,7 +2763,9 @@ describe("WorkspaceGitService checkout observation", () => { await vi.waitFor(() => { expect(workspaceRuntimeStore.get(REPO_CWD)).not.toBeNull(); expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0); + expect(service.getMetrics().workspaceRefreshInFlightCount).toBe(0); }); + const statusCallsBeforeEvent = getCheckoutStatus.mock.calls.length; isDirty = true; diffStat = { additions: 7, deletions: 3 }; @@ -2776,7 +2784,7 @@ describe("WorkspaceGitService checkout observation", () => { ); expect(diffListener).toHaveBeenCalledTimes(1); }); - expect(getCheckoutStatus).toHaveBeenCalledTimes(2); + expect(getCheckoutStatus).toHaveBeenCalledTimes(statusCallsBeforeEvent + 1); diffSubscription.unsubscribe(); 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 1b43026c7d..f9d4af8a9d 100644 --- a/packages/server/src/server/workspace-git-service.primitive.test.ts +++ b/packages/server/src/server/workspace-git-service.primitive.test.ts @@ -992,7 +992,9 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => { await flushPromises(); await vi.waitFor(() => expect(subscribe).toHaveBeenCalledTimes(2)); - expect(getCheckoutStatus).toHaveBeenCalledTimes(1); + // Setup itself does no snapshot read; once the watcher covers the tree, one + // checkout read catches changes made since the forced read. + await vi.waitFor(() => expect(getCheckoutStatus).toHaveBeenCalledTimes(2)); subscription.unsubscribe(); service.dispose(); diff --git a/packages/server/src/server/workspace-git-service.test.ts b/packages/server/src/server/workspace-git-service.test.ts index 96ca7c333d..9d91610cd8 100644 --- a/packages/server/src/server/workspace-git-service.test.ts +++ b/packages/server/src/server/workspace-git-service.test.ts @@ -313,6 +313,8 @@ function buildDefaultTestServiceDeps() { }; } +// Registering a workspace reads the checkout twice: once at start and once +// when observation setup establishes watcher coverage. function createService(options?: CreateServiceTestOptions) { const deps = { ...buildDefaultTestServiceDeps(), ...options }; deps.getCheckoutWorktreeState = @@ -1178,7 +1180,7 @@ describe("WorkspaceGitServiceImpl", () => { const first = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn()); const second = service.registerWorkspace({ cwd: secondCwd }, vi.fn()); await vi.waitFor(() => { - expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(2); + expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(4); expect(runGitFetch).toHaveBeenCalledTimes(1); expect(service.getMetrics().workspaceRefreshInFlightCount).toBe(0); }); @@ -1193,7 +1195,7 @@ describe("WorkspaceGitServiceImpl", () => { await flushPromises(); await vi.advanceTimersByTimeAsync(1_000); await vi.waitFor(() => { - expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(2); + expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(4); expect(getCheckoutRefDerivedState).toHaveBeenCalledTimes(2); }); @@ -1241,7 +1243,7 @@ describe("WorkspaceGitServiceImpl", () => { await vi.waitFor(() => { expect(runGitFetch).toHaveBeenCalledTimes(1); expect(service.getMetrics().fetchInFlightCount).toBe(0); - expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(2); + expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(4); }); getCheckoutSnapshotFacts.mockClear(); @@ -1303,7 +1305,7 @@ describe("WorkspaceGitServiceImpl", () => { ); await vi.waitFor( () => { - expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(2); + expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(4); expect(service.getMetrics()).toMatchObject({ repositoryWorkspaceLinkCount: 2, workspaceObservationSetupInFlightCount: 0, @@ -1320,7 +1322,7 @@ describe("WorkspaceGitServiceImpl", () => { }); await vi.advanceTimersByTimeAsync(1_000); await vi.waitFor(() => { - expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(4); + expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(6); }); expect(getCheckoutRefDerivedState).not.toHaveBeenCalled(); @@ -1360,7 +1362,7 @@ describe("WorkspaceGitServiceImpl", () => { const listener = vi.fn(); const subscription = service.registerWorkspace({ cwd: REPO_CWD }, listener); await vi.waitFor(() => { - expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(1); + expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(2); expect(service.getMetrics().fetchInFlightCount).toBe(1); }); @@ -1372,8 +1374,10 @@ describe("WorkspaceGitServiceImpl", () => { await vi.advanceTimersByTimeAsync(1_000); await vi.waitFor(() => { expect(getCheckoutRefDerivedState).toHaveBeenCalledTimes(1); - expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(2); - expect(getCheckoutShortstat).toHaveBeenCalledTimes(2); + expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(3); + // Initial read, the checkout read once the watcher covers the tree, + // then the fallback. + expect(getCheckoutShortstat).toHaveBeenCalledTimes(3); }); expect(listener.mock.lastCall?.[0].git.diffStat).toEqual({ additions: 2, deletions: 0 }); diff --git a/packages/server/src/server/workspace-git-service.ts b/packages/server/src/server/workspace-git-service.ts index 00c7084c7b..c086fa41be 100644 --- a/packages/server/src/server/workspace-git-service.ts +++ b/packages/server/src/server/workspace-git-service.ts @@ -48,6 +48,7 @@ import { runGitCommand, type RunGitCommand, } from "../utils/run-git-command.js"; +import type { GitProcessPriority } from "../utils/git-process-scheduler.js"; import { GitObservationPaths } from "./git-observation-paths.js"; import { branchNameFromRef } from "../utils/worktree-metadata.js"; import { listPaseoWorktrees, type PaseoWorktreeInfo } from "../utils/worktree.js"; @@ -209,7 +210,10 @@ export interface WorkspaceGitService { /** Publishes computed snapshot changes independently of workspace UI subscriptions. */ onSnapshotUpdated(listener: WorkspaceGitSnapshotUpdatedListener): WorkspaceGitSubscription; - getCheckout(cwd: string): Promise; + getCheckout( + cwd: string, + options?: WorkspaceGitCheckoutReadOptions, + ): Promise; getSnapshot( cwd: string, options?: WorkspaceGitSnapshotOptions, @@ -329,7 +333,28 @@ export type WorkspaceGitSnapshotOptions = reason: string; }; +/** + * Who is waiting for a refresh. "demand" refreshes answer a caller; "observation" + * refreshes react to watchers, polls and fetches and run in the background Git + * lane so they never queue in front of interactive Git work. + */ +type WorkspaceGitRefreshOrigin = "demand" | "observation"; + +export interface WorkspaceGitCheckoutReadOptions { + /** Git lane for the fresh checkout read; defaults to the caller's ambient lane. */ + priority?: GitProcessPriority; +} + +// A caller waiting on a merged refresh keeps it out of the background lane. +function mergeRefreshOrigins( + pending: WorkspaceGitRefreshOrigin, + request: WorkspaceGitRefreshOrigin, +): WorkspaceGitRefreshOrigin { + return pending === "demand" ? pending : request; +} + interface WorkspaceGitRefreshRequest { + origin: WorkspaceGitRefreshOrigin; force: boolean; refreshStructure: boolean; refreshWorktree: boolean; @@ -341,6 +366,7 @@ interface WorkspaceGitRefreshRequest { } interface ScheduledWorkspaceGitRefreshOptions { + origin: WorkspaceGitRefreshOrigin; force?: boolean; scope?: "refs" | "structure" | "worktree"; includeForge?: boolean; @@ -782,13 +808,19 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { return this.requestWorkspaceSnapshot(target, request); } - async getCheckout(cwd: string): Promise { + async getCheckout( + cwd: string, + options?: WorkspaceGitCheckoutReadOptions, + ): Promise { this.assertNotDisposed(); const normalizedCwd = resolve(cwd); const status = await this.deps.getCheckoutStatus(normalizedCwd, { paseoHome: this.paseoHome, worktreesRoot: this.worktreesRoot, logger: this.logger, + ...(options?.priority + ? { runGitCommand: createRunGitCommand("checkout-read", { priority: options.priority }) } + : {}), }); const cached = this.getCachedSnapshot(normalizedCwd); if (cached && cached.git.isGit !== status.isGit) { @@ -1009,6 +1041,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { await this.resumeWorkspaceIfFolderRestored(target); } await this.refreshWorkspaceTarget(target, { + origin: "demand", force: false, refreshStructure: true, refreshWorktree: true, @@ -1048,7 +1081,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { void this.resumeWorkspaceIfFolderRestored(target); return; } - this.scheduleWorkspaceRefresh(target, { queueIfBusy: false }); + this.scheduleWorkspaceRefresh(target, { origin: "demand", queueIfBusy: false }); } onWorkspaceStateMayHaveChanged(cwd: string): void { @@ -1064,6 +1097,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { return; } this.scheduleWorkspaceRefresh(target, { + origin: "demand", force: true, includeForge: true, reason: "external-state-change", @@ -1292,6 +1326,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { return; } void this.refreshWorkspaceTarget(target, { + origin: "demand", force: false, refreshStructure: true, refreshWorktree: true, @@ -1371,9 +1406,27 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { await this.ensureRepoTarget(target); if (this.isActiveObservedWorkspaceTarget(target)) { target.observationSetupComplete = true; + this.refreshAfterWorkingTreeCoverage(target); } } + // Watchers report only changes made after they subscribe, and the initial + // snapshot is read in parallel with observation setup, so an edit or branch + // switch in between reaches no watcher. Read the checkout once more now that + // every later change will. Degraded polling keeps its own cadence. + private refreshAfterWorkingTreeCoverage(target: WorkspaceGitTarget): void { + if (!target.workingTreeWatchTarget?.subscription) { + return; + } + void this.refreshWorkspaceTarget( + target, + this.buildScheduledRefreshRequest({ + origin: "observation", + reason: "observation-established", + }), + ); + } + private async getFactsForObservation(target: WorkspaceGitTarget): Promise { if (target.latestFacts) { return target.latestFacts; @@ -1762,6 +1815,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { target.latestFacts = null; target.folderMissing = false; await this.refreshWorkspaceTarget(target, { + origin: "observation", force: false, refreshStructure: true, refreshWorktree: true, @@ -1805,6 +1859,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { return false; } const changed = await this.refreshDegradedPollTarget(workspaceTarget, { + origin: "observation", force: false, refreshStructure: target.repoRoot === null, refreshWorktree: true, @@ -2286,7 +2341,11 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { for (const workspaceKey of target.workspaceKeys) { const workspaceTarget = this.workspaceTargets.get(workspaceKey); if (workspaceTarget) { - this.scheduleWorkspaceRefresh(workspaceTarget, { scope: "worktree", reason }); + this.scheduleWorkspaceRefresh(workspaceTarget, { + origin: "observation", + scope: "worktree", + reason, + }); } } } @@ -2869,6 +2928,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { } } this.scheduleWorkspaceRefresh(workspaceTarget, { + origin: "observation", scope, emitUnchanged: refresh.refreshBase, reason, @@ -2909,6 +2969,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { workingTreeTargets.add(workingTreeTarget); } return await this.refreshDegradedPollTarget(workspaceTarget, { + origin: "observation", force: false, refreshStructure: true, refreshWorktree: true, @@ -2940,7 +3001,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { private scheduleWorkspaceRefresh( targetOrCwd: WorkspaceGitTarget | string, - options?: ScheduledWorkspaceGitRefreshOptions, + options: ScheduledWorkspaceGitRefreshOptions, ): void { const target = typeof targetOrCwd === "string" @@ -3309,6 +3370,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { const force = options?.force === true; return { + origin: "demand", force, refreshStructure: true, refreshWorktree: true, @@ -3332,10 +3394,11 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { } private buildScheduledRefreshRequest( - options: ScheduledWorkspaceGitRefreshOptions | undefined, + options: ScheduledWorkspaceGitRefreshOptions, ): WorkspaceGitRefreshRequest { const scope = options?.scope; return { + origin: options.origin, force: options?.force === true, refreshStructure: scope !== "worktree" && scope !== "refs", refreshWorktree: scope !== "structure" && scope !== "refs", @@ -3366,6 +3429,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { request.movedRemoteRefs, ); return { + origin: mergeRefreshOrigins(pending.origin, request.origin), force, refreshStructure: pending.refreshStructure || request.refreshStructure, refreshWorktree: pending.refreshWorktree || request.refreshWorktree, @@ -3404,7 +3468,11 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { }); } try { - const runRefreshGitCommand = createRunGitCommand(`workspace-refresh:${request.reason}`); + // Demand refreshes keep the caller's lane, so a user action running at + // "high" is not demoted. + const runRefreshGitCommand = createRunGitCommand(`workspace-refresh:${request.reason}`, { + priority: request.origin === "observation" ? "background" : undefined, + }); const admittedSnapshot = await this.workspaceRefreshLimit(() => { if ( target.closed || @@ -3792,7 +3860,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { } }, }, - createRunGitCommand("background-fetch"), + createRunGitCommand("background-fetch", { priority: "background" }), ); } catch (error) { this.logger.warn( diff --git a/packages/server/src/server/workspace-reconciliation-service.test.ts b/packages/server/src/server/workspace-reconciliation-service.test.ts index dd6805c359..3d0057bcae 100644 --- a/packages/server/src/server/workspace-reconciliation-service.test.ts +++ b/packages/server/src/server/workspace-reconciliation-service.test.ts @@ -772,6 +772,62 @@ describe("WorkspaceReconciliationService", () => { expect(projects.get("p1")?.kind).toBe("git"); }); + test("reads sibling checkouts a few at a time in the background Git lane", async () => { + const projectRoot = "/reconcile/fan-out/root"; + const { projects, workspaces, projectRegistry, workspaceRegistry } = createTestRegistries(); + projects.set( + "p1", + createPersistedProjectRecord({ + projectId: "p1", + rootPath: projectRoot, + kind: "git", + displayName: "fan-out", + createdAt: timestamp, + updatedAt: timestamp, + }), + ); + const siblingCwds = Array.from({ length: 40 }, (_, index) => `${projectRoot}-${index}`); + for (const [index, cwd] of siblingCwds.entries()) { + workspaces.set( + `w${index}`, + createPersistedWorkspaceRecord({ + workspaceId: `w${index}`, + projectId: "p1", + cwd, + kind: "worktree", + displayName: `sibling-${index}`, + createdAt: timestamp, + updatedAt: timestamp, + }), + ); + } + const reads: Array<{ cwd: string; priority: string | undefined }> = []; + let inFlight = 0; + let maxInFlight = 0; + const service = new WorkspaceReconciliationService({ + projectRegistry, + workspaceRegistry, + logger: createTestLogger(), + inspectDirectory: async () => "directory", + workspaceGitService: { + async getCheckout(cwd, options) { + reads.push({ cwd, priority: options?.priority }); + inFlight += 1; + maxInFlight = Math.max(maxInFlight, inFlight); + await new Promise((resolve) => setTimeout(resolve, 1)); + inFlight -= 1; + return createCheckout(cwd, { isGit: true, currentBranch: "main", worktreeRoot: cwd }); + }, + }, + }); + + await service.runOnce(); + + expect(reads.map((read) => read.cwd).sort()).toEqual([projectRoot, ...siblingCwds].sort()); + expect(new Set(reads.map((read) => read.priority))).toEqual(new Set(["background"])); + expect(maxInFlight).toBe(4); + }); + test("deduplicates equivalent project and workspace paths across legacy duplicate projects", async () => { const projectRoot = realpathSync(mkdtempSync(path.join(tmpdir(), "reconcile-global-root-"))); const workspaceRoot = realpathSync( diff --git a/packages/server/src/server/workspace-reconciliation-service.ts b/packages/server/src/server/workspace-reconciliation-service.ts index a007dedb59..f1fe5c9444 100644 --- a/packages/server/src/server/workspace-reconciliation-service.ts +++ b/packages/server/src/server/workspace-reconciliation-service.ts @@ -1,6 +1,7 @@ import { watch as watchPath } from "node:fs"; import { stat } from "node:fs/promises"; import { setImmediate as yieldToLoop } from "node:timers/promises"; +import pLimit from "p-limit"; import type { ProjectCheckoutLitePayload } from "@getpaseo/protocol/messages"; import type pino from "pino"; import type { @@ -22,6 +23,9 @@ import { deriveProjectKey } from "./project-key.js"; const DEFAULT_RESCAN_INTERVAL_MS = 5 * 60_000; const DEFAULT_DEBOUNCE_MS = 100; const INSPECTION_CONCURRENCY = 2; +// Each checkout read runs ~15 Git commands. Reading every sibling at once +// queued ~1,300 commands per rescan on a project with ~90 worktrees. +const CHECKOUT_READ_CONCURRENCY = 4; const DEFAULT_INSPECTION_TIMEOUT_MS = 5000; export type ProjectUpdate = @@ -346,10 +350,11 @@ export class WorkspaceReconciliationService { changes: ReconciliationChange[], ): Promise { const checkoutReads: CachedCheckoutRead[] = []; + const checkoutReadLimit = pLimit(CHECKOUT_READ_CONCURRENCY); const readCheckout = (cwd: string): Promise => { const existing = checkoutReads.find((read) => areEquivalentPaths(read.cwd, cwd)); if (existing) return existing.checkout; - const checkout = this.readCheckout(cwd); + const checkout = checkoutReadLimit(() => this.readCheckout(cwd)); checkoutReads.push({ cwd, checkout }); return checkout; }; @@ -581,7 +586,9 @@ export class WorkspaceReconciliationService { mainRepoRoot: null, }; } - return this.workspaceGitService.getCheckout(cwd); + // Reconciliation converges in the background; it must not queue ahead of + // Git work a user is waiting for. + return this.workspaceGitService.getCheckout(cwd, { priority: "background" }); } private wakeInspectionWaiters(): void { diff --git a/packages/server/src/utils/git-process-scheduler.test.ts b/packages/server/src/utils/git-process-scheduler.test.ts index c8a72cd199..74ea99f273 100644 --- a/packages/server/src/utils/git-process-scheduler.test.ts +++ b/packages/server/src/utils/git-process-scheduler.test.ts @@ -68,6 +68,64 @@ 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("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 cd5575d79e..a1897b22af 100644 --- a/packages/server/src/utils/git-process-scheduler.ts +++ b/packages/server/src/utils/git-process-scheduler.ts @@ -16,7 +16,10 @@ export interface ScheduledGitProcess { exited: Promise; } -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) { @@ -44,11 +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 highPriorityQueue: Array<() => void> = []; - private readonly normalPriorityQueue: Array<() => void> = []; + private readonly queues: Record Promise>> = { + high: [], + normal: [], + background: [], + }; private admitted = 0; + private reservedStarts = 0; private running = 0; private waiting = 0; private drainScheduled = false; @@ -59,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 { @@ -76,17 +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(); - }); - }; - const queue = - options?.priority === "high" ? this.highPriorityQueue : this.normalPriorityQueue; - queue.push(admit); + this.queues[options?.priority ?? "normal"].push(() => this.execute(start, resolve, reject)); if (this.admitted === 0 && !this.drainScheduled) { this.drainOne(); } else { @@ -95,11 +96,12 @@ 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.highPriorityQueue.length === 0 && this.normalPriorityQueue.length === 0) { + if (this.drainScheduled || !this.canReserveStart()) { return; } this.drainScheduled = true; @@ -108,20 +110,28 @@ export class GitProcessScheduler { private drainOne(): void { this.drainScheduled = false; - if (this.admitted >= this.policy.maxProcessConcurrency) { - return; - } - const admit = this.highPriorityQueue.shift() ?? this.normalPriorityQueue.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, diff --git a/packages/server/src/utils/run-git-command.ts b/packages/server/src/utils/run-git-command.ts index e65416878f..e1f6fc718b 100644 --- a/packages/server/src/utils/run-git-command.ts +++ b/packages/server/src/utils/run-git-command.ts @@ -67,8 +67,17 @@ export type RunGitCommand = ( options: GitCommandOptions, ) => Promise; -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 { - 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> { - return executeGitCommand(args, options, (output) => output); + return executeGitCommand(args, options, (output) => output, {}); } function executeGitCommand( args: string[], options: GitCommandOptions, decode: (output: Buffer) => Output, - provenance?: string, + submission: GitCommandSubmission, ): Promise> { - 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((resolve) => { @@ -558,9 +574,7 @@ function executeGitCommand( }); 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( 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(" ");