# Files changed · PR \#4727 · pingdotgg/t3code [View on GitCafe](https://git.cafe/pingdotgg/t3code/pull/4727/diffs) Repository: [pingdotgg/t3code](https://git.cafe/pingdotgg/t3code) Visibility: public [Fix Git ref refresh resource storms](https://git.cafe/pingdotgg/t3code/pull/4727.md) Version: 45 Head commit: 2c4c29ca4c1909ff5610c175714d3830e647bf34 Base commit: f2d2fb2f2444bcabc31bf906b5b334aee0d9d79c Comparison base: f2d2fb2f2444bcabc31bf906b5b334aee0d9d79c Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing. ## apps/mobile/src/connection/environment-cache-store.test.ts Status: modified; +36; −0 ```diff @@ -42,6 +42,12 @@ removed.push(id); values.delete(id); }), + clearCacheKind: (environmentId, kind) => + Effect.sync(() => { + for (const key of values.keys()) { + if (key.startsWith(`${environmentId}:${kind}:`)) values.delete(key); + } + }), clearEnvironmentCache: (environmentId) => Effect.sync(() => { for (const key of values.keys()) { @@ -80,6 +86,36 @@ }), ); + it.effect("removes one persisted VCS ref snapshot", () => + Effect.gen(function* () { + const memory = makeDatabase(); + const store = yield* make().pipe(Effect.provideService(MobileDatabase, memory.database)); + yield* store.saveVcsRefs(ENVIRONMENT_ID, "/repo", REFS); + + yield* store.removeVcsRefs(ENVIRONMENT_ID, "/repo"); + + expect(yield* store.loadVcsRefs(ENVIRONMENT_ID, "/repo")).toEqual(Option.none()); + expect(memory.removed).toContain(cacheId(ENVIRONMENT_ID, "vcs-refs", "/repo")); + }), + ); + + it.effect("clears every persisted VCS ref snapshot in one environment", () => + Effect.gen(function* () { + const memory = makeDatabase(); + const store = yield* make().pipe(Effect.provideService(MobileDatabase, memory.database)); + const otherEnvironmentId = EnvironmentId.make("environment-2"); + yield* store.saveVcsRefs(ENVIRONMENT_ID, "/repo", REFS); + yield* store.saveVcsRefs(ENVIRONMENT_ID, "/repo-worktree", REFS); + yield* store.saveVcsRefs(otherEnvironmentId, "/repo", REFS); + + yield* store.clearVcsRefs(ENVIRONMENT_ID); + + expect(yield* store.loadVcsRefs(ENVIRONMENT_ID, "/repo")).toEqual(Option.none()); + expect(yield* store.loadVcsRefs(ENVIRONMENT_ID, "/repo-worktree")).toEqual(Option.none()); + expect(yield* store.loadVcsRefs(otherEnvironmentId, "/repo")).toEqual(Option.some(REFS)); + }), + ); + it.effect("clears one environment without touching another", () => Effect.gen(function* () { const memory = makeDatabase(); ``` ## apps/mobile/src/connection/environment-cache-store.ts Status: modified; +10; −0 ```diff @@ -222,7 +222,17 @@ .saveCache(environmentId, "vcs-refs", cwd, VCS_REFS_CACHE_SCHEMA_VERSION, payload) .pipe(Effect.mapError(mapDatabaseError("save-vcs-refs"))); }, + ), + removeVcsRefs: Effect.fn("MobileEnvironmentCache.removeVcsRefs")((environmentId, cwd) => + database + .removeCache(environmentId, "vcs-refs", cwd) + .pipe(Effect.mapError(mapDatabaseError("remove-vcs-refs"))), ), + clearVcsRefs: Effect.fn("MobileEnvironmentCache.clearVcsRefs")((environmentId) => + database + .clearCacheKind(environmentId, "vcs-refs") + .pipe(Effect.mapError(mapDatabaseError("clear-vcs-refs"))), + ), clear: Effect.fn("MobileEnvironmentCache.clear")((environmentId) => database .clearEnvironmentCache(environmentId) ``` ## apps/mobile/src/persistence/mobile-database.ts Status: modified; +17; −0 ```diff @@ -46,6 +46,7 @@ "load-cache", "save-cache", "remove-cache", + "clear-cache-kind", "clear-environment-cache", "clear-all-caches", "inspect-caches", @@ -203,6 +204,10 @@ kind: ClientCacheKind, cacheKey: string, ) => Effect.Effect; + readonly clearCacheKind: ( + environmentId: EnvironmentId, + kind: ClientCacheKind, + ) => Effect.Effect; readonly clearEnvironmentCache: ( environmentId: EnvironmentId, ) => Effect.Effect; @@ -322,6 +327,17 @@ catch: databaseError("remove-cache"), }).pipe(Effect.asVoid), ), + clearCacheKind: Effect.fn("MobileDatabase.clearCacheKind")((environmentId, kind) => + Effect.tryPromise({ + try: () => + database.runAsync( + "DELETE FROM client_cache WHERE environment_id = ? AND kind = ?", + environmentId, + kind, + ), + catch: databaseError("clear-cache-kind"), + }).pipe(Effect.asVoid), + ), clearEnvironmentCache: Effect.fn("MobileDatabase.clearEnvironmentCache")((environmentId) => Effect.tryPromise({ try: () => @@ -392,6 +408,7 @@ loadCache: () => fail, saveCache: () => fail, removeCache: () => fail, + clearCacheKind: () => fail, clearEnvironmentCache: () => fail, clearAllCaches: fail, inspectCaches: fail, ``` ## apps/server/src/git/GitManager.ts Status: modified; +1; −1 ```diff @@ -1794,7 +1794,7 @@ resolvePullRequestWorktreeLocalBranchName(pullRequestWithRemoteInfo); const findLocalHeadBranch = Effect.fn("findLocalHeadBranch")(function* (cwd: string) { - const result = yield* gitCore.listRefs({ cwd }); + const result = yield* gitCore.listRefs({ cwd, refresh: true }); const localBranch = result.refs.find( (branch) => !branch.isRemote && branch.name === localPullRequestBranch, ); ``` ## apps/server/src/vcs/GitVcsDriverCore.test.ts Status: modified; +471; −2 ```diff @@ -1,18 +1,22 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it, describe } from "@effect/vitest"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; import * as PlatformError from "effect/PlatformError"; +import * as Ref from "effect/Ref"; import * as Scope from "effect/Scope"; import * as Sink from "effect/Sink"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; import { GitCommandError } from "@t3tools/contracts"; import { ServerConfig } from "../config.ts"; -import { splitNullSeparatedGitStdoutPaths } from "./GitVcsDriverCore.ts"; +import { makeGitVcsDriverCore, splitNullSeparatedGitStdoutPaths } from "./GitVcsDriverCore.ts"; import * as GitVcsDriver from "./GitVcsDriver.ts"; const ServerConfigLayer = ServerConfig.layerTest(process.cwd(), { @@ -38,6 +42,21 @@ getOutputFd: () => Stream.empty, }); +const makeSuccessfulHandle = (stdout: string) => + ChildProcessSpawner.makeHandle({ + pid: ChildProcessSpawner.ProcessId(1), + exitCode: Effect.succeed(ChildProcessSpawner.ExitCode(0)), + isRunning: Effect.succeed(false), + kill: () => Effect.void, + unref: Effect.succeed(Effect.void), + stdin: Sink.drain, + stdout: Stream.encodeText(Stream.make(stdout)), + stderr: Stream.empty, + all: Stream.empty, + getInputFd: () => Sink.drain, + getOutputFd: () => Stream.empty, + }); + const makeTmpDir = ( prefix = "git-vcs-driver-test-", ): Effect.Effect => @@ -129,11 +148,434 @@ assert.deepStrictEqual(commands, [ { args: ["status", "--porcelain=2", "--branch"], lcAll: "C" }, { args: ["rev-parse", "--abbrev-ref", "HEAD"], lcAll: "C" }, - { args: ["branch", "--no-color", "--no-column"], lcAll: "C" }, + { args: ["rev-parse", "--git-common-dir"], lcAll: "C" }, ]); }).pipe(Effect.provide(layer)); }); +it.effect("coalesces concurrent ref pages into one repository snapshot", () => + Effect.scoped( + Effect.gen(function* () { + const delegate = yield* ChildProcessSpawner.ChildProcessSpawner; + const spawnedArgs = yield* Ref.make>>([]); + const firstWorktreeScanStarted = yield* Deferred.make(); + const remoteNamesScanCompleted = yield* Deferred.make(); + const delayFirstWorktreeScan = yield* Ref.make(true); + const countingSpawner = ChildProcessSpawner.make((command) => + Effect.gen(function* () { + if (!ChildProcess.isStandardCommand(command)) { + return yield* Effect.die("expected a standard Git command"); + } + yield* Ref.update(spawnedArgs, (current) => [...current, command.args]); + const isWorktreeScan = + command.args.includes("worktree") && command.args.includes("--porcelain"); + const shouldDelay = + isWorktreeScan && (yield* Ref.getAndSet(delayFirstWorktreeScan, false)); + if (shouldDelay) { + yield* Deferred.succeed(firstWorktreeScanStarted, undefined); + yield* Effect.sleep("8 seconds"); + } + const handle = yield* delegate.spawn(command); + const isRemoteNamesScan = + command.args.length === 3 && + command.args[0] === "--git-dir" && + command.args[2] === "remote"; + return isRemoteNamesScan + ? ChildProcessSpawner.makeHandle({ + ...handle, + exitCode: handle.exitCode.pipe( + Effect.tap(() => Deferred.succeed(remoteNamesScanCompleted, undefined)), + ), + }) + : handle; + }), + ); + const driver = yield* makeGitVcsDriverCore().pipe( + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, countingSpawner), + ); + const cwd = yield* makeTmpDir(); + const runGit = (args: ReadonlyArray) => + driver.execute({ + operation: "GitVcsDriver.test.coalescedListRefs", + cwd, + args, + timeoutMs: 10_000, + }); + + yield* driver.initRepo({ cwd }); + yield* runGit(["config", "user.email", "test@test.com"]); + yield* runGit(["config", "user.name", "Test"]); + yield* writeTextFile(cwd, "README.md", "# test\n"); + yield* runGit(["add", "."]); + yield* runGit(["commit", "-m", "initial commit"]); + yield* Ref.set(spawnedArgs, []); + + const initialRequest = yield* driver + .listRefs({ cwd, refresh: true, limit: 100 }) + .pipe(Effect.forkChild({ startImmediately: true })); + yield* Deferred.await(firstWorktreeScanStarted); + yield* Deferred.await(remoteNamesScanCompleted); + yield* TestClock.adjust("6 seconds"); + const laterRequests = yield* Effect.all( + Array.from({ length: 30 }, (_, index) => + driver.listRefs({ + cwd, + refresh: true, + query: `missing-${index}`, + limit: 100, + }), + ), + { concurrency: "unbounded" }, + ).pipe(Effect.forkChild({ startImmediately: true })); + yield* TestClock.adjust("2 seconds"); + yield* Fiber.join(initialRequest); + yield* Fiber.join(laterRequests); + yield* driver.listRefs({ cwd, cursor: 1, limit: 100 }); + + const firstSnapshotCommands = yield* Ref.get(spawnedArgs); + const snapshotRefScans = firstSnapshotCommands.filter( + (args) => + args.includes("for-each-ref") && + args.includes("refs/heads") && + args.includes("refs/remotes"), + ); + const worktreeScans = firstSnapshotCommands.filter( + (args) => args.includes("worktree") && args.includes("--porcelain"), + ); + assert.equal(snapshotRefScans.length, 1); + assert.equal(worktreeScans.length, 1); + + yield* driver.createRef({ cwd, refName: "feature/cache-invalidation" }); + const refreshed = yield* driver.listRefs({ cwd, limit: 100 }); + assert.equal( + refreshed.refs.some((ref) => ref.name === "feature/cache-invalidation"), + true, + ); + const allCommands = yield* Ref.get(spawnedArgs); + assert.equal( + allCommands.filter( + (args) => + args.includes("for-each-ref") && + args.includes("refs/heads") && + args.includes("refs/remotes"), + ).length, + 2, + ); + }), + ).pipe(Effect.provide(ServerConfigLayer.pipe(Layer.provideMerge(NodeServices.layer)))), +); + +it.effect("retries an in-flight ref snapshot invalidated by a mutation", () => + Effect.scoped( + Effect.gen(function* () { + const delegate = yield* ChildProcessSpawner.ChildProcessSpawner; + const firstWorktreeScanStarted = yield* Deferred.make(); + const firstRefScanCompleted = yield* Deferred.make(); + const releaseFirstWorktreeScan = yield* Deferred.make(); + const delayFirstWorktreeScan = yield* Ref.make(true); + const refScans = yield* Ref.make(0); + const coordinatingSpawner = ChildProcessSpawner.make((command) => + Effect.gen(function* () { + if (!ChildProcess.isStandardCommand(command)) { + return yield* Effect.die("expected a standard Git command"); + } + const isWorktreeScan = + command.args.includes("worktree") && command.args.includes("--porcelain"); + if (isWorktreeScan && (yield* Ref.getAndSet(delayFirstWorktreeScan, false))) { + yield* Deferred.succeed(firstWorktreeScanStarted, undefined); + yield* Deferred.await(releaseFirstWorktreeScan); + } + const handle = yield* delegate.spawn(command); + const isRefScan = + command.args.includes("for-each-ref") && + command.args.includes("refs/heads") && + command.args.includes("refs/remotes"); + if (!isRefScan) return handle; + const scan = yield* Ref.updateAndGet(refScans, (count) => count + 1); + return scan === 1 + ? ChildProcessSpawner.makeHandle({ + ...handle, + exitCode: handle.exitCode.pipe( + Effect.tap(() => Deferred.succeed(firstRefScanCompleted, undefined)), + ), + }) + : handle; + }), + ); + const driver = yield* makeGitVcsDriverCore().pipe( + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, coordinatingSpawner), + ); + const cwd = yield* makeTmpDir(); + yield* initRepoWithCommit(cwd).pipe(Effect.provideService(GitVcsDriver.GitVcsDriver, driver)); + + const inFlight = yield* driver + .listRefs({ cwd, refresh: true, limit: 100 }) + .pipe(Effect.forkChild({ startImmediately: true })); + yield* Deferred.await(firstWorktreeScanStarted); + yield* Deferred.await(firstRefScanCompleted); + + yield* driver.createRef({ cwd, refName: "feature/during-refresh" }); + yield* Deferred.succeed(releaseFirstWorktreeScan, undefined); + + const refs = yield* Fiber.join(inFlight); + assert.isTrue(refs.refs.some((ref) => ref.name === "feature/during-refresh")); + assert.equal(yield* Ref.get(refScans), 2); + }), + ).pipe(Effect.provide(ServerConfigLayer.pipe(Layer.provideMerge(NodeServices.layer)))), +); + +it.effect("invalidates a ref snapshot when a mutation fails after changing Git", () => + Effect.scoped( + Effect.gen(function* () { + const delegate = yield* ChildProcessSpawner.ChildProcessSpawner; + const partiallyFailingSpawner = ChildProcessSpawner.make((command) => + Effect.gen(function* () { + if (!ChildProcess.isStandardCommand(command)) { + return yield* Effect.die("expected a standard Git command"); + } + if (command.args[0] === "branch" && command.args[1] === "feature/partial-failure") { + const handle = yield* delegate.spawn(command); + yield* handle.exitCode; + return makeNonRepositoryHandle(); + } + return yield* delegate.spawn(command); + }), + ); + const driver = yield* makeGitVcsDriverCore().pipe( + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, partiallyFailingSpawner), + ); + const cwd = yield* makeTmpDir(); + yield* initRepoWithCommit(cwd).pipe(Effect.provideService(GitVcsDriver.GitVcsDriver, driver)); + yield* driver.listRefs({ cwd, refresh: true }); + + yield* driver.createRef({ cwd, refName: "feature/partial-failure" }).pipe(Effect.flip); + + const refs = yield* driver.listRefs({ cwd }); + assert.isTrue(refs.refs.some((ref) => ref.name === "feature/partial-failure")); + }), + ).pipe(Effect.provide(ServerConfigLayer.pipe(Layer.provideMerge(NodeServices.layer)))), +); + +it.effect("fails a ref snapshot when for-each-ref exits unsuccessfully", () => + Effect.scoped( + Effect.gen(function* () { + const delegate = yield* ChildProcessSpawner.ChildProcessSpawner; + const snapshotAttempts = yield* Ref.make(0); + const failingSnapshotSpawner = ChildProcessSpawner.make((command) => + Effect.gen(function* () { + if (!ChildProcess.isStandardCommand(command)) { + return yield* Effect.die("expected a standard Git command"); + } + if (command.args.includes("for-each-ref")) { + yield* Ref.update(snapshotAttempts, (count) => count + 1); + return makeNonRepositoryHandle(); + } + return yield* delegate.spawn(command); + }), + ); + const driver = yield* makeGitVcsDriverCore().pipe( + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, failingSnapshotSpawner), + ); + const cwd = yield* makeTmpDir(); + yield* initRepoWithCommit(cwd).pipe(Effect.provideService(GitVcsDriver.GitVcsDriver, driver)); + + const error = yield* driver.listRefs({ cwd, refresh: true }).pipe(Effect.flip); + + assert.deepInclude(error, { + _tag: "GitCommandError", + operation: "GitVcsDriver.listRefs.snapshotRefs", + detail: "Git ref snapshot enumeration failed.", + exitCode: 128, + }); + assert.equal(yield* Ref.get(snapshotAttempts), 1); + }), + ).pipe(Effect.provide(ServerConfigLayer.pipe(Layer.provideMerge(NodeServices.layer)))), +); + +it.effect("marks the current branch when worktree metadata is unavailable", () => + Effect.scoped( + Effect.gen(function* () { + const delegate = yield* ChildProcessSpawner.ChildProcessSpawner; + const incompleteMetadataSpawner = ChildProcessSpawner.make((command) => + Effect.gen(function* () { + if (!ChildProcess.isStandardCommand(command)) { + return yield* Effect.die("expected a standard Git command"); + } + const isWorktreeRoot = + command.args.includes("rev-parse") && command.args.includes("--show-toplevel"); + const isWorktreeList = + command.args.includes("worktree") && command.args.includes("--porcelain"); + if (isWorktreeRoot || isWorktreeList) { + return makeNonRepositoryHandle(); + } + return yield* delegate.spawn(command); + }), + ); + const driver = yield* makeGitVcsDriverCore().pipe( + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, incompleteMetadataSpawner), + ); + const cwd = yield* makeTmpDir(); + const { initialBranch } = yield* initRepoWithCommit(cwd).pipe( + Effect.provideService(GitVcsDriver.GitVcsDriver, driver), + ); + + const refs = yield* driver.listRefs({ cwd, refresh: true }); + + assert.isTrue(refs.isRepo); + assert.isTrue(refs.refs.find((ref) => ref.name === initialBranch)?.current); + }), + ).pipe(Effect.provide(ServerConfigLayer.pipe(Layer.provideMerge(NodeServices.layer)))), +); + +it.effect("ignores worktree metadata for directories that no longer exist", () => + Effect.scoped( + Effect.gen(function* () { + const delegate = yield* ChildProcessSpawner.ChildProcessSpawner; + const missingWorktreePath = "/missing/deleted-worktree"; + const staleWorktreeSpawner = ChildProcessSpawner.make((command) => + Effect.gen(function* () { + if (!ChildProcess.isStandardCommand(command)) { + return yield* Effect.die("expected a standard Git command"); + } + const isWorktreeList = + command.args.includes("worktree") && command.args.includes("--porcelain"); + if (isWorktreeList) { + return makeSuccessfulHandle( + `worktree ${missingWorktreePath}\0HEAD deadbeef\0branch refs/heads/stale-worktree\0\0`, + ); + } + return yield* delegate.spawn(command); + }), + ); + const driver = yield* makeGitVcsDriverCore().pipe( + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, staleWorktreeSpawner), + ); + const cwd = yield* makeTmpDir(); + yield* initRepoWithCommit(cwd).pipe(Effect.provideService(GitVcsDriver.GitVcsDriver, driver)); + yield* git(cwd, ["branch", "stale-worktree"]).pipe( + Effect.provideService(GitVcsDriver.GitVcsDriver, driver), + ); + + const refs = yield* driver.listRefs({ cwd, refresh: true }); + + assert.equal(refs.refs.find((ref) => ref.name === "stale-worktree")?.worktreePath, null); + }), + ).pipe(Effect.provide(ServerConfigLayer.pipe(Layer.provideMerge(NodeServices.layer)))), +); + +it.effect("refreshes the current branch after an external checkout", () => + Effect.scoped( + Effect.gen(function* () { + const driver = yield* GitVcsDriver.GitVcsDriver; + const cwd = yield* makeTmpDir(); + const { initialBranch } = yield* initRepoWithCommit(cwd); + yield* git(cwd, ["branch", "external-checkout"]); + + const initialRefs = yield* driver.listRefs({ cwd, refresh: true }); + assert.isTrue(initialRefs.refs.find((ref) => ref.name === initialBranch)?.current); + + // Raw execute intentionally bypasses the driver's mutation invalidation, + // matching a checkout performed by another process. + yield* driver.execute({ + operation: "GitVcsDriver.test.externalCheckout", + cwd, + args: ["checkout", "external-checkout"], + timeoutMs: 10_000, + }); + yield* TestClock.adjust("6 seconds"); + + const refreshedRefs = yield* driver.listRefs({ cwd, refresh: true }); + assert.isTrue(refreshedRefs.refs.find((ref) => ref.name === "external-checkout")?.current); + assert.isFalse(refreshedRefs.refs.find((ref) => ref.name === initialBranch)?.current); + }), + ).pipe(Effect.provide(TestLayer)), +); + +it.effect("backs off failed upstream refreshes across linked worktrees", () => + Effect.scoped( + Effect.gen(function* () { + const delegate = yield* ChildProcessSpawner.ChildProcessSpawner; + const fetchAttempts = yield* Ref.make(0); + const failingFetchSpawner = ChildProcessSpawner.make((command) => + Effect.gen(function* () { + if (!ChildProcess.isStandardCommand(command)) { + return yield* Effect.die("expected a standard Git command"); + } + if (command.args.includes("fetch") && command.args.includes("--quiet")) { + yield* Ref.update(fetchAttempts, (count) => count + 1); + return makeNonRepositoryHandle(); + } + return yield* delegate.spawn(command); + }), + ); + const driver = yield* makeGitVcsDriverCore().pipe( + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, failingFetchSpawner), + ); + const fileSystem = yield* FileSystem.FileSystem; + const cwd = yield* makeTmpDir(); + const remote = yield* makeTmpDir("git-vcs-driver-remote-"); + const worktreesRoot = yield* makeTmpDir("git-vcs-driver-worktrees-"); + const pathService = yield* Path.Path; + const worktreePath = pathService.join(worktreesRoot, "linked"); + const runGit = (workingDirectory: string, args: ReadonlyArray) => + driver.execute({ + operation: "GitVcsDriver.test.upstreamRefreshBackoff", + cwd: workingDirectory, + args, + timeoutMs: 10_000, + }); + + yield* driver.initRepo({ cwd }); + yield* runGit(cwd, ["config", "user.email", "test@test.com"]); + yield* runGit(cwd, ["config", "user.name", "Test"]); + yield* writeTextFile(cwd, "README.md", "# test\n"); + yield* runGit(cwd, ["add", "."]); + yield* runGit(cwd, ["commit", "-m", "initial commit"]); + const initialBranch = (yield* runGit(cwd, ["branch", "--show-current"])).stdout.trim(); + yield* runGit(remote, ["init", "--bare"]); + yield* runGit(cwd, ["remote", "add", "origin", remote]); + yield* runGit(cwd, ["push", "-u", "origin", initialBranch]); + yield* runGit(cwd, ["worktree", "add", "-b", "feature/linked", worktreePath]); + yield* runGit(worktreePath, [ + "branch", + "--set-upstream-to", + `origin/${initialBranch}`, + "feature/linked", + ]); + const rootCommonDir = (yield* runGit(cwd, ["rev-parse", "--git-common-dir"])).stdout.trim(); + const linkedCommonDir = (yield* runGit(worktreePath, [ + "rev-parse", + "--git-common-dir", + ])).stdout.trim(); + assert.equal( + yield* fileSystem.realPath(pathService.resolve(cwd, rootCommonDir)), + yield* fileSystem.realPath(pathService.resolve(worktreePath, linkedCommonDir)), + ); + yield* Ref.set(fetchAttempts, 0); + + yield* driver.statusDetailsRemote(cwd); + yield* driver.statusDetailsRemote(worktreePath); + assert.equal(yield* Ref.get(fetchAttempts), 1); + + yield* TestClock.adjust("29 seconds"); + yield* driver.statusDetailsRemote(worktreePath); + assert.equal(yield* Ref.get(fetchAttempts), 1); + + yield* TestClock.adjust("1 second"); + yield* driver.statusDetailsRemote(cwd); + assert.equal(yield* Ref.get(fetchAttempts), 2); + + yield* TestClock.adjust("59 seconds"); + yield* driver.statusDetailsRemote(worktreePath); + assert.equal(yield* Ref.get(fetchAttempts), 2); + + yield* TestClock.adjust("1 second"); + yield* driver.statusDetailsRemote(cwd); + assert.equal(yield* Ref.get(fetchAttempts), 3); + }), + ).pipe(Effect.provide(ServerConfigLayer.pipe(Layer.provideMerge(NodeServices.layer)))), +); + it.layer(TestLayer)("GitVcsDriver core integration", (it) => { describe("process environment", () => { it.effect("preserves the caller locale for general Git subprocesses", () => @@ -661,6 +1103,33 @@ }); describe("worktree operations", () => { + it.effect("preserves newline characters in worktree paths when listing refs", () => + Effect.gen(function* () { + const cwd = yield* makeTmpDir(); + yield* initRepoWithCommit(cwd); + const worktreesRoot = yield* makeTmpDir("git-vcs-driver-worktrees-"); + const fileSystem = yield* FileSystem.FileSystem; + const pathService = yield* Path.Path; + const worktreePath = pathService.join(worktreesRoot, "linked\nworktree"); + const driver = yield* GitVcsDriver.GitVcsDriver; + + yield* git(cwd, ["worktree", "add", "-b", "feature/newline-path", worktreePath]); + + const refs = yield* driver.listRefs({ cwd, refresh: true }); + const listedPath = refs.refs.find( + (ref) => ref.name === "feature/newline-path", + )?.worktreePath; + + if (typeof listedPath !== "string") { + return assert.fail("expected the linked branch to include its worktree path"); + } + assert.equal( + yield* fileSystem.realPath(listedPath), + yield* fileSystem.realPath(worktreePath), + ); + }), + ); + it.effect("creates and removes a worktree for a new refName", () => Effect.gen(function* () { const cwd = yield* makeTmpDir(); ``` ## apps/server/src/vcs/GitVcsDriverCore.ts Status: modified; +526; −311 ```diff @@ -50,8 +50,17 @@ const STATUS_UPSTREAM_REFRESH_INTERVAL = Duration.seconds(15); const STATUS_UPSTREAM_REFRESH_TIMEOUT = Duration.seconds(5); -const STATUS_UPSTREAM_REFRESH_FAILURE_COOLDOWN = Duration.seconds(5); +const STATUS_UPSTREAM_REFRESH_FAILURE_BASE_COOLDOWN = Duration.seconds(30); +const STATUS_UPSTREAM_REFRESH_FAILURE_MAX_COOLDOWN = Duration.minutes(15); const STATUS_UPSTREAM_REFRESH_CACHE_CAPACITY = 2_048; +const REPOSITORY_PATHS_CACHE_CAPACITY = 2_048; +const REPOSITORY_PATHS_CACHE_TTL = Duration.minutes(10); +const REPOSITORY_PATHS_REFRESH_COALESCE_TTL = Duration.seconds(5); +const NON_REPOSITORY_PATHS_CACHE_TTL = Duration.seconds(1); +const LIST_REFS_SNAPSHOT_CACHE_CAPACITY = 64; +const LIST_REFS_SNAPSHOT_CACHE_TTL = Duration.minutes(2); +const LIST_REFS_REFRESH_COALESCE_TTL = Duration.seconds(5); +const LIST_REFS_REFRESH_FAILURE_COOLDOWN = Duration.seconds(30); const STATUS_UPSTREAM_REFRESH_ENV = Object.freeze({ GCM_INTERACTIVE: "never", GIT_ASKPASS: "", @@ -95,6 +104,35 @@ remoteName: string; }> {} +function statusUpstreamRefreshFailureCooldown(consecutiveFailures: number): Duration.Duration { + const exponent = Math.max(0, consecutiveFailures - 1); + const cooldownMs = + Duration.toMillis(STATUS_UPSTREAM_REFRESH_FAILURE_BASE_COOLDOWN) * Math.pow(2, exponent); + return Duration.min(Duration.millis(cooldownMs), STATUS_UPSTREAM_REFRESH_FAILURE_MAX_COOLDOWN); +} + +class GitRefsSnapshotCacheKey extends Data.Class<{ + gitCommonDir: string; + epoch: number; +}> {} + +class GitRefsRefreshCacheKey extends Data.Class<{ + gitCommonDir: string; + generation: number; +}> {} + +interface GitRepositoryPaths { + readonly gitCommonDir: string; + readonly worktreeRoot: string | null; + readonly currentBranch: string | null; +} + +interface GitRefsSnapshot { + readonly localBranches: ReadonlyArray; + readonly remoteBranches: ReadonlyArray; + readonly hasPrimaryRemote: boolean; +} + interface ExecuteGitOptions { stdin?: string | undefined; timeoutMs?: number | undefined; @@ -161,21 +199,6 @@ return filePath.length > 0 ? filePath : null; } -function parseBranchLine(line: string): { name: string; current: boolean } | null { - const trimmed = line.trim(); - if (trimmed.length === 0) return null; - - const name = trimmed.replace(/^[*+]\s+/, ""); - // Exclude symbolic refs like: "origin/HEAD -> origin/main". - // Exclude detached HEAD pseudo-refs like: "(HEAD detached at origin/main)". - if (name.includes(" -> ") || name.startsWith("(")) return null; - - return { - name, - current: trimmed.startsWith("* "), - }; -} - function filterBranchesForListQuery( refs: ReadonlyArray, query?: string, @@ -210,6 +233,37 @@ }; } +function parseWorktreeBranchPaths(stdout: string): ReadonlyMap { + const worktreePaths = new Map(); + let currentPath: string | null = null; + let currentBranch: string | null = null; + let currentPrunable = false; + + const flush = () => { + if (currentPath !== null && currentBranch !== null && !currentPrunable) { + worktreePaths.set(currentBranch, currentPath); + } + currentPath = null; + currentBranch = null; + currentPrunable = false; + }; + + for (const field of stdout.split("\0")) { + if (field === "") { + flush(); + } else if (field.startsWith("worktree ")) { + currentPath = field.slice("worktree ".length); + } else if (field.startsWith("branch refs/heads/")) { + currentBranch = field.slice("branch refs/heads/".length); + } else if (field === "prunable" || field.startsWith("prunable ")) { + currentPrunable = true; + } + } + flush(); + + return worktreePaths; +} + function splitNullSeparatedPaths(input: string, truncated: boolean): string[] { const parts = input.split("\0"); if (parts.length === 0) return []; @@ -937,35 +991,183 @@ fetchCwd, ["--git-dir", gitCommonDir, "fetch", "--quiet", "--no-tags", remoteName], { - allowNonZeroExit: true, env: STATUS_UPSTREAM_REFRESH_ENV, + fallbackErrorDetail: "Background Git fetch exited with a non-zero status.", timeoutMs: Duration.toMillis(STATUS_UPSTREAM_REFRESH_TIMEOUT), }, ).pipe(Effect.asVoid); }; + const resolveRepositoryPathsUncached = Effect.fn("resolveRepositoryPathsUncached")(function* ( + cwd: string, + ) { + const commonDirResult = yield* executeGitWithStableDiagnostics( + "GitVcsDriver.resolveRepositoryPaths.commonDir", + cwd, + ["rev-parse", "--git-common-dir"], + { + timeoutMs: 5_000, + allowNonZeroExit: true, + }, + ); + if (commonDirResult.exitCode !== 0) { + const stderr = commonDirResult.stderr.trim(); + if (isNonRepositoryGitStderr(stderr)) { + return null; + } + return yield* new GitCommandError({ + ...gitCommandContext({ + operation: "GitVcsDriver.resolveRepositoryPaths.commonDir", + cwd, + args: ["rev-parse", "--git-common-dir"], + }), + detail: "Failed to resolve the Git common directory.", + exitCode: commonDirResult.exitCode, + stdoutLength: commonDirResult.stdout.length, + stderrLength: commonDirResult.stderr.length, + }); + } + + const commonDirOutput = commonDirResult.stdout.trim(); + const resolvedGitCommonDir = path.isAbsolute(commonDirOutput) + ? path.normalize(commonDirOutput) + : path.resolve(cwd, commonDirOutput); + const gitCommonDir = yield* fileSystem + .realPath(resolvedGitCommonDir) + .pipe(Effect.orElseSucceed(() => resolvedGitCommonDir)); + const [worktreeRootResult, currentBranchResult] = yield* Effect.all( + [ + executeGit( + "GitVcsDriver.resolveRepositoryPaths.worktreeRoot", + cwd, + ["rev-parse", "--show-toplevel"], + { + timeoutMs: 5_000, + allowNonZeroExit: true, + }, + ), + executeGit( + "GitVcsDriver.resolveRepositoryPaths.currentBranch", + cwd, + ["symbolic-ref", "--quiet", "--short", "HEAD"], + { + timeoutMs: 5_000, + allowNonZeroExit: true, + }, + ), + ], + { concurrency: 2 }, + ); + const worktreeRootOutput = worktreeRootResult.stdout.trim(); + const worktreeRoot = + worktreeRootResult.exitCode === 0 && worktreeRootOutput.length > 0 + ? path.normalize( + path.isAbsolute(worktreeRootOutput) + ? worktreeRootOutput + : path.resolve(cwd, worktreeRootOutput), + ) + : null; + const currentBranchOutput = currentBranchResult.stdout.trim(); + const currentBranch = + currentBranchResult.exitCode === 0 && currentBranchOutput.length > 0 + ? currentBranchOutput + : null; + + return { + gitCommonDir, + worktreeRoot, + currentBranch, + } satisfies GitRepositoryPaths; + }); + + const repositoryPathsCache = yield* Cache.makeWith( + (cwd: string) => resolveRepositoryPathsUncached(cwd), + { + capacity: REPOSITORY_PATHS_CACHE_CAPACITY, + timeToLive: Exit.match({ + onSuccess: (repositoryPaths) => + repositoryPaths === null ? NON_REPOSITORY_PATHS_CACHE_TTL : REPOSITORY_PATHS_CACHE_TTL, + onFailure: () => Duration.zero, + }), + }, + ); + const repositoryPathsRefreshCache = yield* Cache.makeWith( + (cwd: string) => + Cache.invalidate(repositoryPathsCache, cwd).pipe( + Effect.andThen(Cache.get(repositoryPathsCache, cwd)), + ), + { + capacity: REPOSITORY_PATHS_CACHE_CAPACITY, + timeToLive: Exit.match({ + onSuccess: (repositoryPaths) => + repositoryPaths === null + ? NON_REPOSITORY_PATHS_CACHE_TTL + : REPOSITORY_PATHS_REFRESH_COALESCE_TTL, + onFailure: () => Duration.zero, + }), + }, + ); + const normalizeRepositoryPathsCacheKey = (cwd: string) => path.normalize(path.resolve(cwd)); + const resolveRepositoryPaths = (cwd: string, refresh = false) => { + const cacheKey = normalizeRepositoryPathsCacheKey(cwd); + return Cache.get(refresh ? repositoryPathsRefreshCache : repositoryPathsCache, cacheKey); + }; + const resolveGitCommonDir = Effect.fn("resolveGitCommonDir")(function* (cwd: string) { + const repositoryPaths = yield* resolveRepositoryPaths(cwd); + if (repositoryPaths !== null) { + return repositoryPaths.gitCommonDir; + } + return yield* new GitCommandError({ + ...gitCommandContext({ + operation: "GitVcsDriver.resolveGitCommonDir", - const gitCommonDir = yield* runGitStdout("GitVcsDriver.resolveGitCommonDir", cwd, [ - "rev-parse", + cwd, + args: ["rev-parse", "--git-common-dir"], - "--git-common-dir", - ]).pipe(Effect.map((stdout) => stdout.trim())); - return path.isAbsolute(gitCommonDir) ? gitCommonDir : path.resolve(cwd, gitCommonDir); + }), + detail: "Cannot resolve a Git common directory outside a repository.", + }); }); + const statusRemoteRefreshFailureCounts = new Map(); + const statusRemoteRefreshFailureKey = (cacheKey: StatusRemoteRefreshCacheKey) => + `${cacheKey.gitCommonDir}\0${cacheKey.remoteName}`; + const recordStatusRemoteRefreshFailure = (cacheKey: StatusRemoteRefreshCacheKey) => { + const key = statusRemoteRefreshFailureKey(cacheKey); + const nextCount = (statusRemoteRefreshFailureCounts.get(key) ?? 0) + 1; + statusRemoteRefreshFailureCounts.delete(key); + statusRemoteRefreshFailureCounts.set(key, nextCount); + if (statusRemoteRefreshFailureCounts.size > STATUS_UPSTREAM_REFRESH_CACHE_CAPACITY) { + const oldestKey = statusRemoteRefreshFailureCounts.keys().next().value; + if (oldestKey !== undefined) { + statusRemoteRefreshFailureCounts.delete(oldestKey); + } + } + }; + const clearStatusRemoteRefreshFailures = (cacheKey: StatusRemoteRefreshCacheKey) => { + statusRemoteRefreshFailureCounts.delete(statusRemoteRefreshFailureKey(cacheKey)); + }; const refreshStatusRemoteCacheEntry = Effect.fn("refreshStatusRemoteCacheEntry")(function* ( cacheKey: StatusRemoteRefreshCacheKey, ) { + return yield* fetchRemoteForStatus(cacheKey.gitCommonDir, cacheKey.remoteName).pipe( + Effect.tap(() => Effect.sync(() => clearStatusRemoteRefreshFailures(cacheKey))), + Effect.tapError(() => Effect.sync(() => recordStatusRemoteRefreshFailure(cacheKey))), + Effect.as(true as const), + ); - yield* fetchRemoteForStatus(cacheKey.gitCommonDir, cacheKey.remoteName); - return true as const; }); const statusRemoteRefreshCache = yield* Cache.makeWith(refreshStatusRemoteCacheEntry, { capacity: STATUS_UPSTREAM_REFRESH_CACHE_CAPACITY, - // Keep successful refreshes warm and briefly back off failed refreshes to avoid retry storms. - timeToLive: (exit) => + // A failed background fetch is intentionally cached and exponentially + // backed off. Status reads swallow this failure and use the last fetched + // refs, so repeated thread mounts cannot turn a slow or unavailable remote + // into a repository-wide Git subprocess storm. + timeToLive: (exit, cacheKey) => Exit.isSuccess(exit) ? STATUS_UPSTREAM_REFRESH_INTERVAL + : statusUpstreamRefreshFailureCooldown( + statusRemoteRefreshFailureCounts.get(statusRemoteRefreshFailureKey(cacheKey)) ?? 1, - : STATUS_UPSTREAM_REFRESH_FAILURE_COOLDOWN, + ), }); const refreshStatusUpstreamIfStale = Effect.fn("refreshStatusUpstreamIfStale")(function* ( @@ -1273,42 +1475,6 @@ }; }); - const readBranchRecency = Effect.fn("readBranchRecency")(function* (cwd: string) { - const branchRecency = yield* executeGit( - "GitVcsDriver.readBranchRecency", - cwd, - [ - "for-each-ref", - "--format=%(refname:short)%09%(committerdate:unix)", - "refs/heads", - "refs/remotes", - ], - { - timeoutMs: 15_000, - allowNonZeroExit: true, - }, - ); - - const branchLastCommit = new Map(); - if (branchRecency.exitCode !== 0) { - return branchLastCommit; - } - - for (const line of branchRecency.stdout.split("\n")) { - if (line.length === 0) { - continue; - } - const [name, lastCommitRaw] = line.split("\t"); - if (!name) { - continue; - } - const lastCommit = Number.parseInt(lastCommitRaw ?? "0", 10); - branchLastCommit.set(name, Number.isFinite(lastCommit) ? lastCommit : 0); - } - - return branchLastCommit; - }); - const readStatusDetailsLocal = Effect.fn("readStatusDetailsLocal")(function* (cwd: string) { const statusResult = yield* executeGitWithStableDiagnostics( "GitVcsDriver.statusDetails.status", @@ -1987,257 +2153,283 @@ Effect.map((trimmed) => (trimmed.length > 0 ? trimmed : null)), ); + const readGitRefsSnapshot = Effect.fn("readGitRefsSnapshot")(function* (gitCommonDir: string) { + const fetchCwd = + path.basename(gitCommonDir) === ".git" ? path.dirname(gitCommonDir) : gitCommonDir; + const gitDirArgs = ["--git-dir", gitCommonDir] as const; + const [refsResult, defaultRefResult, worktreeListResult, remoteNamesResult] = yield* Effect.all( + [ + executeGitWithStableDiagnostics( + "GitVcsDriver.listRefs.snapshotRefs", + fetchCwd, + [ + ...gitDirArgs, + "for-each-ref", + "--format=%(refname)%09%(committerdate:unix)%09%(symref)", + "refs/heads", + "refs/remotes", + ], + { + timeoutMs: 30_000, + maxOutputBytes: 16 * 1024 * 1024, + fallbackErrorDetail: "Git ref snapshot enumeration failed.", + }, + ), + executeGit( - const listRefs: GitVcsDriver.GitVcsDriver["Service"]["listRefs"] = Effect.fn("listRefs")( - function* (input) { + "GitVcsDriver.listRefs.defaultRef", + fetchCwd, - const branchRecencyPromise = readBranchRecency(input.cwd).pipe( - Effect.orElseSucceed(() => new Map()), - ); - const localBranchResult = yield* executeGitWithStableDiagnostics( - "GitVcsDriver.listRefs.branchNoColor", - input.cwd, - ["branch", "--no-color", "--no-column"], - { - timeoutMs: 10_000, + [...gitDirArgs, "symbolic-ref", "refs/remotes/origin/HEAD"], + { + timeoutMs: 5_000, + allowNonZeroExit: true, + }, + ), + executeGit( + "GitVcsDriver.listRefs.worktreeList", + fetchCwd, + [...gitDirArgs, "worktree", "list", "--porcelain", "-z"], + { + timeoutMs: 30_000, + allowNonZeroExit: true, + maxOutputBytes: 16 * 1024 * 1024, + }, + ), + executeGit("GitVcsDriver.listRefs.remoteNames", fetchCwd, [...gitDirArgs, "remote"], { + timeoutMs: 5_000, allowNonZeroExit: true, - }, - ).pipe( - Effect.catchTags({ - GitCommandError: (error) => - isMissingGitCwdError(error) - ? Effect.succeed({ - exitCode: ChildProcessSpawner.ExitCode(128), - stdout: "", - stderr: "fatal: not a git repository", - stdoutTruncated: false, - stderrTruncated: false, - }) - : Effect.fail(error), }), + ], + { concurrency: 2 }, + ); + + const remoteNames = + remoteNamesResult.exitCode === 0 ? parseRemoteNames(remoteNamesResult.stdout) : []; + if (remoteNamesResult.exitCode !== 0 && remoteNamesResult.stderr.trim().length > 0) { + yield* Effect.logWarning( + `GitVcsDriver.listRefs: remote name lookup returned code ${remoteNamesResult.exitCode} for ${gitCommonDir}: ${remoteNamesResult.stderr.trim()}. Falling back to an empty remote name list.`, ); + } + const defaultBranch = + defaultRefResult.exitCode === 0 + ? defaultRefResult.stdout.trim().replace(/^refs\/remotes\/origin\//, "") + : null; + const parsedWorktreeEntries = + worktreeListResult.exitCode === 0 + ? [...parseWorktreeBranchPaths(worktreeListResult.stdout)].map( + ([branchName, worktreePath]) => + [branchName, path.normalize(path.resolve(worktreePath))] as const, + ) + : []; + const existingWorktreeEntries = yield* Effect.filter( + parsedWorktreeEntries, + ([, worktreePath]) => + fileSystem.stat(worktreePath).pipe( + Effect.as(true), + Effect.orElseSucceed(() => false), + ), + { concurrency: 16 }, + ); + const worktreeMap = new Map(existingWorktreeEntries); + const localBranches: Array<{ readonly ref: VcsRef; readonly lastCommit: number }> = []; + const remoteBranches: Array<{ readonly ref: VcsRef; readonly lastCommit: number }> = []; - if (localBranchResult.exitCode !== 0) { + for (const line of refsResult.stdout.split("\n")) { + if (line.length === 0) continue; + const [fullRefName, lastCommitRaw, symbolicTarget] = line.split("\t"); + if (!fullRefName || symbolicTarget) continue; - const stderr = localBranchResult.stderr.trim(); + const parsedLastCommit = Number.parseInt(lastCommitRaw ?? "0", 10); - if (isNonRepositoryGitStderr(stderr)) { - return { - refs: [], - isRepo: false, - hasPrimaryRemote: false, - nextCursor: null, - totalCount: 0, + const lastCommit = Number.isFinite(parsedLastCommit) ? parsedLastCommit : 0; + + if (fullRefName.startsWith("refs/heads/")) { + const name = fullRefName.slice("refs/heads/".length); + localBranches.push({ + ref: { + name, + current: false, + isRemote: false, + isDefault: name === defaultBranch, + worktreePath: worktreeMap.get(name) ?? null, + }, + lastCommit, - }; - } - return yield* new GitCommandError({ - ...gitCommandContext({ - operation: "GitVcsDriver.listRefs", - cwd: input.cwd, - args: ["branch", "--no-color", "--no-column"], - }), - detail: "Git branch listing failed.", - exitCode: localBranchResult.exitCode, - stdoutLength: localBranchResult.stdout.length, - stderrLength: localBranchResult.stderr.length, }); + continue; } + if (!fullRefName.startsWith("refs/remotes/")) continue; + const name = fullRefName.slice("refs/remotes/".length); + const parsedRemoteRef = parseRemoteRefWithRemoteNames(name, remoteNames); + const remoteBranch: VcsRef = { + name, + current: false, + isRemote: true, + isDefault: + defaultBranch !== null && + parsedRemoteRef?.remoteName === "origin" && + parsedRemoteRef.branchName === defaultBranch, + worktreePath: null, + ...(parsedRemoteRef ? { remoteName: parsedRemoteRef.remoteName } : {}), + }; - const remoteBranchResultEffect = executeGit( - "GitVcsDriver.listRefs.remoteBranches", - input.cwd, - ["branch", "--no-color", "--no-column", "--remotes"], - { - timeoutMs: 10_000, - allowNonZeroExit: true, - }, - ).pipe( - Effect.catchTags({ - GitCommandError: (error) => - Effect.logWarning( - "Git remote ref lookup failed; falling back to an empty remote ref list.", - { - operation: error.operation, + remoteBranches.push({ ref: remoteBranch, lastCommit }); + } - command: error.command, - cwd: error.cwd, - detail: error.detail, - cause: error, - }, - ).pipe( - Effect.as({ - exitCode: ChildProcessSpawner.ExitCode(1), - stdout: "", - stderr: "", - stdoutTruncated: false, - stderrTruncated: false, - } satisfies GitVcsDriver.ExecuteGitResult), - ), - }), - ); - const remoteNamesResultEffect = executeGit( - "GitVcsDriver.listRefs.remoteNames", - input.cwd, - ["remote"], - { - timeoutMs: 5_000, - allowNonZeroExit: true, - }, - ).pipe( - Effect.catchTags({ - GitCommandError: (error) => - Effect.logWarning( - "Git remote name lookup failed; falling back to an empty remote name list.", - { - operation: error.operation, - command: error.command, - cwd: error.cwd, - detail: error.detail, - cause: error, - }, - ).pipe( - Effect.as({ + const byRecencyThenName = ( - exitCode: ChildProcessSpawner.ExitCode(1), - stdout: "", - stderr: "", - stdoutTruncated: false, - stderrTruncated: false, - } satisfies GitVcsDriver.ExecuteGitResult), - ), - }), - ); + left: { readonly ref: VcsRef; readonly lastCommit: number }, + right: { readonly ref: VcsRef; readonly lastCommit: number }, + ) => + left.lastCommit !== right.lastCommit + ? right.lastCommit - left.lastCommit + : left.ref.name.localeCompare(right.ref.name); + return { + localBranches: localBranches.toSorted(byRecencyThenName).map(({ ref }) => ref), + remoteBranches: remoteBranches.toSorted(byRecencyThenName).map(({ ref }) => ref), - const [defaultRef, worktreeList, remoteBranchResult, remoteNamesResult, branchLastCommit] = + hasPrimaryRemote: remoteNames.includes("origin"), + } satisfies GitRefsSnapshot; + }); - yield* Effect.all( - [ - executeGit( - "GitVcsDriver.listRefs.defaultRef", - input.cwd, - ["symbolic-ref", "refs/remotes/origin/HEAD"], - { - timeoutMs: 5_000, - allowNonZeroExit: true, - }, - ), - executeGit( - "GitVcsDriver.listRefs.worktreeList", - input.cwd, - ["worktree", "list", "--porcelain"], - { - timeoutMs: 5_000, - allowNonZeroExit: true, - }, - ), - remoteBranchResultEffect, - remoteNamesResultEffect, - branchRecencyPromise, - ], - { concurrency: "unbounded" }, - ); + const listRefsEpochByCommonDir = new Map(); + let listRefsEpochSequence = 0; + const bumpListRefsEpoch = (gitCommonDir: string): number => { + const nextEpoch = ++listRefsEpochSequence; + listRefsEpochByCommonDir.delete(gitCommonDir); + listRefsEpochByCommonDir.set(gitCommonDir, nextEpoch); + if (listRefsEpochByCommonDir.size > LIST_REFS_SNAPSHOT_CACHE_CAPACITY) { - const remoteNames = - remoteNamesResult.exitCode === 0 ? parseRemoteNames(remoteNamesResult.stdout) : []; - if (remoteBranchResult.exitCode !== 0 && remoteBranchResult.stderr.trim().length > 0) { - yield* Effect.logWarning( - `GitVcsDriver.listRefs: remote refName lookup returned code ${remoteBranchResult.exitCode} for ${input.cwd}: ${remoteBranchResult.stderr.trim()}. Falling back to an empty remote refName list.`, - ); + const oldestKey = listRefsEpochByCommonDir.keys().next().value; + if (oldestKey !== undefined) { + listRefsEpochByCommonDir.delete(oldestKey); } + } + return nextEpoch; + }; + const listRefsGenerationByCommonDir = new Map(); - if (remoteNamesResult.exitCode !== 0 && remoteNamesResult.stderr.trim().length > 0) { - yield* Effect.logWarning( - `GitVcsDriver.listRefs: remote name lookup returned code ${remoteNamesResult.exitCode} for ${input.cwd}: ${remoteNamesResult.stderr.trim()}. Falling back to an empty remote name list.`, + let listRefsGenerationSequence = 0; + const setListRefsGeneration = (gitCommonDir: string, generation: number): number => { - ); + listRefsGenerationByCommonDir.delete(gitCommonDir); + listRefsGenerationByCommonDir.set(gitCommonDir, generation); + if (listRefsGenerationByCommonDir.size > LIST_REFS_SNAPSHOT_CACHE_CAPACITY) { + const oldestKey = listRefsGenerationByCommonDir.keys().next().value; + if (oldestKey !== undefined) { + listRefsGenerationByCommonDir.delete(oldestKey); } - - const defaultBranch = - defaultRef.exitCode === 0 - ? defaultRef.stdout.trim().replace(/^refs\/remotes\/origin\//, "") - : null; - - const worktreeMap = new Map(); + } + return generation; + }; + const currentListRefsGeneration = (gitCommonDir: string): number => { - if (worktreeList.exitCode === 0) { + const current = listRefsGenerationByCommonDir.get(gitCommonDir); - let currentPath: string | null = null; - for (const line of worktreeList.stdout.split("\n")) { - if (line.startsWith("worktree ")) { - const candidatePath = line.slice("worktree ".length); - const exists = yield* fileSystem.stat(candidatePath).pipe( - Effect.map(() => true), - Effect.orElseSucceed(() => false), + return current === undefined + ? setListRefsGeneration(gitCommonDir, ++listRefsGenerationSequence) + : setListRefsGeneration(gitCommonDir, current); + }; + const bumpListRefsGeneration = (gitCommonDir: string): number => + setListRefsGeneration(gitCommonDir, ++listRefsGenerationSequence); + const listRefsSnapshotCache = yield* Cache.makeWith( + (cacheKey: GitRefsSnapshotCacheKey) => readGitRefsSnapshot(cacheKey.gitCommonDir), + { + capacity: LIST_REFS_SNAPSHOT_CACHE_CAPACITY, + timeToLive: (exit) => (Exit.isSuccess(exit) ? LIST_REFS_SNAPSHOT_CACHE_TTL : Duration.zero), + }, + ); + const listRefsRefreshSnapshotCache = yield* Cache.makeWith( + (cacheKey: GitRefsRefreshCacheKey) => + Effect.suspend(() => { + const epoch = bumpListRefsEpoch(cacheKey.gitCommonDir); + return Cache.get( + listRefsSnapshotCache, + new GitRefsSnapshotCacheKey({ gitCommonDir: cacheKey.gitCommonDir, epoch }), + ); + }), + { + capacity: LIST_REFS_SNAPSHOT_CACHE_CAPACITY, + timeToLive: (exit) => + Exit.isSuccess(exit) ? LIST_REFS_REFRESH_COALESCE_TTL : LIST_REFS_REFRESH_FAILURE_COOLDOWN, + }, + ); + const resolveListRefsSnapshot = Effect.fn("resolveListRefsSnapshot")(function* ( + gitCommonDir: string, + refresh: boolean, + ) { + while (true) { + const generation = currentListRefsGeneration(gitCommonDir); + const currentEpoch = listRefsEpochByCommonDir.get(gitCommonDir); + const snapshot = + refresh || currentEpoch === undefined + ? // The refresh cache owns the complete snapshot read, rather than only the + // epoch bump. Slow repositories therefore remain singleflight for the + // entire Git scan even when more refresh requests arrive after the + // coalescing TTL would otherwise have elapsed. + yield* Cache.get( + listRefsRefreshSnapshotCache, + new GitRefsRefreshCacheKey({ gitCommonDir, generation }), + ) + : yield* Cache.get( + listRefsSnapshotCache, + new GitRefsSnapshotCacheKey({ gitCommonDir, epoch: currentEpoch }), ); - currentPath = exists ? candidatePath : null; - } else if (line.startsWith("branch refs/heads/") && currentPath) { + if (currentListRefsGeneration(gitCommonDir) === generation) { + return snapshot; - worktreeMap.set(line.slice("branch refs/heads/".length), currentPath); - } else if (line === "") { - currentPath = null; - } - } } + } + }); + const invalidateListRefsSnapshot = Effect.fn("invalidateListRefsSnapshot")(function* ( + cwd: string, + ) { + const repositoryPathsCacheKey = normalizeRepositoryPathsCacheKey(cwd); + const repositoryPaths = yield* Cache.get(repositoryPathsCache, repositoryPathsCacheKey); + if (repositoryPaths === null) return; + const previousGeneration = currentListRefsGeneration(repositoryPaths.gitCommonDir); + bumpListRefsGeneration(repositoryPaths.gitCommonDir); + bumpListRefsEpoch(repositoryPaths.gitCommonDir); + yield* Cache.invalidate( + listRefsRefreshSnapshotCache, + new GitRefsRefreshCacheKey({ + gitCommonDir: repositoryPaths.gitCommonDir, + generation: previousGeneration, + }), + ); + yield* Cache.invalidate(repositoryPathsRefreshCache, repositoryPathsCacheKey); + yield* Cache.invalidate(repositoryPathsCache, repositoryPathsCacheKey); + }); + const listRefs: GitVcsDriver.GitVcsDriver["Service"]["listRefs"] = Effect.fn("listRefs")( - const localBranches = Arr.filterMap(localBranchResult.stdout.split("\n"), (line) => { - const refName = parseBranchLine(line); - return refName === null - ? Result.failVoid - : Result.succeed({ - name: refName.name, - current: refName.current, - isRemote: false, - isDefault: refName.name === defaultBranch, - worktreePath: worktreeMap.get(refName.name) ?? null, - }); - }).toSorted((a, b) => { - const aPriority = a.current ? 0 : a.isDefault ? 1 : 2; + function* (input) { + const repositoryPaths = yield* resolveRepositoryPaths(input.cwd, input.refresh === true).pipe( + Effect.catchTags({ + GitCommandError: (error) => - const bPriority = b.current ? 0 : b.isDefault ? 1 : 2; + isMissingGitCwdError(error) ? Effect.succeed(null) : Effect.fail(error), - if (aPriority !== bPriority) return aPriority - bPriority; - - const aLastCommit = branchLastCommit.get(a.name) ?? 0; - const bLastCommit = branchLastCommit.get(b.name) ?? 0; - if (aLastCommit !== bLastCommit) return bLastCommit - aLastCommit; - return a.name.localeCompare(b.name); - }); + }), + ); + if (repositoryPaths === null) { + return { + refs: [], + isRepo: false, + hasPrimaryRemote: false, + nextCursor: null, + totalCount: 0, + }; + } - const remoteBranches = - remoteBranchResult.exitCode === 0 + const snapshot = yield* resolveListRefsSnapshot( + repositoryPaths.gitCommonDir, + input.refresh === true, + ); + const hasCurrentWorktreeBranch = + repositoryPaths.worktreeRoot !== null && + snapshot.localBranches.some((ref) => ref.worktreePath === repositoryPaths.worktreeRoot); + const localBranches = snapshot.localBranches.map((ref) => ({ + ...ref, + current: hasCurrentWorktreeBranch + ? ref.worktreePath === repositoryPaths.worktreeRoot + : ref.name === repositoryPaths.currentBranch, + })); - ? Arr.filterMap(remoteBranchResult.stdout.split("\n"), (line) => { - const refName = parseBranchLine(line); - if (refName === null) { - return Result.failVoid; - } - const parsedRemoteRef = parseRemoteRefWithRemoteNames(refName.name, remoteNames); - const remoteBranch: { - name: string; - current: boolean; - isRemote: boolean; - remoteName?: string; - isDefault: boolean; - worktreePath: string | null; - } = { - name: refName.name, - current: false, - isRemote: true, - // origin/HEAD's target is the repo default even when no local - // copy of the default branch exists. - isDefault: - defaultBranch !== null && - parsedRemoteRef?.remoteName === "origin" && - parsedRemoteRef.branchName === defaultBranch, - worktreePath: null, - }; - if (parsedRemoteRef) { - remoteBranch.remoteName = parsedRemoteRef.remoteName; - } - return Result.succeed(remoteBranch); - }).toSorted((a, b) => { - const aLastCommit = branchLastCommit.get(a.name) ?? 0; - const bLastCommit = branchLastCommit.get(b.name) ?? 0; - if (aLastCommit !== bLastCommit) return bLastCommit - aLastCommit; - return a.name.localeCompare(b.name); - }) - : []; - const combinedBranches = input.includeMatchingRemoteRefs - ? [...localBranches, ...remoteBranches] - : dedupeRemoteBranchesWithLocalMatches([...localBranches, ...remoteBranches]); + ? [...localBranches, ...snapshot.remoteBranches] + : dedupeRemoteBranchesWithLocalMatches([...localBranches, ...snapshot.remoteBranches]); // Keep current/default refs on the first page even when the default // only exists as origin/ (remote refs sort after all locals). + const allBranches = combinedBranches.toSorted((left, right) => { + const leftPriority = left.current ? 0 : left.isDefault ? 1 : 2; + const rightPriority = right.current ? 0 : right.isDefault ? 1 : 2; + return leftPriority - rightPriority; - const allBranches = combinedBranches.toSorted((a, b) => { - const aPriority = a.current ? 0 : a.isDefault ? 1 : 2; - const bPriority = b.current ? 0 : b.isDefault ? 1 : 2; - return aPriority - bPriority; }); const branchesForKind = input.refKind === "local" @@ -2254,7 +2446,7 @@ return { refs: [...refs.refs], isRepo: true, - hasPrimaryRemote: remoteNames.includes("origin"), + hasPrimaryRemote: snapshot.hasPrimaryRemote, nextCursor: refs.nextCursor, totalCount: refs.totalCount, }; @@ -2548,6 +2740,25 @@ }), ); + const withListRefsInvalidation = ( + cwd: string, + effect: Effect.Effect, + ): Effect.Effect => + effect.pipe(Effect.ensuring(invalidateListRefsSnapshot(cwd).pipe(Effect.ignore))); + const initRepoWithListRefsInvalidation: GitVcsDriver.GitVcsDriver["Service"]["initRepo"] = ( + input, + ) => + initRepo(input).pipe( + Effect.ensuring( + Effect.gen(function* () { + const cacheKey = normalizeRepositoryPathsCacheKey(input.cwd); + yield* Cache.invalidate(repositoryPathsRefreshCache, cacheKey); + yield* Cache.invalidate(repositoryPathsCache, cacheKey); + yield* invalidateListRefsSnapshot(input.cwd).pipe(Effect.ignore); + }), + ), + ); + return GitVcsDriver.GitVcsDriver.of({ execute, status, @@ -2555,27 +2766,31 @@ statusDetailsLocal, statusDetailsRemote, prepareCommitContext, + commit: (cwd, subject, body, options) => + withListRefsInvalidation(cwd, commit(cwd, subject, body, options)), - commit, - pushCurrentBranch, - pullCurrentBranch, + pushCurrentBranch: (cwd, fallbackBranch, options) => + withListRefsInvalidation(cwd, pushCurrentBranch(cwd, fallbackBranch, options)), + pullCurrentBranch: (cwd) => withListRefsInvalidation(cwd, pullCurrentBranch(cwd)), readRangeContext, getReviewDiffPreview, readConfigValue, listRefs, + createWorktree: (input) => withListRefsInvalidation(input.cwd, createWorktree(input)), + fetchPullRequestBranch: (input) => + withListRefsInvalidation(input.cwd, fetchPullRequestBranch(input)), + ensureRemote: (input) => withListRefsInvalidation(input.cwd, ensureRemote(input)), - createWorktree, - fetchPullRequestBranch, - ensureRemote, resolvePrimaryRemoteName, - fetchRemote, + fetchRemote: (input) => withListRefsInvalidation(input.cwd, fetchRemote(input)), resolveRemoteTrackingCommit, - fetchRemoteBranch, - fetchRemoteTrackingBranch, - setBranchUpstream, - removeWorktree, + fetchRemoteBranch: (input) => withListRefsInvalidation(input.cwd, fetchRemoteBranch(input)), + fetchRemoteTrackingBranch: (input) => + withListRefsInvalidation(input.cwd, fetchRemoteTrackingBranch(input)), + setBranchUpstream: (input) => withListRefsInvalidation(input.cwd, setBranchUpstream(input)), + removeWorktree: (input) => withListRefsInvalidation(input.cwd, removeWorktree(input)), + renameBranch: (input) => withListRefsInvalidation(input.cwd, renameBranch(input)), + createRef: (input) => withListRefsInvalidation(input.cwd, createRef(input)), - renameBranch, - createRef, - switchRef, + switchRef: (input) => withListRefsInvalidation(input.cwd, switchRef(input)), + initRepo: initRepoWithListRefsInvalidation, - initRepo, listLocalBranchNames, }); }); ``` ## apps/web/src/components/BranchToolbarBranchSelector.tsx Status: modified; +26; −30 ```diff @@ -24,6 +24,7 @@ import { writeTextToClipboard } from "../hooks/useCopyToClipboard"; import { readLocalApi } from "../localApi"; import { useOpenPrLink } from "../lib/openPullRequestLink"; +import { shouldLoadNextBranchPageAfterScroll } from "../state/paginatedBranches"; import { usePaginatedBranches } from "../state/queries"; import { useProject, useThread } from "../state/entities"; import { useEnvironmentQuery } from "../state/query"; @@ -230,7 +231,7 @@ const refs = branchRefState.refs; const hasNextPage = branchRefState.data?.nextCursor !== null && branchRefState.data?.nextCursor !== undefined; - const isFetchingNextPage = branchRefState.isPending && branchRefState.data !== null; + const isFetchingNextPage = branchRefState.isFetchingNextPage; const isInitialBranchesLoadPending = branchRefState.isPending && branchRefState.data === null; const currentGitBranch = branchStatusQuery.data?.refName ?? refs.find((refName) => refName.current)?.name ?? null; @@ -505,19 +506,16 @@ // --------------------------------------------------------------------------- // Combobox / list plumbing // --------------------------------------------------------------------------- - const handleOpenChange = useCallback( - (open: boolean) => { - setIsBranchMenuOpen(open); - if (!open) { - setBranchQuery(""); - return; - } - branchRefState.refresh(); - }, - [branchRefState.refresh], - ); - const branchListScrollElementRef = useRef(null); + const previousBranchListScrollTopRef = useRef(null); + const handleOpenChange = useCallback((open: boolean) => { + previousBranchListScrollTopRef.current = null; + setIsBranchMenuOpen(open); + if (!open) { + setBranchQuery(""); + } + }, []); + const [showTopBranchScrollFade, setShowTopBranchScrollFade] = useState(false); const [showBottomBranchScrollFade, setShowBottomBranchScrollFade] = useState(false); const fetchNextBranchPage = useCallback(() => { @@ -528,18 +526,24 @@ branchRefState.loadNext(); }, [branchRefState.loadNext, hasNextPage, isFetchingNextPage]); const maybeFetchNextBranchPage = useCallback(() => { - if (!isBranchMenuOpen || !hasNextPage || isFetchingNextPage) { - return; - } - const scrollElement = branchListScrollElementRef.current; if (!scrollElement) { return; } + const previousScrollTop = previousBranchListScrollTopRef.current; + previousBranchListScrollTopRef.current = scrollElement.scrollTop; + if ( + !isBranchMenuOpen || - const distanceFromBottom = - scrollElement.scrollHeight - scrollElement.scrollTop - scrollElement.clientHeight; - if (distanceFromBottom > 96) { + !hasNextPage || + isFetchingNextPage || + !shouldLoadNextBranchPageAfterScroll({ + previousScrollTop, + scrollTop: scrollElement.scrollTop, + scrollHeight: scrollElement.scrollHeight, + clientHeight: scrollElement.clientHeight, + }) + ) { return; } @@ -589,10 +593,6 @@ void branchListRef.current?.scrollToOffset?.({ offset: 0, animated: false }); }, [deferredTrimmedBranchQuery, isBranchMenuOpen]); - useEffect(() => { - maybeFetchNextBranchPage(); - }, [refs.length, maybeFetchNextBranchPage]); - const triggerLabel = resolveBranchTriggerLabel({ activeWorktreePath, effectiveEnvMode, @@ -791,14 +791,10 @@ renderItem={({ item, index }) => renderPickerItem(item, index)} estimatedItemSize={28} drawDistance={336} - onEndReached={() => { - if (hasNextPage && !isFetchingNextPage) { - fetchNextBranchPage(); - } - }} onLayout={() => { updateBranchListScrollFades(); + previousBranchListScrollTopRef.current = + branchListScrollElementRef.current?.scrollTop ?? null; - maybeFetchNextBranchPage(); }} onScroll={() => { updateBranchListScrollFades(); ``` ## apps/web/src/connection/storage.ts Status: modified; +14; −0 ```diff @@ -105,6 +105,8 @@ | "save-server-config" | "load-vcs-refs" | "save-vcs-refs" + | "remove-vcs-refs" + | "clear-vcs-refs" | "clear-environment", cause: unknown, ) { @@ -619,6 +621,18 @@ : persistenceError("save-vcs-refs", cause), ), ), + removeVcsRefs: (environmentId, cwd) => + removeDatabaseValue( + database, + VCS_REFS_STORE_NAME, + vcsRefsCacheKey(environmentId, cwd), + ).pipe(Effect.mapError((cause) => persistenceError("remove-vcs-refs", cause))), + clearVcsRefs: (environmentId) => + removeDatabaseValuesInRange( + database, + VCS_REFS_STORE_NAME, + IDBKeyRange.bound(`${environmentId}:`, `${environmentId}:\uffff`), + ).pipe(Effect.mapError((cause) => persistenceError("clear-vcs-refs", cause))), removeThread: (environmentId, threadId) => removeDatabaseValue( database, ``` ## apps/web/src/state/paginatedBranches.test.ts Status: added; +111; −0 ```diff @@ -1,0 +1,111 @@ +import type { VcsListRefsResult } from "@t3tools/contracts"; +import { AsyncResult } from "effect/unstable/reactivity"; +import { describe, expect, it } from "vite-plus/test"; + +import { + isPaginatedBranchesNextPagePending, + shouldLoadNextBranchPageAfterScroll, +} from "./paginatedBranches"; + +const FIRST_PAGE: VcsListRefsResult = { + refs: [], + isRepo: true, + hasPrimaryRemote: true, + nextCursor: 100, + totalCount: 150, +}; + +const LAST_PAGE: VcsListRefsResult = { + ...FIRST_PAGE, + nextCursor: null, +}; + +describe("paginated branch loading state", () => { + it("does not label a first-page background refresh as loading more refs", () => { + expect( + isPaginatedBranchesNextPagePending([ + AsyncResult.success(FIRST_PAGE, { + waiting: true, + }), + ]), + ).toBe(false); + }); + + it("only reports loading more while a new cursor has no value", () => { + expect( + isPaginatedBranchesNextPagePending([ + AsyncResult.success(FIRST_PAGE), + AsyncResult.initial(true), + ]), + ).toBe(true); + + expect( + isPaginatedBranchesNextPagePending([ + AsyncResult.success(FIRST_PAGE), + AsyncResult.success(LAST_PAGE), + ]), + ).toBe(false); + }); +}); + +describe("paginated branch scroll trigger", () => { + it("does not load from mount, layout, or an unchanged scroll position", () => { + expect( + shouldLoadNextBranchPageAfterScroll({ + previousScrollTop: null, + scrollTop: 0, + scrollHeight: 224, + clientHeight: 224, + }), + ).toBe(false); + + expect( + shouldLoadNextBranchPageAfterScroll({ + previousScrollTop: 120, + scrollTop: 120, + scrollHeight: 344, + clientHeight: 224, + }), + ).toBe(false); + }); + + it("does not load when scrolling upward near the bottom", () => { + expect( + shouldLoadNextBranchPageAfterScroll({ + previousScrollTop: 160, + scrollTop: 150, + scrollHeight: 450, + clientHeight: 224, + }), + ).toBe(false); + }); + + it("loads only after a downward scroll reaches the end threshold", () => { + expect( + shouldLoadNextBranchPageAfterScroll({ + previousScrollTop: 0, + scrollTop: 100, + scrollHeight: 800, + clientHeight: 224, + }), + ).toBe(false); + + expect( + shouldLoadNextBranchPageAfterScroll({ + previousScrollTop: 100, + scrollTop: 500, + scrollHeight: 800, + clientHeight: 224, + }), + ).toBe(true); + + expect( + shouldLoadNextBranchPageAfterScroll({ + previousScrollTop: 499.5, + scrollTop: 500, + scrollHeight: 800, + clientHeight: 224, + }), + ).toBe(true); + }); +}); ``` ## apps/web/src/state/paginatedBranches.ts Status: added; +36; −0 ```diff @@ -1,0 +1,36 @@ +import type { VcsListRefsResult } from "@t3tools/contracts"; +import * as Option from "effect/Option"; +import { AsyncResult } from "effect/unstable/reactivity"; + +const DEFAULT_NEXT_PAGE_DISTANCE_PX = 96; + +export function isPaginatedBranchesNextPagePending( + results: ReadonlyArray>, +): boolean { + const lastResult = results.at(-1); + return ( + results.length > 1 && + lastResult?.waiting === true && + Option.isNone(AsyncResult.value(lastResult)) + ); +} + +export function shouldLoadNextBranchPageAfterScroll({ + previousScrollTop, + scrollTop, + scrollHeight, + clientHeight, + distanceThreshold = DEFAULT_NEXT_PAGE_DISTANCE_PX, +}: { + previousScrollTop: number | null; + scrollTop: number; + scrollHeight: number; + clientHeight: number; + distanceThreshold?: number; +}): boolean { + if (previousScrollTop === null || scrollTop <= previousScrollTop) { + return false; + } + + return scrollHeight - scrollTop - clientHeight <= distanceThreshold; +} ``` ## apps/web/src/state/queries.ts Status: modified; +3; −0 ```diff @@ -18,6 +18,7 @@ import { appAtomRegistry } from "../rpc/atomRegistry"; import { orchestrationEnvironment } from "./orchestration"; +import { isPaginatedBranchesNextPagePending } from "./paginatedBranches"; import { projectEnvironment } from "./projects"; import { useEnvironmentQuery } from "./query"; import { useEnvironmentThread } from "./threads"; @@ -142,6 +143,7 @@ totalCount: Math.max(...values.map((value) => value.totalCount)), }; const failed = results.find((result) => result._tag === "Failure"); + const isFetchingNextPage = isPaginatedBranchesNextPagePending(results); const error = failed?._tag === "Failure" ? (() => { @@ -176,6 +178,7 @@ refs: data?.refs ?? EMPTY_REFS, error, isPending: results.some((result) => result.waiting), + isFetchingNextPage, refresh, loadNext, }; ``` ## packages/client-runtime/src/connection/registry.test.ts Status: modified; +2; −0 ```diff @@ -251,6 +251,8 @@ saveServerConfig: () => Effect.void, loadVcsRefs: () => Effect.succeed(Option.none()), saveVcsRefs: () => Effect.void, + removeVcsRefs: () => Effect.void, + clearVcsRefs: () => Effect.void, clear: (environmentId) => Ref.update(shellCache, (current) => { const next = new Map(current); ``` ## packages/client-runtime/src/platform/persistence.ts Status: modified; +14; −0 ```diff @@ -30,6 +30,8 @@ "save-server-config", "load-vcs-refs", "save-vcs-refs", + "remove-vcs-refs", + "clear-vcs-refs", "clear-environment", ]), message: Schema.String, @@ -102,6 +104,18 @@ cwd: string, refs: VcsListRefsResult, ) => Effect.Effect; + readonly removeVcsRefs: ( + environmentId: EnvironmentId, + cwd: string, + ) => Effect.Effect; + /** + * Removes every persisted branch-list snapshot for an environment. Git ref + * mutations are repository-wide, and linked worktrees may have cached the + * same refs under different working-directory keys. + */ + readonly clearVcsRefs: ( + environmentId: EnvironmentId, + ) => Effect.Effect; readonly clear: ( environmentId: EnvironmentId, ) => Effect.Effect; ``` ## packages/client-runtime/src/state/runtime.ts Status: modified; +41; −3 ```diff @@ -30,6 +30,17 @@ readonly environmentId: EnvironmentIdType; readonly input: Input; }>; +} + +interface EnvironmentCommandAtomOptions extends Omit< + EnvironmentAtomOptions, + "execute" +> { + readonly execute: ( + input: Input, + registry: AtomRegistry.AtomRegistry, + environmentId: EnvironmentIdType, + ) => Effect.Effect; } interface EnvironmentQueryAtomOptions extends EnvironmentAtomOptions< @@ -520,13 +531,17 @@ export function createEnvironmentCommand( runtime: Atom.AtomRuntime, - options: EnvironmentAtomOptions, + options: EnvironmentCommandAtomOptions, ) { return createRuntimeCommand(runtime, { label: options.label, ...(options.scheduler === undefined ? {} : { scheduler: options.scheduler }), ...(options.concurrency === undefined ? {} : { concurrency: options.concurrency }), + execute: (target, registry) => + runInEnvironment( + target.environmentId, + options.execute(target.input, registry, target.environmentId), - execute: (target) => runInEnvironment(target.environmentId, options.execute(target.input)), + ), }); } @@ -616,13 +631,36 @@ readonly environmentId: EnvironmentIdType; readonly input: EnvironmentRpcInput; }>; + readonly onSuccess?: ( + target: { + readonly environmentId: EnvironmentIdType; + readonly input: EnvironmentRpcInput; + }, + registry: AtomRegistry.AtomRegistry, + ) => Effect.Effect; + readonly onSettled?: ( + target: { + readonly environmentId: EnvironmentIdType; + readonly input: EnvironmentRpcInput; + }, + registry: AtomRegistry.AtomRegistry, + ) => Effect.Effect; }, ) { return createEnvironmentCommand(runtime, { label: options.label, ...(options.scheduler === undefined ? {} : { scheduler: options.scheduler }), ...(options.concurrency === undefined ? {} : { concurrency: options.concurrency }), + execute: (input: EnvironmentRpcInput, registry, environmentId) => { + const target = { + environmentId, + input, + }; + return request(options.tag, input).pipe( + Effect.tap(() => options.onSuccess?.(target, registry) ?? Effect.void), + Effect.ensuring(options.onSettled?.(target, registry) ?? Effect.void), + ); - execute: (input: EnvironmentRpcInput) => request(options.tag, input), + }, }); } ``` ## packages/client-runtime/src/state/server.test.ts Status: modified; +4; −0 ```diff @@ -152,6 +152,8 @@ saveServerConfig: (_environmentId, config) => Queue.offer(savedConfigs, config), loadVcsRefs: () => Effect.succeed(Option.none()), saveVcsRefs: () => Effect.void, + removeVcsRefs: () => Effect.void, + clearVcsRefs: () => Effect.void, clear: () => Effect.void, }); @@ -211,6 +213,8 @@ saveServerConfig: (_environmentId, config) => Queue.offer(savedConfigs, config), loadVcsRefs: () => Effect.succeed(Option.none()), saveVcsRefs: () => Effect.void, + removeVcsRefs: () => Effect.void, + clearVcsRefs: () => Effect.void, clear: () => Effect.void, }); ``` ## packages/client-runtime/src/state/shell-sync.test.ts Status: modified; +6; −0 ```diff @@ -87,6 +87,8 @@ saveServerConfig: () => Effect.void, loadVcsRefs: () => Effect.succeed(Option.none()), saveVcsRefs: () => Effect.void, + removeVcsRefs: () => Effect.void, + clearVcsRefs: () => Effect.void, clear: () => Effect.void, }); // Cold cache with no HTTP snapshot available → falls back to the @@ -201,6 +203,8 @@ saveServerConfig: () => Effect.void, loadVcsRefs: () => Effect.succeed(Option.none()), saveVcsRefs: () => Effect.void, + removeVcsRefs: () => Effect.void, + clearVcsRefs: () => Effect.void, clear: () => Effect.void, }); const snapshotLoader = ShellSnapshotLoader.of({ @@ -270,6 +274,8 @@ saveServerConfig: () => Effect.void, loadVcsRefs: () => Effect.succeed(Option.none()), saveVcsRefs: () => Effect.void, + removeVcsRefs: () => Effect.void, + clearVcsRefs: () => Effect.void, clear: () => Effect.void, }); const snapshotLoader = ShellSnapshotLoader.of({ ``` ## packages/client-runtime/src/state/sourceControl.test.ts Status: added; +169; −0 ```diff @@ -1,0 +1,169 @@ +import { + EnvironmentId, + WS_METHODS, + type SourceControlPublishRepositoryResult, +} from "@t3tools/contracts"; +import { describe, expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as SubscriptionRef from "effect/SubscriptionRef"; +import { AsyncResult, Atom, AtomRegistry } from "effect/unstable/reactivity"; + +import { + AVAILABLE_CONNECTION_STATE, + PrimaryConnectionTarget, + type PreparedConnection, + type SupervisorConnectionState, +} from "../connection/model.ts"; +import * as EnvironmentRegistry from "../connection/registry.ts"; +import * as EnvironmentSupervisor from "../connection/supervisor.ts"; +import * as Persistence from "../platform/persistence.ts"; +import { EnvironmentRpcUnavailableError } from "../rpc/client.ts"; +import type { WsRpcProtocolClient } from "../rpc/protocol.ts"; +import type { RpcSession } from "../rpc/session.ts"; +import { createSourceControlEnvironmentAtoms } from "./sourceControl.ts"; +import { vcsRefsCacheStateAtom } from "./vcsRefInvalidation.ts"; + +const TARGET = new PrimaryConnectionTarget({ + environmentId: EnvironmentId.make("environment-1"), + label: "Test environment", + httpBaseUrl: "https://environment.example.test", + wsBaseUrl: "wss://environment.example.test", +}); + +const PUBLISH_RESULT: SourceControlPublishRepositoryResult = { + repository: { + provider: "github", + nameWithOwner: "t3tools/t3code", + url: "https://github.com/t3tools/t3code", + sshUrl: "git@github.com:t3tools/t3code.git", + }, + remoteName: "origin", + remoteUrl: "git@github.com:t3tools/t3code.git", + branch: "main", + upstreamBranch: "origin/main", + status: "pushed", +}; + +function session(client: WsRpcProtocolClient): RpcSession { + return { + client, + initialConfig: Effect.never, + ready: Effect.void, + probe: Effect.void, + closed: Effect.never, + }; +} + +describe("source control environment atoms", () => { + it.effect("invalidates cached refs after successful and failed publishing", () => + Effect.scoped( + Effect.gen(function* () { + const connectionState: SupervisorConnectionState = { + ...AVAILABLE_CONNECTION_STATE, + desired: true, + network: "online", + phase: "connected", + attempt: 1, + generation: 1, + }; + let publishAttempts = 0; + const client = { + [WS_METHODS.sourceControlPublishRepository]: () => { + publishAttempts += 1; + return publishAttempts === 1 + ? Effect.succeed(PUBLISH_RESULT) + : Effect.fail( + new EnvironmentRpcUnavailableError({ + environmentId: TARGET.environmentId, + message: "push failed after adding the remote", + }), + ); + }, + } as unknown as WsRpcProtocolClient; + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: yield* SubscriptionRef.make(connectionState), + session: yield* SubscriptionRef.make(Option.some(session(client))), + prepared: yield* SubscriptionRef.make(Option.none()), + connect: Effect.void, + disconnect: Effect.void, + retryNow: Effect.void, + } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]); + const run: EnvironmentRegistry.EnvironmentRegistry["Service"]["run"] = ( + _environmentId, + effect, + ) => Effect.provideService(effect, EnvironmentSupervisor.EnvironmentSupervisor, supervisor); + const environmentRegistry = EnvironmentRegistry.EnvironmentRegistry.of({ + run, + } as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]); + const removed = new Array(); + const cache = Persistence.EnvironmentCacheStore.of({ + loadShell: () => Effect.succeed(Option.none()), + saveShell: () => Effect.void, + loadThread: () => Effect.succeed(Option.none()), + saveThread: () => Effect.void, + removeThread: () => Effect.void, + loadServerConfig: () => Effect.succeed(Option.none()), + saveServerConfig: () => Effect.void, + loadVcsRefs: () => Effect.succeed(Option.none()), + saveVcsRefs: () => Effect.void, + removeVcsRefs: (environmentId, cwd) => + Effect.sync(() => { + removed.push(`${environmentId}:${cwd}`); + }), + clearVcsRefs: (environmentId) => + Effect.sync(() => { + removed.push(`${environmentId}:*`); + }), + clear: () => Effect.void, + }); + const runtime = Atom.runtime( + Layer.merge( + Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry), + Layer.succeed(Persistence.EnvironmentCacheStore, cache), + ), + ); + const atoms = createSourceControlEnvironmentAtoms(runtime); + const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) => + Effect.sync(() => registry.dispose()), + ); + const state = vcsRefsCacheStateAtom({ environmentId: TARGET.environmentId }); + + expect(registry.get(state).revision).toBe(0); + const publishResult = yield* Effect.promise(() => + atoms.publishRepository.run(registry, { + environmentId: TARGET.environmentId, + input: { + cwd: "/repo", + provider: "github", + repository: "t3tools/t3code", + visibility: "private", + }, + }), + ); + + expect(AsyncResult.isSuccess(publishResult)).toBe(true); + expect(registry.get(state).revision).toBe(1); + expect(removed).toEqual([`${TARGET.environmentId}:*`]); + + const failedPublish = yield* Effect.promise(() => + atoms.publishRepository.run(registry, { + environmentId: TARGET.environmentId, + input: { + cwd: "/repo", + provider: "github", + repository: "t3tools/t3code", + visibility: "private", + }, + }), + ); + + expect(AsyncResult.isFailure(failedPublish)).toBe(true); + expect(registry.get(state).revision).toBe(2); + expect(removed).toEqual([`${TARGET.environmentId}:*`, `${TARGET.environmentId}:*`]); + }), + ), + ); +}); ``` ## packages/client-runtime/src/state/sourceControl.ts Status: modified; +8; −1 ```diff @@ -7,10 +7,12 @@ createEnvironmentRpcQueryAtomFamily, } from "./runtime.ts"; import type { EnvironmentRegistry } from "../connection/registry.ts"; +import { EnvironmentCacheStore } from "../platform/persistence.ts"; import { vcsCommandConcurrency, vcsCommandScheduler } from "./vcsCommandScheduler.ts"; +import { invalidateCachedVcsRefs } from "./vcsRefInvalidation.ts"; export function createSourceControlEnvironmentAtoms( + runtime: Atom.AtomRuntime, - runtime: Atom.AtomRuntime, ) { const commandScheduler = createAtomCommandScheduler(); return { @@ -36,6 +38,11 @@ tag: WS_METHODS.sourceControlPublishRepository, scheduler: vcsCommandScheduler, concurrency: vcsCommandConcurrency, + onSettled: (target, registry) => + invalidateCachedVcsRefs(registry, { + environmentId: target.environmentId, + cwd: target.input.cwd, + }), }), }; } ``` ## packages/client-runtime/src/state/threads-sync.test.ts Status: modified; +2; −0 ```diff @@ -217,6 +217,8 @@ saveServerConfig: () => Effect.void, loadVcsRefs: () => Effect.succeed(Option.none()), saveVcsRefs: () => Effect.void, + removeVcsRefs: () => Effect.void, + clearVcsRefs: () => Effect.void, clear: () => Effect.void, }); const threadState = yield* makeEnvironmentThreadState(THREAD_ID).pipe( ``` ## packages/client-runtime/src/state/vcs.test.ts Status: modified; +432; −20 ```diff @@ -1,12 +1,19 @@ -import { EnvironmentId, WS_METHODS, type VcsListRefsResult } from "@t3tools/contracts"; +import { + EnvironmentId, + WS_METHODS, + type VcsListRefsInput, + type VcsListRefsResult, +} from "@t3tools/contracts"; import { describe, expect, it } from "@effect/vitest"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; import * as TestClock from "effect/testing/TestClock"; +import { AsyncResult, Atom, AtomRegistry } from "effect/unstable/reactivity"; import { AVAILABLE_CONNECTION_STATE, @@ -14,11 +21,23 @@ type PreparedConnection, type SupervisorConnectionState, } from "../connection/model.ts"; +import * as EnvironmentRegistry from "../connection/registry.ts"; import * as EnvironmentSupervisor from "../connection/supervisor.ts"; import * as Persistence from "../platform/persistence.ts"; +import { EnvironmentRpcUnavailableError } from "../rpc/client.ts"; import type { WsRpcProtocolClient } from "../rpc/protocol.ts"; import type { RpcSession } from "../rpc/session.ts"; + +import { + commitVcsRefsRefresh, + createVcsEnvironmentAtoms, + makeCachedVcsRefsChanges, +} from "./vcs.ts"; +import { + invalidateCachedVcsRefs, + invalidateVcsRefs, + vcsRefsCacheStateAtom, -import { makeCachedVcsRefsChanges } from "./vcs.ts"; +} from "./vcsRefInvalidation.ts"; const TARGET = new PrimaryConnectionTarget({ environmentId: EnvironmentId.make("environment-1"), @@ -73,7 +92,10 @@ }; } +function cacheWithRefs( + refs: Option.Option, + overrides: Partial = {}, -function cacheWithRefs(refs: Option.Option) { +) { return Persistence.EnvironmentCacheStore.of({ loadShell: () => Effect.succeed(Option.none()), saveShell: () => Effect.void, @@ -84,11 +106,345 @@ saveServerConfig: () => Effect.void, loadVcsRefs: () => Effect.succeed(refs), saveVcsRefs: () => Effect.void, + removeVcsRefs: () => Effect.void, + clearVcsRefs: () => Effect.void, clear: () => Effect.void, + ...overrides, }); } describe("cached VCS refs", () => { + it("invalidates all ref streams in the mutated environment", () => { + const registry = AtomRegistry.make(); + const environment = { + environmentId: TARGET.environmentId, + }; + const otherEnvironment = { + environmentId: EnvironmentId.make("environment-2"), + }; + + expect(registry.get(vcsRefsCacheStateAtom(environment))).toEqual({ + revision: 0, + persistedCacheReadable: true, + }); + expect(registry.get(vcsRefsCacheStateAtom(otherEnvironment))).toEqual({ + revision: 0, + persistedCacheReadable: true, + }); + + invalidateVcsRefs(registry, environment); + + expect(registry.get(vcsRefsCacheStateAtom(environment))).toEqual({ + revision: 1, + persistedCacheReadable: true, + }); + expect(registry.get(vcsRefsCacheStateAtom(otherEnvironment))).toEqual({ + revision: 0, + persistedCacheReadable: true, + }); + registry.dispose(); + }); + + it.effect("preserves the caller's repository snapshot refresh policy", () => + Effect.scoped( + Effect.gen(function* () { + const requests = yield* Ref.make>([]); + const client = { + [WS_METHODS.vcsListRefs]: (input: VcsListRefsInput) => + Ref.update(requests, (current) => [...current, input]).pipe(Effect.as(LIVE_REFS)), + } as unknown as WsRpcProtocolClient; + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE), + session: yield* SubscriptionRef.make(Option.some(session(client))), + prepared: yield* SubscriptionRef.make(Option.none()), + connect: Effect.void, + disconnect: Effect.void, + retryNow: Effect.void, + } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]); + + yield* Stream.unwrap( + makeCachedVcsRefsChanges({ + cwd: "/repo", + limit: 20, + query: "release", + refKind: "remote", + }).pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cacheWithRefs(Option.none())), + ), + ).pipe(Stream.runHead); + + expect(yield* Ref.get(requests)).toEqual([ + { + cwd: "/repo", + limit: 20, + query: "release", + refKind: "remote", + }, + ]); + + yield* Stream.unwrap( + makeCachedVcsRefsChanges({ + cwd: "/repo", + limit: 20, + query: "release", + refKind: "remote", + refresh: true, + }).pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cacheWithRefs(Option.none())), + ), + ).pipe(Stream.runHead); + + expect(yield* Ref.get(requests)).toEqual([ + { + cwd: "/repo", + limit: 20, + query: "release", + refKind: "remote", + }, + { + cwd: "/repo", + limit: 20, + query: "release", + refKind: "remote", + refresh: true, + }, + ]); + }), + ), + ); + + it.effect("does not repersist a refresh superseded by ref invalidation", () => + Effect.scoped( + Effect.gen(function* () { + const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) => + Effect.sync(() => registry.dispose()), + ); + const saved = yield* Ref.make>([]); + const clears = yield* Ref.make(0); + const revisionsObservedDuringClear = yield* Ref.make>([]); + const cache = cacheWithRefs(Option.none(), { + saveVcsRefs: (_environmentId, _cwd, refs) => + Ref.update(saved, (current) => [...current, refs]), + clearVcsRefs: () => + Effect.all([ + Ref.update(clears, (count) => count + 1), + Ref.update(revisionsObservedDuringClear, (current) => [ + ...current, + registry.get(vcsRefsCacheStateAtom(TARGET)).revision, + ]), + ]).pipe(Effect.asVoid), + }); + + yield* invalidateCachedVcsRefs(registry, { + environmentId: TARGET.environmentId, + cwd: "/repo-worktree", + }).pipe(Effect.provideService(Persistence.EnvironmentCacheStore, cache)); + + expect( + yield* commitVcsRefsRefresh(registry, cache, { + environmentId: TARGET.environmentId, + cwd: "/repo", + refs: CACHED_REFS, + expectedRevision: 0, + persist: true, + }), + ).toBe(false); + + expect(yield* Ref.get(saved)).toEqual([]); + expect(yield* Ref.get(clears)).toBe(1); + expect(yield* Ref.get(revisionsObservedDuringClear)).toEqual([0]); + expect(registry.get(vcsRefsCacheStateAtom(TARGET))).toEqual({ + revision: 1, + persistedCacheReadable: true, + }); + + expect( + yield* commitVcsRefsRefresh(registry, cache, { + environmentId: TARGET.environmentId, + cwd: "/repo", + refs: LIVE_REFS, + expectedRevision: 1, + persist: true, + }), + ).toBe(true); + expect(yield* Ref.get(saved)).toEqual([LIVE_REFS]); + }), + ), + ); + + it.effect("invalidates persisted refs when ref-affecting commands settle", () => + Effect.scoped( + Effect.gen(function* () { + const expectedError = new EnvironmentRpcUnavailableError({ + environmentId: TARGET.environmentId, + message: "pull failed after fetching refs", + }); + const client = { + [WS_METHODS.vcsPull]: () => Effect.fail(expectedError), + [WS_METHODS.vcsRefreshStatus]: () => Effect.void, + } as unknown as WsRpcProtocolClient; + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE), + session: yield* SubscriptionRef.make(Option.some(session(client))), + prepared: yield* SubscriptionRef.make(Option.none()), + connect: Effect.void, + disconnect: Effect.void, + retryNow: Effect.void, + } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]); + const run: EnvironmentRegistry.EnvironmentRegistry["Service"]["run"] = ( + _environmentId, + effect, + ) => Effect.provideService(effect, EnvironmentSupervisor.EnvironmentSupervisor, supervisor); + const environmentRegistry = EnvironmentRegistry.EnvironmentRegistry.of({ + run, + } as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]); + const clears = yield* Ref.make(0); + const runtime = Atom.runtime( + Layer.merge( + Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry), + Layer.succeed( + Persistence.EnvironmentCacheStore, + cacheWithRefs(Option.none(), { + clearVcsRefs: () => Ref.update(clears, (count) => count + 1), + }), + ), + ), + ); + const atoms = createVcsEnvironmentAtoms(runtime); + const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) => + Effect.sync(() => registry.dispose()), + ); + + const result = yield* Effect.promise(() => + atoms.pull.run(registry, { + environmentId: TARGET.environmentId, + input: { cwd: "/repo" }, + }), + ); + + expect(AsyncResult.isFailure(result)).toBe(true); + expect(yield* Ref.get(clears)).toBe(1); + expect(registry.get(vcsRefsCacheStateAtom(TARGET)).revision).toBe(1); + + const refreshResult = yield* Effect.promise(() => + atoms.refreshStatus.run(registry, { + environmentId: TARGET.environmentId, + input: { cwd: "/repo" }, + }), + ); + + expect(AsyncResult.isSuccess(refreshResult)).toBe(true); + expect(yield* Ref.get(clears)).toBe(2); + expect(registry.get(vcsRefsCacheStateAtom(TARGET)).revision).toBe(2); + }), + ), + ); + + it.effect("suppresses persisted snapshots after an environment-wide clear fails", () => + Effect.scoped( + Effect.gen(function* () { + const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) => + Effect.sync(() => registry.dispose()), + ); + const persisted = yield* Ref.make>( + Option.some(CACHED_REFS), + ); + const loads = yield* Ref.make(0); + const clearAttempts = yield* Ref.make(0); + const cache = cacheWithRefs(Option.none(), { + loadVcsRefs: () => + Ref.update(loads, (count) => count + 1).pipe(Effect.andThen(Ref.get(persisted))), + saveVcsRefs: (_environmentId, _cwd, refs) => Ref.set(persisted, Option.some(refs)), + clearVcsRefs: () => + Ref.updateAndGet(clearAttempts, (count) => count + 1).pipe( + Effect.flatMap((attempt) => + attempt === 1 + ? Effect.fail( + new Persistence.ConnectionPersistenceError({ + operation: "clear-vcs-refs", + message: "storage unavailable", + }), + ) + : Ref.set(persisted, Option.none()), + ), + ), + }); + + yield* invalidateCachedVcsRefs(registry, { + environmentId: TARGET.environmentId, + cwd: "/repo", + }).pipe(Effect.provideService(Persistence.EnvironmentCacheStore, cache)); + + const state = registry.get(vcsRefsCacheStateAtom(TARGET)); + expect(state).toEqual({ + revision: 1, + persistedCacheReadable: false, + }); + + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE), + session: yield* SubscriptionRef.make(Option.none()), + prepared: yield* SubscriptionRef.make(Option.none()), + connect: Effect.void, + disconnect: Effect.void, + retryNow: Effect.void, + } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]); + const refs = yield* Stream.unwrap( + makeCachedVcsRefsChanges( + { cwd: "/repo", limit: 100 }, + state.revision, + registry, + state.persistedCacheReadable, + ).pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cache), + ), + ).pipe(Stream.runHead, Effect.forkChild({ startImmediately: true })); + + yield* Effect.yieldNow; + expect(yield* Ref.get(loads)).toBe(0); + yield* Fiber.interrupt(refs); + expect(registry.get(vcsRefsCacheStateAtom(TARGET))).toEqual(state); + + expect( + yield* commitVcsRefsRefresh(registry, cache, { + environmentId: TARGET.environmentId, + cwd: "/repo", + refs: LIVE_REFS, + expectedRevision: state.revision, + persist: true, + }), + ).toBe(true); + const recoveredState = registry.get(vcsRefsCacheStateAtom(TARGET)); + expect(recoveredState).toEqual({ + revision: 1, + persistedCacheReadable: true, + }); + expect(yield* Ref.get(clearAttempts)).toBe(2); + + const recoveredRefs = yield* Stream.unwrap( + makeCachedVcsRefsChanges( + { cwd: "/repo", limit: 100 }, + recoveredState.revision, + registry, + recoveredState.persistedCacheReadable, + ).pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cache), + ), + ).pipe(Stream.runHead); + + expect(Option.getOrThrow(recoveredRefs)).toEqual(LIVE_REFS); + expect(yield* Ref.get(loads)).toBe(1); + }), + ), + ); + it.effect("loads an unfiltered branch list without a connection", () => Effect.scoped( Effect.gen(function* () { @@ -116,11 +472,12 @@ ), ); - it.effect("continues polling after a transient live failure", () => + it.effect("retries a transient live failure while connected", () => Effect.scoped( Effect.gen(function* () { const expectedError = new Error("Could not list Git refs."); const calls = yield* Ref.make(0); + const connectionState = yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE); const client = { [WS_METHODS.vcsListRefs]: () => Ref.updateAndGet(calls, (count) => count + 1).pipe( @@ -131,7 +488,7 @@ } as unknown as WsRpcProtocolClient; const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ target: TARGET, - state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE), + state: connectionState, session: yield* SubscriptionRef.make(Option.some(session(client))), prepared: yield* SubscriptionRef.make(Option.none()), connect: Effect.void, @@ -142,10 +499,7 @@ const result = Stream.unwrap( makeCachedVcsRefsChanges({ cwd: "/repo", limit: 100 }).pipe( Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cacheWithRefs(Option.none())), - Effect.provideService( - Persistence.EnvironmentCacheStore, - cacheWithRefs(Option.some(CACHED_REFS)), - ), ), ).pipe(Stream.runHead); const fiber = yield* Effect.forkChild(result); @@ -155,24 +509,80 @@ } expect(yield* Ref.get(calls)).toBe(1); - yield* TestClock.adjust("5 seconds"); + yield* TestClock.adjust("1 second"); expect(Option.getOrThrow(yield* Fiber.join(fiber))).toEqual(LIVE_REFS); + expect(yield* Ref.get(calls)).toBe(2); }).pipe(Effect.provide(TestClock.layer())), ), ); - it.effect("revalidates connected refs every five seconds", () => + it.effect("cancels an in-flight refresh when the connection generation changes", () => Effect.scoped( Effect.gen(function* () { const calls = yield* Ref.make(0); + const interruptions = yield* Ref.make(0); + const connectionState = yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE); const client = { [WS_METHODS.vcsListRefs]: () => Ref.updateAndGet(calls, (count) => count + 1).pipe( + Effect.flatMap((count) => + count === 1 - Effect.map((count) => (count === 1 ? CACHED_REFS : LIVE_REFS)), + ? Effect.never.pipe( + Effect.onInterrupt(() => + Ref.update(interruptions, (interruptions) => interruptions + 1), + ), + ) + : Effect.succeed(LIVE_REFS), + ), ), } as unknown as WsRpcProtocolClient; const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ target: TARGET, + state: connectionState, + session: yield* SubscriptionRef.make(Option.some(session(client))), + prepared: yield* SubscriptionRef.make(Option.none()), + connect: Effect.void, + disconnect: Effect.void, + retryNow: Effect.void, + } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]); + + const result = Stream.unwrap( + makeCachedVcsRefsChanges({ cwd: "/repo", limit: 100 }).pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cacheWithRefs(Option.none())), + ), + ).pipe(Stream.runHead); + const fiber = yield* Effect.forkChild(result); + + for (let attempt = 0; attempt < 100 && (yield* Ref.get(calls)) < 1; attempt += 1) { + yield* Effect.yieldNow; + } + yield* SubscriptionRef.set(connectionState, AVAILABLE_CONNECTION_STATE); + for (let attempt = 0; attempt < 100 && (yield* Ref.get(interruptions)) < 1; attempt += 1) { + yield* Effect.yieldNow; + } + expect(yield* Ref.get(interruptions)).toBe(1); + + yield* SubscriptionRef.set(connectionState, { + ...CONNECTED_CONNECTION_STATE, + generation: 2, + }); + expect(Option.getOrThrow(yield* Fiber.join(fiber))).toEqual(LIVE_REFS); + expect(yield* Ref.get(calls)).toBe(2); + }), + ), + ); + + it.effect("does not poll refs while the connection remains stable", () => + Effect.scoped( + Effect.gen(function* () { + const calls = yield* Ref.make(0); + const client = { + [WS_METHODS.vcsListRefs]: () => + Ref.update(calls, (count) => count + 1).pipe(Effect.as(CACHED_REFS)), + } as unknown as WsRpcProtocolClient; + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE), session: yield* SubscriptionRef.make(Option.some(session(client))), prepared: yield* SubscriptionRef.make(Option.none()), @@ -180,26 +590,28 @@ disconnect: Effect.void, retryNow: Effect.void, } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]); + const stream = Stream.unwrap( - const results = Stream.unwrap( makeCachedVcsRefsChanges({ cwd: "/repo", limit: 100 }).pipe( Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), Effect.provideService(Persistence.EnvironmentCacheStore, cacheWithRefs(Option.none())), ), - ).pipe(Stream.take(2), Stream.runCollect); - const fiber = yield* Effect.forkChild(results); + ).pipe(Stream.runDrain); + const fiber = yield* Effect.forkChild(stream); for (let attempt = 0; attempt < 100 && (yield* Ref.get(calls)) < 1; attempt += 1) { yield* Effect.yieldNow; } expect(yield* Ref.get(calls)).toBe(1); + yield* TestClock.adjust("1 minute"); + yield* Effect.yieldNow; - yield* TestClock.adjust("5 seconds"); + expect(yield* Ref.get(calls)).toBe(1); - expect(Array.from(yield* Fiber.join(fiber))).toEqual([CACHED_REFS, LIVE_REFS]); + yield* Fiber.interrupt(fiber); }).pipe(Effect.provide(TestClock.layer())), ), ); - it.effect("does not emit persisted refs before a live refresh", () => + it.effect("emits persisted refs before a live refresh", () => Effect.scoped( Effect.gen(function* () { const client = { @@ -223,9 +635,9 @@ cacheWithRefs(Option.some(CACHED_REFS)), ), ), + ).pipe(Stream.take(2), Stream.runCollect); - ).pipe(Stream.runHead); + expect(refs).toEqual([CACHED_REFS, LIVE_REFS]); - expect(Option.getOrThrow(refs)).toEqual(LIVE_REFS); }), ), ); ``` [Next page](https://git.cafe/pingdotgg/t3code/pull/4727/diffs?after=InBhY2thZ2VzL2NsaWVudC1ydW50aW1lL3NyYy9zdGF0ZS92Y3MudGVzdC50cyI&expectedVersion=45&format=markdown)