authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-07-29 00:42:50+02:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-07-29 00:42:50+02:00
log38a6e3ce65b1d4fa58788faa96ff9410a8c26ffe
treea5f780653c12e2e13649499a3a540d601ba66c2d
parentf2d2fb2f2444bcabc31bf906b5b334aee0d9d79c
signature Signed by PGP key B5690EEEBB952194

Fix Git ref refresh resource storms (#4727)

Co-authored-by: codex <codex@users.noreply.github.com>

25 files changed, 2349 insertions(+), 432 deletions(-)

apps/mobile/src/connection/environment-cache-store.test.ts+36
......@@ -42,6 +42,12 @@ function makeDatabase() {
4242 removed.push(id);
4343 values.delete(id);
4444 }),
45 clearCacheKind: (environmentId, kind) =>
46 Effect.sync(() => {
47 for (const key of values.keys()) {
48 if (key.startsWith(`${environmentId}:${kind}:`)) values.delete(key);
49 }
50 }),
4551 clearEnvironmentCache: (environmentId) =>
4652 Effect.sync(() => {
4753 for (const key of values.keys()) {
......@@ -80,6 +86,36 @@ describe("mobile SQLite environment cache store", () => {
8086 }),
8187 );
8288
89 it.effect("removes one persisted VCS ref snapshot", () =>
90 Effect.gen(function* () {
91 const memory = makeDatabase();
92 const store = yield* make().pipe(Effect.provideService(MobileDatabase, memory.database));
93 yield* store.saveVcsRefs(ENVIRONMENT_ID, "/repo", REFS);
94
95 yield* store.removeVcsRefs(ENVIRONMENT_ID, "/repo");
96
97 expect(yield* store.loadVcsRefs(ENVIRONMENT_ID, "/repo")).toEqual(Option.none());
98 expect(memory.removed).toContain(cacheId(ENVIRONMENT_ID, "vcs-refs", "/repo"));
99 }),
100 );
101
102 it.effect("clears every persisted VCS ref snapshot in one environment", () =>
103 Effect.gen(function* () {
104 const memory = makeDatabase();
105 const store = yield* make().pipe(Effect.provideService(MobileDatabase, memory.database));
106 const otherEnvironmentId = EnvironmentId.make("environment-2");
107 yield* store.saveVcsRefs(ENVIRONMENT_ID, "/repo", REFS);
108 yield* store.saveVcsRefs(ENVIRONMENT_ID, "/repo-worktree", REFS);
109 yield* store.saveVcsRefs(otherEnvironmentId, "/repo", REFS);
110
111 yield* store.clearVcsRefs(ENVIRONMENT_ID);
112
113 expect(yield* store.loadVcsRefs(ENVIRONMENT_ID, "/repo")).toEqual(Option.none());
114 expect(yield* store.loadVcsRefs(ENVIRONMENT_ID, "/repo-worktree")).toEqual(Option.none());
115 expect(yield* store.loadVcsRefs(otherEnvironmentId, "/repo")).toEqual(Option.some(REFS));
116 }),
117 );
118
83119 it.effect("clears one environment without touching another", () =>
84120 Effect.gen(function* () {
85121 const memory = makeDatabase();
apps/mobile/src/connection/environment-cache-store.ts+10
......@@ -223,6 +223,16 @@ export const make = Effect.fn("MobileEnvironmentCacheStore.make")(function* () {
223223 .pipe(Effect.mapError(mapDatabaseError("save-vcs-refs")));
224224 },
225225 ),
226 removeVcsRefs: Effect.fn("MobileEnvironmentCache.removeVcsRefs")((environmentId, cwd) =>
227 database
228 .removeCache(environmentId, "vcs-refs", cwd)
229 .pipe(Effect.mapError(mapDatabaseError("remove-vcs-refs"))),
230 ),
231 clearVcsRefs: Effect.fn("MobileEnvironmentCache.clearVcsRefs")((environmentId) =>
232 database
233 .clearCacheKind(environmentId, "vcs-refs")
234 .pipe(Effect.mapError(mapDatabaseError("clear-vcs-refs"))),
235 ),
226236 clear: Effect.fn("MobileEnvironmentCache.clear")((environmentId) =>
227237 database
228238 .clearEnvironmentCache(environmentId)
apps/mobile/src/persistence/mobile-database.ts+17
......@@ -46,6 +46,7 @@ const MobileDatabaseOperation = Schema.Literals([
4646 "load-cache",
4747 "save-cache",
4848 "remove-cache",
49 "clear-cache-kind",
4950 "clear-environment-cache",
5051 "clear-all-caches",
5152 "inspect-caches",
......@@ -203,6 +204,10 @@ export class MobileDatabase extends Context.Service<
203204 kind: ClientCacheKind,
204205 cacheKey: string,
205206 ) => Effect.Effect<void, MobileDatabaseError>;
207 readonly clearCacheKind: (
208 environmentId: EnvironmentId,
209 kind: ClientCacheKind,
210 ) => Effect.Effect<void, MobileDatabaseError>;
206211 readonly clearEnvironmentCache: (
207212 environmentId: EnvironmentId,
208213 ) => Effect.Effect<void, MobileDatabaseError>;
......@@ -322,6 +327,17 @@ const makeAvailable = Effect.gen(function* () {
322327 catch: databaseError("remove-cache"),
323328 }).pipe(Effect.asVoid),
324329 ),
330 clearCacheKind: Effect.fn("MobileDatabase.clearCacheKind")((environmentId, kind) =>
331 Effect.tryPromise({
332 try: () =>
333 database.runAsync(
334 "DELETE FROM client_cache WHERE environment_id = ? AND kind = ?",
335 environmentId,
336 kind,
337 ),
338 catch: databaseError("clear-cache-kind"),
339 }).pipe(Effect.asVoid),
340 ),
325341 clearEnvironmentCache: Effect.fn("MobileDatabase.clearEnvironmentCache")((environmentId) =>
326342 Effect.tryPromise({
327343 try: () =>
......@@ -392,6 +408,7 @@ function makeUnavailable(error: MobileDatabaseError): MobileDatabase["Service"]
392408 loadCache: () => fail,
393409 saveCache: () => fail,
394410 removeCache: () => fail,
411 clearCacheKind: () => fail,
395412 clearEnvironmentCache: () => fail,
396413 clearAllCaches: fail,
397414 inspectCaches: fail,
apps/server/src/git/GitManager.ts+1-1
......@@ -1794,7 +1794,7 @@ export const make = Effect.gen(function* () {
17941794 resolvePullRequestWorktreeLocalBranchName(pullRequestWithRemoteInfo);
17951795
17961796 const findLocalHeadBranch = Effect.fn("findLocalHeadBranch")(function* (cwd: string) {
1797 const result = yield* gitCore.listRefs({ cwd });
1797 const result = yield* gitCore.listRefs({ cwd, refresh: true });
17981798 const localBranch = result.refs.find(
17991799 (branch) => !branch.isRemote && branch.name === localPullRequestBranch,
18001800 );
apps/server/src/vcs/GitVcsDriverCore.test.ts+471-2
......@@ -1,18 +1,22 @@
11import * as NodeServices from "@effect/platform-node/NodeServices";
22import { assert, it, describe } from "@effect/vitest";
3import * as Deferred from "effect/Deferred";
34import * as Effect from "effect/Effect";
45import * as FileSystem from "effect/FileSystem";
6import * as Fiber from "effect/Fiber";
57import * as Layer from "effect/Layer";
68import * as Path from "effect/Path";
79import * as PlatformError from "effect/PlatformError";
10import * as Ref from "effect/Ref";
811import * as Scope from "effect/Scope";
912import * as Sink from "effect/Sink";
1013import * as Stream from "effect/Stream";
14import * as TestClock from "effect/testing/TestClock";
1115import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process";
1216
1317import { GitCommandError } from "@t3tools/contracts";
1418import { ServerConfig } from "../config.ts";
15import { splitNullSeparatedGitStdoutPaths } from "./GitVcsDriverCore.ts";
19import { makeGitVcsDriverCore, splitNullSeparatedGitStdoutPaths } from "./GitVcsDriverCore.ts";
1620import * as GitVcsDriver from "./GitVcsDriver.ts";
1721
1822const ServerConfigLayer = ServerConfig.layerTest(process.cwd(), {
......@@ -38,6 +42,21 @@ const makeNonRepositoryHandle = () =>
3842 getOutputFd: () => Stream.empty,
3943 });
4044
45const makeSuccessfulHandle = (stdout: string) =>
46 ChildProcessSpawner.makeHandle({
47 pid: ChildProcessSpawner.ProcessId(1),
48 exitCode: Effect.succeed(ChildProcessSpawner.ExitCode(0)),
49 isRunning: Effect.succeed(false),
50 kill: () => Effect.void,
51 unref: Effect.succeed(Effect.void),
52 stdin: Sink.drain,
53 stdout: Stream.encodeText(Stream.make(stdout)),
54 stderr: Stream.empty,
55 all: Stream.empty,
56 getInputFd: () => Sink.drain,
57 getOutputFd: () => Stream.empty,
58 });
59
4160const makeTmpDir = (
4261 prefix = "git-vcs-driver-test-",
4362): Effect.Effect<string, PlatformError.PlatformError, FileSystem.FileSystem | Scope.Scope> =>
......@@ -129,11 +148,434 @@ it.effect("uses stable diagnostics for every parsed non-repository command", ()
129148 assert.deepStrictEqual(commands, [
130149 { args: ["status", "--porcelain=2", "--branch"], lcAll: "C" },
131150 { args: ["rev-parse", "--abbrev-ref", "HEAD"], lcAll: "C" },
132 { args: ["branch", "--no-color", "--no-column"], lcAll: "C" },
151 { args: ["rev-parse", "--git-common-dir"], lcAll: "C" },
133152 ]);
134153 }).pipe(Effect.provide(layer));
135154});
136155
156it.effect("coalesces concurrent ref pages into one repository snapshot", () =>
157 Effect.scoped(
158 Effect.gen(function* () {
159 const delegate = yield* ChildProcessSpawner.ChildProcessSpawner;
160 const spawnedArgs = yield* Ref.make<ReadonlyArray<ReadonlyArray<string>>>([]);
161 const firstWorktreeScanStarted = yield* Deferred.make<void>();
162 const remoteNamesScanCompleted = yield* Deferred.make<void>();
163 const delayFirstWorktreeScan = yield* Ref.make(true);
164 const countingSpawner = ChildProcessSpawner.make((command) =>
165 Effect.gen(function* () {
166 if (!ChildProcess.isStandardCommand(command)) {
167 return yield* Effect.die("expected a standard Git command");
168 }
169 yield* Ref.update(spawnedArgs, (current) => [...current, command.args]);
170 const isWorktreeScan =
171 command.args.includes("worktree") && command.args.includes("--porcelain");
172 const shouldDelay =
173 isWorktreeScan && (yield* Ref.getAndSet(delayFirstWorktreeScan, false));
174 if (shouldDelay) {
175 yield* Deferred.succeed(firstWorktreeScanStarted, undefined);
176 yield* Effect.sleep("8 seconds");
177 }
178 const handle = yield* delegate.spawn(command);
179 const isRemoteNamesScan =
180 command.args.length === 3 &&
181 command.args[0] === "--git-dir" &&
182 command.args[2] === "remote";
183 return isRemoteNamesScan
184 ? ChildProcessSpawner.makeHandle({
185 ...handle,
186 exitCode: handle.exitCode.pipe(
187 Effect.tap(() => Deferred.succeed(remoteNamesScanCompleted, undefined)),
188 ),
189 })
190 : handle;
191 }),
192 );
193 const driver = yield* makeGitVcsDriverCore().pipe(
194 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, countingSpawner),
195 );
196 const cwd = yield* makeTmpDir();
197 const runGit = (args: ReadonlyArray<string>) =>
198 driver.execute({
199 operation: "GitVcsDriver.test.coalescedListRefs",
200 cwd,
201 args,
202 timeoutMs: 10_000,
203 });
204
205 yield* driver.initRepo({ cwd });
206 yield* runGit(["config", "user.email", "test@test.com"]);
207 yield* runGit(["config", "user.name", "Test"]);
208 yield* writeTextFile(cwd, "README.md", "# test\n");
209 yield* runGit(["add", "."]);
210 yield* runGit(["commit", "-m", "initial commit"]);
211 yield* Ref.set(spawnedArgs, []);
212
213 const initialRequest = yield* driver
214 .listRefs({ cwd, refresh: true, limit: 100 })
215 .pipe(Effect.forkChild({ startImmediately: true }));
216 yield* Deferred.await(firstWorktreeScanStarted);
217 yield* Deferred.await(remoteNamesScanCompleted);
218 yield* TestClock.adjust("6 seconds");
219 const laterRequests = yield* Effect.all(
220 Array.from({ length: 30 }, (_, index) =>
221 driver.listRefs({
222 cwd,
223 refresh: true,
224 query: `missing-${index}`,
225 limit: 100,
226 }),
227 ),
228 { concurrency: "unbounded" },
229 ).pipe(Effect.forkChild({ startImmediately: true }));
230 yield* TestClock.adjust("2 seconds");
231 yield* Fiber.join(initialRequest);
232 yield* Fiber.join(laterRequests);
233 yield* driver.listRefs({ cwd, cursor: 1, limit: 100 });
234
235 const firstSnapshotCommands = yield* Ref.get(spawnedArgs);
236 const snapshotRefScans = firstSnapshotCommands.filter(
237 (args) =>
238 args.includes("for-each-ref") &&
239 args.includes("refs/heads") &&
240 args.includes("refs/remotes"),
241 );
242 const worktreeScans = firstSnapshotCommands.filter(
243 (args) => args.includes("worktree") && args.includes("--porcelain"),
244 );
245 assert.equal(snapshotRefScans.length, 1);
246 assert.equal(worktreeScans.length, 1);
247
248 yield* driver.createRef({ cwd, refName: "feature/cache-invalidation" });
249 const refreshed = yield* driver.listRefs({ cwd, limit: 100 });
250 assert.equal(
251 refreshed.refs.some((ref) => ref.name === "feature/cache-invalidation"),
252 true,
253 );
254 const allCommands = yield* Ref.get(spawnedArgs);
255 assert.equal(
256 allCommands.filter(
257 (args) =>
258 args.includes("for-each-ref") &&
259 args.includes("refs/heads") &&
260 args.includes("refs/remotes"),
261 ).length,
262 2,
263 );
264 }),
265 ).pipe(Effect.provide(ServerConfigLayer.pipe(Layer.provideMerge(NodeServices.layer)))),
266);
267
268it.effect("retries an in-flight ref snapshot invalidated by a mutation", () =>
269 Effect.scoped(
270 Effect.gen(function* () {
271 const delegate = yield* ChildProcessSpawner.ChildProcessSpawner;
272 const firstWorktreeScanStarted = yield* Deferred.make<void>();
273 const firstRefScanCompleted = yield* Deferred.make<void>();
274 const releaseFirstWorktreeScan = yield* Deferred.make<void>();
275 const delayFirstWorktreeScan = yield* Ref.make(true);
276 const refScans = yield* Ref.make(0);
277 const coordinatingSpawner = ChildProcessSpawner.make((command) =>
278 Effect.gen(function* () {
279 if (!ChildProcess.isStandardCommand(command)) {
280 return yield* Effect.die("expected a standard Git command");
281 }
282 const isWorktreeScan =
283 command.args.includes("worktree") && command.args.includes("--porcelain");
284 if (isWorktreeScan && (yield* Ref.getAndSet(delayFirstWorktreeScan, false))) {
285 yield* Deferred.succeed(firstWorktreeScanStarted, undefined);
286 yield* Deferred.await(releaseFirstWorktreeScan);
287 }
288 const handle = yield* delegate.spawn(command);
289 const isRefScan =
290 command.args.includes("for-each-ref") &&
291 command.args.includes("refs/heads") &&
292 command.args.includes("refs/remotes");
293 if (!isRefScan) return handle;
294 const scan = yield* Ref.updateAndGet(refScans, (count) => count + 1);
295 return scan === 1
296 ? ChildProcessSpawner.makeHandle({
297 ...handle,
298 exitCode: handle.exitCode.pipe(
299 Effect.tap(() => Deferred.succeed(firstRefScanCompleted, undefined)),
300 ),
301 })
302 : handle;
303 }),
304 );
305 const driver = yield* makeGitVcsDriverCore().pipe(
306 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, coordinatingSpawner),
307 );
308 const cwd = yield* makeTmpDir();
309 yield* initRepoWithCommit(cwd).pipe(Effect.provideService(GitVcsDriver.GitVcsDriver, driver));
310
311 const inFlight = yield* driver
312 .listRefs({ cwd, refresh: true, limit: 100 })
313 .pipe(Effect.forkChild({ startImmediately: true }));
314 yield* Deferred.await(firstWorktreeScanStarted);
315 yield* Deferred.await(firstRefScanCompleted);
316
317 yield* driver.createRef({ cwd, refName: "feature/during-refresh" });
318 yield* Deferred.succeed(releaseFirstWorktreeScan, undefined);
319
320 const refs = yield* Fiber.join(inFlight);
321 assert.isTrue(refs.refs.some((ref) => ref.name === "feature/during-refresh"));
322 assert.equal(yield* Ref.get(refScans), 2);
323 }),
324 ).pipe(Effect.provide(ServerConfigLayer.pipe(Layer.provideMerge(NodeServices.layer)))),
325);
326
327it.effect("invalidates a ref snapshot when a mutation fails after changing Git", () =>
328 Effect.scoped(
329 Effect.gen(function* () {
330 const delegate = yield* ChildProcessSpawner.ChildProcessSpawner;
331 const partiallyFailingSpawner = ChildProcessSpawner.make((command) =>
332 Effect.gen(function* () {
333 if (!ChildProcess.isStandardCommand(command)) {
334 return yield* Effect.die("expected a standard Git command");
335 }
336 if (command.args[0] === "branch" && command.args[1] === "feature/partial-failure") {
337 const handle = yield* delegate.spawn(command);
338 yield* handle.exitCode;
339 return makeNonRepositoryHandle();
340 }
341 return yield* delegate.spawn(command);
342 }),
343 );
344 const driver = yield* makeGitVcsDriverCore().pipe(
345 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, partiallyFailingSpawner),
346 );
347 const cwd = yield* makeTmpDir();
348 yield* initRepoWithCommit(cwd).pipe(Effect.provideService(GitVcsDriver.GitVcsDriver, driver));
349 yield* driver.listRefs({ cwd, refresh: true });
350
351 yield* driver.createRef({ cwd, refName: "feature/partial-failure" }).pipe(Effect.flip);
352
353 const refs = yield* driver.listRefs({ cwd });
354 assert.isTrue(refs.refs.some((ref) => ref.name === "feature/partial-failure"));
355 }),
356 ).pipe(Effect.provide(ServerConfigLayer.pipe(Layer.provideMerge(NodeServices.layer)))),
357);
358
359it.effect("fails a ref snapshot when for-each-ref exits unsuccessfully", () =>
360 Effect.scoped(
361 Effect.gen(function* () {
362 const delegate = yield* ChildProcessSpawner.ChildProcessSpawner;
363 const snapshotAttempts = yield* Ref.make(0);
364 const failingSnapshotSpawner = ChildProcessSpawner.make((command) =>
365 Effect.gen(function* () {
366 if (!ChildProcess.isStandardCommand(command)) {
367 return yield* Effect.die("expected a standard Git command");
368 }
369 if (command.args.includes("for-each-ref")) {
370 yield* Ref.update(snapshotAttempts, (count) => count + 1);
371 return makeNonRepositoryHandle();
372 }
373 return yield* delegate.spawn(command);
374 }),
375 );
376 const driver = yield* makeGitVcsDriverCore().pipe(
377 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, failingSnapshotSpawner),
378 );
379 const cwd = yield* makeTmpDir();
380 yield* initRepoWithCommit(cwd).pipe(Effect.provideService(GitVcsDriver.GitVcsDriver, driver));
381
382 const error = yield* driver.listRefs({ cwd, refresh: true }).pipe(Effect.flip);
383
384 assert.deepInclude(error, {
385 _tag: "GitCommandError",
386 operation: "GitVcsDriver.listRefs.snapshotRefs",
387 detail: "Git ref snapshot enumeration failed.",
388 exitCode: 128,
389 });
390 assert.equal(yield* Ref.get(snapshotAttempts), 1);
391 }),
392 ).pipe(Effect.provide(ServerConfigLayer.pipe(Layer.provideMerge(NodeServices.layer)))),
393);
394
395it.effect("marks the current branch when worktree metadata is unavailable", () =>
396 Effect.scoped(
397 Effect.gen(function* () {
398 const delegate = yield* ChildProcessSpawner.ChildProcessSpawner;
399 const incompleteMetadataSpawner = ChildProcessSpawner.make((command) =>
400 Effect.gen(function* () {
401 if (!ChildProcess.isStandardCommand(command)) {
402 return yield* Effect.die("expected a standard Git command");
403 }
404 const isWorktreeRoot =
405 command.args.includes("rev-parse") && command.args.includes("--show-toplevel");
406 const isWorktreeList =
407 command.args.includes("worktree") && command.args.includes("--porcelain");
408 if (isWorktreeRoot || isWorktreeList) {
409 return makeNonRepositoryHandle();
410 }
411 return yield* delegate.spawn(command);
412 }),
413 );
414 const driver = yield* makeGitVcsDriverCore().pipe(
415 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, incompleteMetadataSpawner),
416 );
417 const cwd = yield* makeTmpDir();
418 const { initialBranch } = yield* initRepoWithCommit(cwd).pipe(
419 Effect.provideService(GitVcsDriver.GitVcsDriver, driver),
420 );
421
422 const refs = yield* driver.listRefs({ cwd, refresh: true });
423
424 assert.isTrue(refs.isRepo);
425 assert.isTrue(refs.refs.find((ref) => ref.name === initialBranch)?.current);
426 }),
427 ).pipe(Effect.provide(ServerConfigLayer.pipe(Layer.provideMerge(NodeServices.layer)))),
428);
429
430it.effect("ignores worktree metadata for directories that no longer exist", () =>
431 Effect.scoped(
432 Effect.gen(function* () {
433 const delegate = yield* ChildProcessSpawner.ChildProcessSpawner;
434 const missingWorktreePath = "/missing/deleted-worktree";
435 const staleWorktreeSpawner = ChildProcessSpawner.make((command) =>
436 Effect.gen(function* () {
437 if (!ChildProcess.isStandardCommand(command)) {
438 return yield* Effect.die("expected a standard Git command");
439 }
440 const isWorktreeList =
441 command.args.includes("worktree") && command.args.includes("--porcelain");
442 if (isWorktreeList) {
443 return makeSuccessfulHandle(
444 `worktree ${missingWorktreePath}\0HEAD deadbeef\0branch refs/heads/stale-worktree\0\0`,
445 );
446 }
447 return yield* delegate.spawn(command);
448 }),
449 );
450 const driver = yield* makeGitVcsDriverCore().pipe(
451 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, staleWorktreeSpawner),
452 );
453 const cwd = yield* makeTmpDir();
454 yield* initRepoWithCommit(cwd).pipe(Effect.provideService(GitVcsDriver.GitVcsDriver, driver));
455 yield* git(cwd, ["branch", "stale-worktree"]).pipe(
456 Effect.provideService(GitVcsDriver.GitVcsDriver, driver),
457 );
458
459 const refs = yield* driver.listRefs({ cwd, refresh: true });
460
461 assert.equal(refs.refs.find((ref) => ref.name === "stale-worktree")?.worktreePath, null);
462 }),
463 ).pipe(Effect.provide(ServerConfigLayer.pipe(Layer.provideMerge(NodeServices.layer)))),
464);
465
466it.effect("refreshes the current branch after an external checkout", () =>
467 Effect.scoped(
468 Effect.gen(function* () {
469 const driver = yield* GitVcsDriver.GitVcsDriver;
470 const cwd = yield* makeTmpDir();
471 const { initialBranch } = yield* initRepoWithCommit(cwd);
472 yield* git(cwd, ["branch", "external-checkout"]);
473
474 const initialRefs = yield* driver.listRefs({ cwd, refresh: true });
475 assert.isTrue(initialRefs.refs.find((ref) => ref.name === initialBranch)?.current);
476
477 // Raw execute intentionally bypasses the driver's mutation invalidation,
478 // matching a checkout performed by another process.
479 yield* driver.execute({
480 operation: "GitVcsDriver.test.externalCheckout",
481 cwd,
482 args: ["checkout", "external-checkout"],
483 timeoutMs: 10_000,
484 });
485 yield* TestClock.adjust("6 seconds");
486
487 const refreshedRefs = yield* driver.listRefs({ cwd, refresh: true });
488 assert.isTrue(refreshedRefs.refs.find((ref) => ref.name === "external-checkout")?.current);
489 assert.isFalse(refreshedRefs.refs.find((ref) => ref.name === initialBranch)?.current);
490 }),
491 ).pipe(Effect.provide(TestLayer)),
492);
493
494it.effect("backs off failed upstream refreshes across linked worktrees", () =>
495 Effect.scoped(
496 Effect.gen(function* () {
497 const delegate = yield* ChildProcessSpawner.ChildProcessSpawner;
498 const fetchAttempts = yield* Ref.make(0);
499 const failingFetchSpawner = ChildProcessSpawner.make((command) =>
500 Effect.gen(function* () {
501 if (!ChildProcess.isStandardCommand(command)) {
502 return yield* Effect.die("expected a standard Git command");
503 }
504 if (command.args.includes("fetch") && command.args.includes("--quiet")) {
505 yield* Ref.update(fetchAttempts, (count) => count + 1);
506 return makeNonRepositoryHandle();
507 }
508 return yield* delegate.spawn(command);
509 }),
510 );
511 const driver = yield* makeGitVcsDriverCore().pipe(
512 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, failingFetchSpawner),
513 );
514 const fileSystem = yield* FileSystem.FileSystem;
515 const cwd = yield* makeTmpDir();
516 const remote = yield* makeTmpDir("git-vcs-driver-remote-");
517 const worktreesRoot = yield* makeTmpDir("git-vcs-driver-worktrees-");
518 const pathService = yield* Path.Path;
519 const worktreePath = pathService.join(worktreesRoot, "linked");
520 const runGit = (workingDirectory: string, args: ReadonlyArray<string>) =>
521 driver.execute({
522 operation: "GitVcsDriver.test.upstreamRefreshBackoff",
523 cwd: workingDirectory,
524 args,
525 timeoutMs: 10_000,
526 });
527
528 yield* driver.initRepo({ cwd });
529 yield* runGit(cwd, ["config", "user.email", "test@test.com"]);
530 yield* runGit(cwd, ["config", "user.name", "Test"]);
531 yield* writeTextFile(cwd, "README.md", "# test\n");
532 yield* runGit(cwd, ["add", "."]);
533 yield* runGit(cwd, ["commit", "-m", "initial commit"]);
534 const initialBranch = (yield* runGit(cwd, ["branch", "--show-current"])).stdout.trim();
535 yield* runGit(remote, ["init", "--bare"]);
536 yield* runGit(cwd, ["remote", "add", "origin", remote]);
537 yield* runGit(cwd, ["push", "-u", "origin", initialBranch]);
538 yield* runGit(cwd, ["worktree", "add", "-b", "feature/linked", worktreePath]);
539 yield* runGit(worktreePath, [
540 "branch",
541 "--set-upstream-to",
542 `origin/${initialBranch}`,
543 "feature/linked",
544 ]);
545 const rootCommonDir = (yield* runGit(cwd, ["rev-parse", "--git-common-dir"])).stdout.trim();
546 const linkedCommonDir = (yield* runGit(worktreePath, [
547 "rev-parse",
548 "--git-common-dir",
549 ])).stdout.trim();
550 assert.equal(
551 yield* fileSystem.realPath(pathService.resolve(cwd, rootCommonDir)),
552 yield* fileSystem.realPath(pathService.resolve(worktreePath, linkedCommonDir)),
553 );
554 yield* Ref.set(fetchAttempts, 0);
555
556 yield* driver.statusDetailsRemote(cwd);
557 yield* driver.statusDetailsRemote(worktreePath);
558 assert.equal(yield* Ref.get(fetchAttempts), 1);
559
560 yield* TestClock.adjust("29 seconds");
561 yield* driver.statusDetailsRemote(worktreePath);
562 assert.equal(yield* Ref.get(fetchAttempts), 1);
563
564 yield* TestClock.adjust("1 second");
565 yield* driver.statusDetailsRemote(cwd);
566 assert.equal(yield* Ref.get(fetchAttempts), 2);
567
568 yield* TestClock.adjust("59 seconds");
569 yield* driver.statusDetailsRemote(worktreePath);
570 assert.equal(yield* Ref.get(fetchAttempts), 2);
571
572 yield* TestClock.adjust("1 second");
573 yield* driver.statusDetailsRemote(cwd);
574 assert.equal(yield* Ref.get(fetchAttempts), 3);
575 }),
576 ).pipe(Effect.provide(ServerConfigLayer.pipe(Layer.provideMerge(NodeServices.layer)))),
577);
578
137579it.layer(TestLayer)("GitVcsDriver core integration", (it) => {
138580 describe("process environment", () => {
139581 it.effect("preserves the caller locale for general Git subprocesses", () =>
......@@ -661,6 +1103,33 @@ it.layer(TestLayer)("GitVcsDriver core integration", (it) => {
6611103 });
6621104
6631105 describe("worktree operations", () => {
1106 it.effect("preserves newline characters in worktree paths when listing refs", () =>
1107 Effect.gen(function* () {
1108 const cwd = yield* makeTmpDir();
1109 yield* initRepoWithCommit(cwd);
1110 const worktreesRoot = yield* makeTmpDir("git-vcs-driver-worktrees-");
1111 const fileSystem = yield* FileSystem.FileSystem;
1112 const pathService = yield* Path.Path;
1113 const worktreePath = pathService.join(worktreesRoot, "linked\nworktree");
1114 const driver = yield* GitVcsDriver.GitVcsDriver;
1115
1116 yield* git(cwd, ["worktree", "add", "-b", "feature/newline-path", worktreePath]);
1117
1118 const refs = yield* driver.listRefs({ cwd, refresh: true });
1119 const listedPath = refs.refs.find(
1120 (ref) => ref.name === "feature/newline-path",
1121 )?.worktreePath;
1122
1123 if (typeof listedPath !== "string") {
1124 return assert.fail("expected the linked branch to include its worktree path");
1125 }
1126 assert.equal(
1127 yield* fileSystem.realPath(listedPath),
1128 yield* fileSystem.realPath(worktreePath),
1129 );
1130 }),
1131 );
1132
6641133 it.effect("creates and removes a worktree for a new refName", () =>
6651134 Effect.gen(function* () {
6661135 const cwd = yield* makeTmpDir();
apps/server/src/vcs/GitVcsDriverCore.ts+528-313
......@@ -50,8 +50,17 @@ const WORKSPACE_FILES_MAX_OUTPUT_BYTES = 120_000;
5050const STATUS_UPSTREAM_REFRESH_INTERVAL = Duration.seconds(15);
5151const STATUS_UPSTREAM_REFRESH_TIMEOUT = Duration.seconds(5);
5252
53const STATUS_UPSTREAM_REFRESH_FAILURE_COOLDOWN = Duration.seconds(5);
53const STATUS_UPSTREAM_REFRESH_FAILURE_BASE_COOLDOWN = Duration.seconds(30);
54const STATUS_UPSTREAM_REFRESH_FAILURE_MAX_COOLDOWN = Duration.minutes(15);
5455const STATUS_UPSTREAM_REFRESH_CACHE_CAPACITY = 2_048;
56const REPOSITORY_PATHS_CACHE_CAPACITY = 2_048;
57const REPOSITORY_PATHS_CACHE_TTL = Duration.minutes(10);
58const REPOSITORY_PATHS_REFRESH_COALESCE_TTL = Duration.seconds(5);
59const NON_REPOSITORY_PATHS_CACHE_TTL = Duration.seconds(1);
60const LIST_REFS_SNAPSHOT_CACHE_CAPACITY = 64;
61const LIST_REFS_SNAPSHOT_CACHE_TTL = Duration.minutes(2);
62const LIST_REFS_REFRESH_COALESCE_TTL = Duration.seconds(5);
63const LIST_REFS_REFRESH_FAILURE_COOLDOWN = Duration.seconds(30);
5564const STATUS_UPSTREAM_REFRESH_ENV = Object.freeze({
5665 GCM_INTERACTIVE: "never",
5766 GIT_ASKPASS: "",
......@@ -95,6 +104,35 @@ class StatusRemoteRefreshCacheKey extends Data.Class<{
95104 remoteName: string;
96105}> {}
97106
107function statusUpstreamRefreshFailureCooldown(consecutiveFailures: number): Duration.Duration {
108 const exponent = Math.max(0, consecutiveFailures - 1);
109 const cooldownMs =
110 Duration.toMillis(STATUS_UPSTREAM_REFRESH_FAILURE_BASE_COOLDOWN) * Math.pow(2, exponent);
111 return Duration.min(Duration.millis(cooldownMs), STATUS_UPSTREAM_REFRESH_FAILURE_MAX_COOLDOWN);
112}
113
114class GitRefsSnapshotCacheKey extends Data.Class<{
115 gitCommonDir: string;
116 epoch: number;
117}> {}
118
119class GitRefsRefreshCacheKey extends Data.Class<{
120 gitCommonDir: string;
121 generation: number;
122}> {}
123
124interface GitRepositoryPaths {
125 readonly gitCommonDir: string;
126 readonly worktreeRoot: string | null;
127 readonly currentBranch: string | null;
128}
129
130interface GitRefsSnapshot {
131 readonly localBranches: ReadonlyArray<VcsRef>;
132 readonly remoteBranches: ReadonlyArray<VcsRef>;
133 readonly hasPrimaryRemote: boolean;
134}
135
98136interface ExecuteGitOptions {
99137 stdin?: string | undefined;
100138 timeoutMs?: number | undefined;
......@@ -161,21 +199,6 @@ function parsePorcelainPath(line: string): string | null {
161199 return filePath.length > 0 ? filePath : null;
162200}
163201
164function parseBranchLine(line: string): { name: string; current: boolean } | null {
165 const trimmed = line.trim();
166 if (trimmed.length === 0) return null;
167
168 const name = trimmed.replace(/^[*+]\s+/, "");
169 // Exclude symbolic refs like: "origin/HEAD -> origin/main".
170 // Exclude detached HEAD pseudo-refs like: "(HEAD detached at origin/main)".
171 if (name.includes(" -> ") || name.startsWith("(")) return null;
172
173 return {
174 name,
175 current: trimmed.startsWith("* "),
176 };
177}
178
179202function filterBranchesForListQuery(
180203 refs: ReadonlyArray<VcsRef>,
181204 query?: string,
......@@ -210,6 +233,37 @@ function paginateBranches(input: {
210233 };
211234}
212235
236function parseWorktreeBranchPaths(stdout: string): ReadonlyMap<string, string> {
237 const worktreePaths = new Map<string, string>();
238 let currentPath: string | null = null;
239 let currentBranch: string | null = null;
240 let currentPrunable = false;
241
242 const flush = () => {
243 if (currentPath !== null && currentBranch !== null && !currentPrunable) {
244 worktreePaths.set(currentBranch, currentPath);
245 }
246 currentPath = null;
247 currentBranch = null;
248 currentPrunable = false;
249 };
250
251 for (const field of stdout.split("\0")) {
252 if (field === "") {
253 flush();
254 } else if (field.startsWith("worktree ")) {
255 currentPath = field.slice("worktree ".length);
256 } else if (field.startsWith("branch refs/heads/")) {
257 currentBranch = field.slice("branch refs/heads/".length);
258 } else if (field === "prunable" || field.startsWith("prunable ")) {
259 currentPrunable = true;
260 }
261 }
262 flush();
263
264 return worktreePaths;
265}
266
213267function splitNullSeparatedPaths(input: string, truncated: boolean): string[] {
214268 const parts = input.split("\0");
215269 if (parts.length === 0) return [];
......@@ -937,35 +991,183 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function*
937991 fetchCwd,
938992 ["--git-dir", gitCommonDir, "fetch", "--quiet", "--no-tags", remoteName],
939993 {
940 allowNonZeroExit: true,
941994 env: STATUS_UPSTREAM_REFRESH_ENV,
995 fallbackErrorDetail: "Background Git fetch exited with a non-zero status.",
942996 timeoutMs: Duration.toMillis(STATUS_UPSTREAM_REFRESH_TIMEOUT),
943997 },
944998 ).pipe(Effect.asVoid);
945999 };
9461000
1001 const resolveRepositoryPathsUncached = Effect.fn("resolveRepositoryPathsUncached")(function* (
1002 cwd: string,
1003 ) {
1004 const commonDirResult = yield* executeGitWithStableDiagnostics(
1005 "GitVcsDriver.resolveRepositoryPaths.commonDir",
1006 cwd,
1007 ["rev-parse", "--git-common-dir"],
1008 {
1009 timeoutMs: 5_000,
1010 allowNonZeroExit: true,
1011 },
1012 );
1013 if (commonDirResult.exitCode !== 0) {
1014 const stderr = commonDirResult.stderr.trim();
1015 if (isNonRepositoryGitStderr(stderr)) {
1016 return null;
1017 }
1018 return yield* new GitCommandError({
1019 ...gitCommandContext({
1020 operation: "GitVcsDriver.resolveRepositoryPaths.commonDir",
1021 cwd,
1022 args: ["rev-parse", "--git-common-dir"],
1023 }),
1024 detail: "Failed to resolve the Git common directory.",
1025 exitCode: commonDirResult.exitCode,
1026 stdoutLength: commonDirResult.stdout.length,
1027 stderrLength: commonDirResult.stderr.length,
1028 });
1029 }
1030
1031 const commonDirOutput = commonDirResult.stdout.trim();
1032 const resolvedGitCommonDir = path.isAbsolute(commonDirOutput)
1033 ? path.normalize(commonDirOutput)
1034 : path.resolve(cwd, commonDirOutput);
1035 const gitCommonDir = yield* fileSystem
1036 .realPath(resolvedGitCommonDir)
1037 .pipe(Effect.orElseSucceed(() => resolvedGitCommonDir));
1038 const [worktreeRootResult, currentBranchResult] = yield* Effect.all(
1039 [
1040 executeGit(
1041 "GitVcsDriver.resolveRepositoryPaths.worktreeRoot",
1042 cwd,
1043 ["rev-parse", "--show-toplevel"],
1044 {
1045 timeoutMs: 5_000,
1046 allowNonZeroExit: true,
1047 },
1048 ),
1049 executeGit(
1050 "GitVcsDriver.resolveRepositoryPaths.currentBranch",
1051 cwd,
1052 ["symbolic-ref", "--quiet", "--short", "HEAD"],
1053 {
1054 timeoutMs: 5_000,
1055 allowNonZeroExit: true,
1056 },
1057 ),
1058 ],
1059 { concurrency: 2 },
1060 );
1061 const worktreeRootOutput = worktreeRootResult.stdout.trim();
1062 const worktreeRoot =
1063 worktreeRootResult.exitCode === 0 && worktreeRootOutput.length > 0
1064 ? path.normalize(
1065 path.isAbsolute(worktreeRootOutput)
1066 ? worktreeRootOutput
1067 : path.resolve(cwd, worktreeRootOutput),
1068 )
1069 : null;
1070 const currentBranchOutput = currentBranchResult.stdout.trim();
1071 const currentBranch =
1072 currentBranchResult.exitCode === 0 && currentBranchOutput.length > 0
1073 ? currentBranchOutput
1074 : null;
1075
1076 return {
1077 gitCommonDir,
1078 worktreeRoot,
1079 currentBranch,
1080 } satisfies GitRepositoryPaths;
1081 });
1082
1083 const repositoryPathsCache = yield* Cache.makeWith(
1084 (cwd: string) => resolveRepositoryPathsUncached(cwd),
1085 {
1086 capacity: REPOSITORY_PATHS_CACHE_CAPACITY,
1087 timeToLive: Exit.match({
1088 onSuccess: (repositoryPaths) =>
1089 repositoryPaths === null ? NON_REPOSITORY_PATHS_CACHE_TTL : REPOSITORY_PATHS_CACHE_TTL,
1090 onFailure: () => Duration.zero,
1091 }),
1092 },
1093 );
1094 const repositoryPathsRefreshCache = yield* Cache.makeWith(
1095 (cwd: string) =>
1096 Cache.invalidate(repositoryPathsCache, cwd).pipe(
1097 Effect.andThen(Cache.get(repositoryPathsCache, cwd)),
1098 ),
1099 {
1100 capacity: REPOSITORY_PATHS_CACHE_CAPACITY,
1101 timeToLive: Exit.match({
1102 onSuccess: (repositoryPaths) =>
1103 repositoryPaths === null
1104 ? NON_REPOSITORY_PATHS_CACHE_TTL
1105 : REPOSITORY_PATHS_REFRESH_COALESCE_TTL,
1106 onFailure: () => Duration.zero,
1107 }),
1108 },
1109 );
1110 const normalizeRepositoryPathsCacheKey = (cwd: string) => path.normalize(path.resolve(cwd));
1111 const resolveRepositoryPaths = (cwd: string, refresh = false) => {
1112 const cacheKey = normalizeRepositoryPathsCacheKey(cwd);
1113 return Cache.get(refresh ? repositoryPathsRefreshCache : repositoryPathsCache, cacheKey);
1114 };
1115
9471116 const resolveGitCommonDir = Effect.fn("resolveGitCommonDir")(function* (cwd: string) {
948 const gitCommonDir = yield* runGitStdout("GitVcsDriver.resolveGitCommonDir", cwd, [
949 "rev-parse",
950 "--git-common-dir",
951 ]).pipe(Effect.map((stdout) => stdout.trim()));
952 return path.isAbsolute(gitCommonDir) ? gitCommonDir : path.resolve(cwd, gitCommonDir);
1117 const repositoryPaths = yield* resolveRepositoryPaths(cwd);
1118 if (repositoryPaths !== null) {
1119 return repositoryPaths.gitCommonDir;
1120 }
1121 return yield* new GitCommandError({
1122 ...gitCommandContext({
1123 operation: "GitVcsDriver.resolveGitCommonDir",
1124 cwd,
1125 args: ["rev-parse", "--git-common-dir"],
1126 }),
1127 detail: "Cannot resolve a Git common directory outside a repository.",
1128 });
9531129 });
9541130
1131 const statusRemoteRefreshFailureCounts = new Map<string, number>();
1132 const statusRemoteRefreshFailureKey = (cacheKey: StatusRemoteRefreshCacheKey) =>
1133 `${cacheKey.gitCommonDir}\0${cacheKey.remoteName}`;
1134 const recordStatusRemoteRefreshFailure = (cacheKey: StatusRemoteRefreshCacheKey) => {
1135 const key = statusRemoteRefreshFailureKey(cacheKey);
1136 const nextCount = (statusRemoteRefreshFailureCounts.get(key) ?? 0) + 1;
1137 statusRemoteRefreshFailureCounts.delete(key);
1138 statusRemoteRefreshFailureCounts.set(key, nextCount);
1139 if (statusRemoteRefreshFailureCounts.size > STATUS_UPSTREAM_REFRESH_CACHE_CAPACITY) {
1140 const oldestKey = statusRemoteRefreshFailureCounts.keys().next().value;
1141 if (oldestKey !== undefined) {
1142 statusRemoteRefreshFailureCounts.delete(oldestKey);
1143 }
1144 }
1145 };
1146 const clearStatusRemoteRefreshFailures = (cacheKey: StatusRemoteRefreshCacheKey) => {
1147 statusRemoteRefreshFailureCounts.delete(statusRemoteRefreshFailureKey(cacheKey));
1148 };
9551149 const refreshStatusRemoteCacheEntry = Effect.fn("refreshStatusRemoteCacheEntry")(function* (
9561150 cacheKey: StatusRemoteRefreshCacheKey,
9571151 ) {
958 yield* fetchRemoteForStatus(cacheKey.gitCommonDir, cacheKey.remoteName);
959 return true as const;
1152 return yield* fetchRemoteForStatus(cacheKey.gitCommonDir, cacheKey.remoteName).pipe(
1153 Effect.tap(() => Effect.sync(() => clearStatusRemoteRefreshFailures(cacheKey))),
1154 Effect.tapError(() => Effect.sync(() => recordStatusRemoteRefreshFailure(cacheKey))),
1155 Effect.as(true as const),
1156 );
9601157 });
9611158
9621159 const statusRemoteRefreshCache = yield* Cache.makeWith(refreshStatusRemoteCacheEntry, {
9631160 capacity: STATUS_UPSTREAM_REFRESH_CACHE_CAPACITY,
964 // Keep successful refreshes warm and briefly back off failed refreshes to avoid retry storms.
965 timeToLive: (exit) =>
1161 // A failed background fetch is intentionally cached and exponentially
1162 // backed off. Status reads swallow this failure and use the last fetched
1163 // refs, so repeated thread mounts cannot turn a slow or unavailable remote
1164 // into a repository-wide Git subprocess storm.
1165 timeToLive: (exit, cacheKey) =>
9661166 Exit.isSuccess(exit)
9671167 ? STATUS_UPSTREAM_REFRESH_INTERVAL
968 : STATUS_UPSTREAM_REFRESH_FAILURE_COOLDOWN,
1168 : statusUpstreamRefreshFailureCooldown(
1169 statusRemoteRefreshFailureCounts.get(statusRemoteRefreshFailureKey(cacheKey)) ?? 1,
1170 ),
9691171 });
9701172
9711173 const refreshStatusUpstreamIfStale = Effect.fn("refreshStatusUpstreamIfStale")(function* (
......@@ -1273,42 +1475,6 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function*
12731475 };
12741476 });
12751477
1276 const readBranchRecency = Effect.fn("readBranchRecency")(function* (cwd: string) {
1277 const branchRecency = yield* executeGit(
1278 "GitVcsDriver.readBranchRecency",
1279 cwd,
1280 [
1281 "for-each-ref",
1282 "--format=%(refname:short)%09%(committerdate:unix)",
1283 "refs/heads",
1284 "refs/remotes",
1285 ],
1286 {
1287 timeoutMs: 15_000,
1288 allowNonZeroExit: true,
1289 },
1290 );
1291
1292 const branchLastCommit = new Map<string, number>();
1293 if (branchRecency.exitCode !== 0) {
1294 return branchLastCommit;
1295 }
1296
1297 for (const line of branchRecency.stdout.split("\n")) {
1298 if (line.length === 0) {
1299 continue;
1300 }
1301 const [name, lastCommitRaw] = line.split("\t");
1302 if (!name) {
1303 continue;
1304 }
1305 const lastCommit = Number.parseInt(lastCommitRaw ?? "0", 10);
1306 branchLastCommit.set(name, Number.isFinite(lastCommit) ? lastCommit : 0);
1307 }
1308
1309 return branchLastCommit;
1310 });
1311
13121478 const readStatusDetailsLocal = Effect.fn("readStatusDetailsLocal")(function* (cwd: string) {
13131479 const statusResult = yield* executeGitWithStableDiagnostics(
13141480 "GitVcsDriver.statusDetails.status",
......@@ -1987,257 +2153,283 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function*
19872153 Effect.map((trimmed) => (trimmed.length > 0 ? trimmed : null)),
19882154 );
19892155
1990 const listRefs: GitVcsDriver.GitVcsDriver["Service"]["listRefs"] = Effect.fn("listRefs")(
1991 function* (input) {
1992 const branchRecencyPromise = readBranchRecency(input.cwd).pipe(
1993 Effect.orElseSucceed(() => new Map<string, number>()),
1994 );
1995 const localBranchResult = yield* executeGitWithStableDiagnostics(
1996 "GitVcsDriver.listRefs.branchNoColor",
1997 input.cwd,
1998 ["branch", "--no-color", "--no-column"],
1999 {
2000 timeoutMs: 10_000,
2156 const readGitRefsSnapshot = Effect.fn("readGitRefsSnapshot")(function* (gitCommonDir: string) {
2157 const fetchCwd =
2158 path.basename(gitCommonDir) === ".git" ? path.dirname(gitCommonDir) : gitCommonDir;
2159 const gitDirArgs = ["--git-dir", gitCommonDir] as const;
2160 const [refsResult, defaultRefResult, worktreeListResult, remoteNamesResult] = yield* Effect.all(
2161 [
2162 executeGitWithStableDiagnostics(
2163 "GitVcsDriver.listRefs.snapshotRefs",
2164 fetchCwd,
2165 [
2166 ...gitDirArgs,
2167 "for-each-ref",
2168 "--format=%(refname)%09%(committerdate:unix)%09%(symref)",
2169 "refs/heads",
2170 "refs/remotes",
2171 ],
2172 {
2173 timeoutMs: 30_000,
2174 maxOutputBytes: 16 * 1024 * 1024,
2175 fallbackErrorDetail: "Git ref snapshot enumeration failed.",
2176 },
2177 ),
2178 executeGit(
2179 "GitVcsDriver.listRefs.defaultRef",
2180 fetchCwd,
2181 [...gitDirArgs, "symbolic-ref", "refs/remotes/origin/HEAD"],
2182 {
2183 timeoutMs: 5_000,
2184 allowNonZeroExit: true,
2185 },
2186 ),
2187 executeGit(
2188 "GitVcsDriver.listRefs.worktreeList",
2189 fetchCwd,
2190 [...gitDirArgs, "worktree", "list", "--porcelain", "-z"],
2191 {
2192 timeoutMs: 30_000,
2193 allowNonZeroExit: true,
2194 maxOutputBytes: 16 * 1024 * 1024,
2195 },
2196 ),
2197 executeGit("GitVcsDriver.listRefs.remoteNames", fetchCwd, [...gitDirArgs, "remote"], {
2198 timeoutMs: 5_000,
20012199 allowNonZeroExit: true,
2002 },
2003 ).pipe(
2004 Effect.catchTags({
2005 GitCommandError: (error) =>
2006 isMissingGitCwdError(error)
2007 ? Effect.succeed({
2008 exitCode: ChildProcessSpawner.ExitCode(128),
2009 stdout: "",
2010 stderr: "fatal: not a git repository",
2011 stdoutTruncated: false,
2012 stderrTruncated: false,
2013 })
2014 : Effect.fail(error),
20152200 }),
2016 );
2201 ],
2202 { concurrency: 2 },
2203 );
20172204
2018 if (localBranchResult.exitCode !== 0) {
2019 const stderr = localBranchResult.stderr.trim();
2020 if (isNonRepositoryGitStderr(stderr)) {
2021 return {
2022 refs: [],
2023 isRepo: false,
2024 hasPrimaryRemote: false,
2025 nextCursor: null,
2026 totalCount: 0,
2027 };
2028 }
2029 return yield* new GitCommandError({
2030 ...gitCommandContext({
2031 operation: "GitVcsDriver.listRefs",
2032 cwd: input.cwd,
2033 args: ["branch", "--no-color", "--no-column"],
2034 }),
2035 detail: "Git branch listing failed.",
2036 exitCode: localBranchResult.exitCode,
2037 stdoutLength: localBranchResult.stdout.length,
2038 stderrLength: localBranchResult.stderr.length,
2205 const remoteNames =
2206 remoteNamesResult.exitCode === 0 ? parseRemoteNames(remoteNamesResult.stdout) : [];
2207 if (remoteNamesResult.exitCode !== 0 && remoteNamesResult.stderr.trim().length > 0) {
2208 yield* Effect.logWarning(
2209 `GitVcsDriver.listRefs: remote name lookup returned code ${remoteNamesResult.exitCode} for ${gitCommonDir}: ${remoteNamesResult.stderr.trim()}. Falling back to an empty remote name list.`,
2210 );
2211 }
2212 const defaultBranch =
2213 defaultRefResult.exitCode === 0
2214 ? defaultRefResult.stdout.trim().replace(/^refs\/remotes\/origin\//, "")
2215 : null;
2216 const parsedWorktreeEntries =
2217 worktreeListResult.exitCode === 0
2218 ? [...parseWorktreeBranchPaths(worktreeListResult.stdout)].map(
2219 ([branchName, worktreePath]) =>
2220 [branchName, path.normalize(path.resolve(worktreePath))] as const,
2221 )
2222 : [];
2223 const existingWorktreeEntries = yield* Effect.filter(
2224 parsedWorktreeEntries,
2225 ([, worktreePath]) =>
2226 fileSystem.stat(worktreePath).pipe(
2227 Effect.as(true),
2228 Effect.orElseSucceed(() => false),
2229 ),
2230 { concurrency: 16 },
2231 );
2232 const worktreeMap = new Map(existingWorktreeEntries);
2233 const localBranches: Array<{ readonly ref: VcsRef; readonly lastCommit: number }> = [];
2234 const remoteBranches: Array<{ readonly ref: VcsRef; readonly lastCommit: number }> = [];
2235
2236 for (const line of refsResult.stdout.split("\n")) {
2237 if (line.length === 0) continue;
2238 const [fullRefName, lastCommitRaw, symbolicTarget] = line.split("\t");
2239 if (!fullRefName || symbolicTarget) continue;
2240 const parsedLastCommit = Number.parseInt(lastCommitRaw ?? "0", 10);
2241 const lastCommit = Number.isFinite(parsedLastCommit) ? parsedLastCommit : 0;
2242
2243 if (fullRefName.startsWith("refs/heads/")) {
2244 const name = fullRefName.slice("refs/heads/".length);
2245 localBranches.push({
2246 ref: {
2247 name,
2248 current: false,
2249 isRemote: false,
2250 isDefault: name === defaultBranch,
2251 worktreePath: worktreeMap.get(name) ?? null,
2252 },
2253 lastCommit,
20392254 });
2255 continue;
20402256 }
2257 if (!fullRefName.startsWith("refs/remotes/")) continue;
2258
2259 const name = fullRefName.slice("refs/remotes/".length);
2260 const parsedRemoteRef = parseRemoteRefWithRemoteNames(name, remoteNames);
2261 const remoteBranch: VcsRef = {
2262 name,
2263 current: false,
2264 isRemote: true,
2265 isDefault:
2266 defaultBranch !== null &&
2267 parsedRemoteRef?.remoteName === "origin" &&
2268 parsedRemoteRef.branchName === defaultBranch,
2269 worktreePath: null,
2270 ...(parsedRemoteRef ? { remoteName: parsedRemoteRef.remoteName } : {}),
2271 };
2272 remoteBranches.push({ ref: remoteBranch, lastCommit });
2273 }
20412274
2042 const remoteBranchResultEffect = executeGit(
2043 "GitVcsDriver.listRefs.remoteBranches",
2044 input.cwd,
2045 ["branch", "--no-color", "--no-column", "--remotes"],
2046 {
2047 timeoutMs: 10_000,
2048 allowNonZeroExit: true,
2049 },
2050 ).pipe(
2051 Effect.catchTags({
2052 GitCommandError: (error) =>
2053 Effect.logWarning(
2054 "Git remote ref lookup failed; falling back to an empty remote ref list.",
2055 {
2056 operation: error.operation,
2057 command: error.command,
2058 cwd: error.cwd,
2059 detail: error.detail,
2060 cause: error,
2061 },
2062 ).pipe(
2063 Effect.as({
2064 exitCode: ChildProcessSpawner.ExitCode(1),
2065 stdout: "",
2066 stderr: "",
2067 stdoutTruncated: false,
2068 stderrTruncated: false,
2069 } satisfies GitVcsDriver.ExecuteGitResult),
2070 ),
2071 }),
2072 );
2073
2074 const remoteNamesResultEffect = executeGit(
2075 "GitVcsDriver.listRefs.remoteNames",
2076 input.cwd,
2077 ["remote"],
2078 {
2079 timeoutMs: 5_000,
2080 allowNonZeroExit: true,
2081 },
2082 ).pipe(
2083 Effect.catchTags({
2084 GitCommandError: (error) =>
2085 Effect.logWarning(
2086 "Git remote name lookup failed; falling back to an empty remote name list.",
2087 {
2088 operation: error.operation,
2089 command: error.command,
2090 cwd: error.cwd,
2091 detail: error.detail,
2092 cause: error,
2093 },
2094 ).pipe(
2095 Effect.as({
2096 exitCode: ChildProcessSpawner.ExitCode(1),
2097 stdout: "",
2098 stderr: "",
2099 stdoutTruncated: false,
2100 stderrTruncated: false,
2101 } satisfies GitVcsDriver.ExecuteGitResult),
2102 ),
2103 }),
2104 );
2275 const byRecencyThenName = (
2276 left: { readonly ref: VcsRef; readonly lastCommit: number },
2277 right: { readonly ref: VcsRef; readonly lastCommit: number },
2278 ) =>
2279 left.lastCommit !== right.lastCommit
2280 ? right.lastCommit - left.lastCommit
2281 : left.ref.name.localeCompare(right.ref.name);
21052282
2106 const [defaultRef, worktreeList, remoteBranchResult, remoteNamesResult, branchLastCommit] =
2107 yield* Effect.all(
2108 [
2109 executeGit(
2110 "GitVcsDriver.listRefs.defaultRef",
2111 input.cwd,
2112 ["symbolic-ref", "refs/remotes/origin/HEAD"],
2113 {
2114 timeoutMs: 5_000,
2115 allowNonZeroExit: true,
2116 },
2117 ),
2118 executeGit(
2119 "GitVcsDriver.listRefs.worktreeList",
2120 input.cwd,
2121 ["worktree", "list", "--porcelain"],
2122 {
2123 timeoutMs: 5_000,
2124 allowNonZeroExit: true,
2125 },
2126 ),
2127 remoteBranchResultEffect,
2128 remoteNamesResultEffect,
2129 branchRecencyPromise,
2130 ],
2131 { concurrency: "unbounded" },
2132 );
2283 return {
2284 localBranches: localBranches.toSorted(byRecencyThenName).map(({ ref }) => ref),
2285 remoteBranches: remoteBranches.toSorted(byRecencyThenName).map(({ ref }) => ref),
2286 hasPrimaryRemote: remoteNames.includes("origin"),
2287 } satisfies GitRefsSnapshot;
2288 });
21332289
2134 const remoteNames =
2135 remoteNamesResult.exitCode === 0 ? parseRemoteNames(remoteNamesResult.stdout) : [];
2136 if (remoteBranchResult.exitCode !== 0 && remoteBranchResult.stderr.trim().length > 0) {
2137 yield* Effect.logWarning(
2138 `GitVcsDriver.listRefs: remote refName lookup returned code ${remoteBranchResult.exitCode} for ${input.cwd}: ${remoteBranchResult.stderr.trim()}. Falling back to an empty remote refName list.`,
2139 );
2290 const listRefsEpochByCommonDir = new Map<string, number>();
2291 let listRefsEpochSequence = 0;
2292 const bumpListRefsEpoch = (gitCommonDir: string): number => {
2293 const nextEpoch = ++listRefsEpochSequence;
2294 listRefsEpochByCommonDir.delete(gitCommonDir);
2295 listRefsEpochByCommonDir.set(gitCommonDir, nextEpoch);
2296 if (listRefsEpochByCommonDir.size > LIST_REFS_SNAPSHOT_CACHE_CAPACITY) {
2297 const oldestKey = listRefsEpochByCommonDir.keys().next().value;
2298 if (oldestKey !== undefined) {
2299 listRefsEpochByCommonDir.delete(oldestKey);
21402300 }
2141 if (remoteNamesResult.exitCode !== 0 && remoteNamesResult.stderr.trim().length > 0) {
2142 yield* Effect.logWarning(
2143 `GitVcsDriver.listRefs: remote name lookup returned code ${remoteNamesResult.exitCode} for ${input.cwd}: ${remoteNamesResult.stderr.trim()}. Falling back to an empty remote name list.`,
2144 );
2301 }
2302 return nextEpoch;
2303 };
2304 const listRefsGenerationByCommonDir = new Map<string, number>();
2305 let listRefsGenerationSequence = 0;
2306 const setListRefsGeneration = (gitCommonDir: string, generation: number): number => {
2307 listRefsGenerationByCommonDir.delete(gitCommonDir);
2308 listRefsGenerationByCommonDir.set(gitCommonDir, generation);
2309 if (listRefsGenerationByCommonDir.size > LIST_REFS_SNAPSHOT_CACHE_CAPACITY) {
2310 const oldestKey = listRefsGenerationByCommonDir.keys().next().value;
2311 if (oldestKey !== undefined) {
2312 listRefsGenerationByCommonDir.delete(oldestKey);
21452313 }
2146
2147 const defaultBranch =
2148 defaultRef.exitCode === 0
2149 ? defaultRef.stdout.trim().replace(/^refs\/remotes\/origin\//, "")
2150 : null;
2151
2152 const worktreeMap = new Map<string, string>();
2153 if (worktreeList.exitCode === 0) {
2154 let currentPath: string | null = null;
2155 for (const line of worktreeList.stdout.split("\n")) {
2156 if (line.startsWith("worktree ")) {
2157 const candidatePath = line.slice("worktree ".length);
2158 const exists = yield* fileSystem.stat(candidatePath).pipe(
2159 Effect.map(() => true),
2160 Effect.orElseSucceed(() => false),
2314 }
2315 return generation;
2316 };
2317 const currentListRefsGeneration = (gitCommonDir: string): number => {
2318 const current = listRefsGenerationByCommonDir.get(gitCommonDir);
2319 return current === undefined
2320 ? setListRefsGeneration(gitCommonDir, ++listRefsGenerationSequence)
2321 : setListRefsGeneration(gitCommonDir, current);
2322 };
2323 const bumpListRefsGeneration = (gitCommonDir: string): number =>
2324 setListRefsGeneration(gitCommonDir, ++listRefsGenerationSequence);
2325 const listRefsSnapshotCache = yield* Cache.makeWith(
2326 (cacheKey: GitRefsSnapshotCacheKey) => readGitRefsSnapshot(cacheKey.gitCommonDir),
2327 {
2328 capacity: LIST_REFS_SNAPSHOT_CACHE_CAPACITY,
2329 timeToLive: (exit) => (Exit.isSuccess(exit) ? LIST_REFS_SNAPSHOT_CACHE_TTL : Duration.zero),
2330 },
2331 );
2332 const listRefsRefreshSnapshotCache = yield* Cache.makeWith(
2333 (cacheKey: GitRefsRefreshCacheKey) =>
2334 Effect.suspend(() => {
2335 const epoch = bumpListRefsEpoch(cacheKey.gitCommonDir);
2336 return Cache.get(
2337 listRefsSnapshotCache,
2338 new GitRefsSnapshotCacheKey({ gitCommonDir: cacheKey.gitCommonDir, epoch }),
2339 );
2340 }),
2341 {
2342 capacity: LIST_REFS_SNAPSHOT_CACHE_CAPACITY,
2343 timeToLive: (exit) =>
2344 Exit.isSuccess(exit) ? LIST_REFS_REFRESH_COALESCE_TTL : LIST_REFS_REFRESH_FAILURE_COOLDOWN,
2345 },
2346 );
2347 const resolveListRefsSnapshot = Effect.fn("resolveListRefsSnapshot")(function* (
2348 gitCommonDir: string,
2349 refresh: boolean,
2350 ) {
2351 while (true) {
2352 const generation = currentListRefsGeneration(gitCommonDir);
2353 const currentEpoch = listRefsEpochByCommonDir.get(gitCommonDir);
2354 const snapshot =
2355 refresh || currentEpoch === undefined
2356 ? // The refresh cache owns the complete snapshot read, rather than only the
2357 // epoch bump. Slow repositories therefore remain singleflight for the
2358 // entire Git scan even when more refresh requests arrive after the
2359 // coalescing TTL would otherwise have elapsed.
2360 yield* Cache.get(
2361 listRefsRefreshSnapshotCache,
2362 new GitRefsRefreshCacheKey({ gitCommonDir, generation }),
2363 )
2364 : yield* Cache.get(
2365 listRefsSnapshotCache,
2366 new GitRefsSnapshotCacheKey({ gitCommonDir, epoch: currentEpoch }),
21612367 );
2162 currentPath = exists ? candidatePath : null;
2163 } else if (line.startsWith("branch refs/heads/") && currentPath) {
2164 worktreeMap.set(line.slice("branch refs/heads/".length), currentPath);
2165 } else if (line === "") {
2166 currentPath = null;
2167 }
2168 }
2368 if (currentListRefsGeneration(gitCommonDir) === generation) {
2369 return snapshot;
21692370 }
2371 }
2372 });
2373 const invalidateListRefsSnapshot = Effect.fn("invalidateListRefsSnapshot")(function* (
2374 cwd: string,
2375 ) {
2376 const repositoryPathsCacheKey = normalizeRepositoryPathsCacheKey(cwd);
2377 const repositoryPaths = yield* Cache.get(repositoryPathsCache, repositoryPathsCacheKey);
2378 if (repositoryPaths === null) return;
2379 const previousGeneration = currentListRefsGeneration(repositoryPaths.gitCommonDir);
2380 bumpListRefsGeneration(repositoryPaths.gitCommonDir);
2381 bumpListRefsEpoch(repositoryPaths.gitCommonDir);
2382 yield* Cache.invalidate(
2383 listRefsRefreshSnapshotCache,
2384 new GitRefsRefreshCacheKey({
2385 gitCommonDir: repositoryPaths.gitCommonDir,
2386 generation: previousGeneration,
2387 }),
2388 );
2389 yield* Cache.invalidate(repositoryPathsRefreshCache, repositoryPathsCacheKey);
2390 yield* Cache.invalidate(repositoryPathsCache, repositoryPathsCacheKey);
2391 });
21702392
2171 const localBranches = Arr.filterMap(localBranchResult.stdout.split("\n"), (line) => {
2172 const refName = parseBranchLine(line);
2173 return refName === null
2174 ? Result.failVoid
2175 : Result.succeed({
2176 name: refName.name,
2177 current: refName.current,
2178 isRemote: false,
2179 isDefault: refName.name === defaultBranch,
2180 worktreePath: worktreeMap.get(refName.name) ?? null,
2181 });
2182 }).toSorted((a, b) => {
2183 const aPriority = a.current ? 0 : a.isDefault ? 1 : 2;
2184 const bPriority = b.current ? 0 : b.isDefault ? 1 : 2;
2185 if (aPriority !== bPriority) return aPriority - bPriority;
2186
2187 const aLastCommit = branchLastCommit.get(a.name) ?? 0;
2188 const bLastCommit = branchLastCommit.get(b.name) ?? 0;
2189 if (aLastCommit !== bLastCommit) return bLastCommit - aLastCommit;
2190 return a.name.localeCompare(b.name);
2191 });
2192
2193 const remoteBranches =
2194 remoteBranchResult.exitCode === 0
2195 ? Arr.filterMap(remoteBranchResult.stdout.split("\n"), (line) => {
2196 const refName = parseBranchLine(line);
2197 if (refName === null) {
2198 return Result.failVoid;
2199 }
2200 const parsedRemoteRef = parseRemoteRefWithRemoteNames(refName.name, remoteNames);
2201 const remoteBranch: {
2202 name: string;
2203 current: boolean;
2204 isRemote: boolean;
2205 remoteName?: string;
2206 isDefault: boolean;
2207 worktreePath: string | null;
2208 } = {
2209 name: refName.name,
2210 current: false,
2211 isRemote: true,
2212 // origin/HEAD's target is the repo default even when no local
2213 // copy of the default branch exists.
2214 isDefault:
2215 defaultBranch !== null &&
2216 parsedRemoteRef?.remoteName === "origin" &&
2217 parsedRemoteRef.branchName === defaultBranch,
2218 worktreePath: null,
2219 };
2220 if (parsedRemoteRef) {
2221 remoteBranch.remoteName = parsedRemoteRef.remoteName;
2222 }
2223 return Result.succeed(remoteBranch);
2224 }).toSorted((a, b) => {
2225 const aLastCommit = branchLastCommit.get(a.name) ?? 0;
2226 const bLastCommit = branchLastCommit.get(b.name) ?? 0;
2227 if (aLastCommit !== bLastCommit) return bLastCommit - aLastCommit;
2228 return a.name.localeCompare(b.name);
2229 })
2230 : [];
2393 const listRefs: GitVcsDriver.GitVcsDriver["Service"]["listRefs"] = Effect.fn("listRefs")(
2394 function* (input) {
2395 const repositoryPaths = yield* resolveRepositoryPaths(input.cwd, input.refresh === true).pipe(
2396 Effect.catchTags({
2397 GitCommandError: (error) =>
2398 isMissingGitCwdError(error) ? Effect.succeed(null) : Effect.fail(error),
2399 }),
2400 );
2401 if (repositoryPaths === null) {
2402 return {
2403 refs: [],
2404 isRepo: false,
2405 hasPrimaryRemote: false,
2406 nextCursor: null,
2407 totalCount: 0,
2408 };
2409 }
22312410
2411 const snapshot = yield* resolveListRefsSnapshot(
2412 repositoryPaths.gitCommonDir,
2413 input.refresh === true,
2414 );
2415 const hasCurrentWorktreeBranch =
2416 repositoryPaths.worktreeRoot !== null &&
2417 snapshot.localBranches.some((ref) => ref.worktreePath === repositoryPaths.worktreeRoot);
2418 const localBranches = snapshot.localBranches.map((ref) => ({
2419 ...ref,
2420 current: hasCurrentWorktreeBranch
2421 ? ref.worktreePath === repositoryPaths.worktreeRoot
2422 : ref.name === repositoryPaths.currentBranch,
2423 }));
22322424 const combinedBranches = input.includeMatchingRemoteRefs
2233 ? [...localBranches, ...remoteBranches]
2234 : dedupeRemoteBranchesWithLocalMatches([...localBranches, ...remoteBranches]);
2425 ? [...localBranches, ...snapshot.remoteBranches]
2426 : dedupeRemoteBranchesWithLocalMatches([...localBranches, ...snapshot.remoteBranches]);
22352427 // Keep current/default refs on the first page even when the default
22362428 // only exists as origin/<default> (remote refs sort after all locals).
2237 const allBranches = combinedBranches.toSorted((a, b) => {
2238 const aPriority = a.current ? 0 : a.isDefault ? 1 : 2;
2239 const bPriority = b.current ? 0 : b.isDefault ? 1 : 2;
2240 return aPriority - bPriority;
2429 const allBranches = combinedBranches.toSorted((left, right) => {
2430 const leftPriority = left.current ? 0 : left.isDefault ? 1 : 2;
2431 const rightPriority = right.current ? 0 : right.isDefault ? 1 : 2;
2432 return leftPriority - rightPriority;
22412433 });
22422434 const branchesForKind =
22432435 input.refKind === "local"
......@@ -2254,7 +2446,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function*
22542446 return {
22552447 refs: [...refs.refs],
22562448 isRepo: true,
2257 hasPrimaryRemote: remoteNames.includes("origin"),
2449 hasPrimaryRemote: snapshot.hasPrimaryRemote,
22582450 nextCursor: refs.nextCursor,
22592451 totalCount: refs.totalCount,
22602452 };
......@@ -2548,6 +2740,25 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function*
25482740 }),
25492741 );
25502742
2743 const withListRefsInvalidation = <A, E>(
2744 cwd: string,
2745 effect: Effect.Effect<A, E>,
2746 ): Effect.Effect<A, E> =>
2747 effect.pipe(Effect.ensuring(invalidateListRefsSnapshot(cwd).pipe(Effect.ignore)));
2748 const initRepoWithListRefsInvalidation: GitVcsDriver.GitVcsDriver["Service"]["initRepo"] = (
2749 input,
2750 ) =>
2751 initRepo(input).pipe(
2752 Effect.ensuring(
2753 Effect.gen(function* () {
2754 const cacheKey = normalizeRepositoryPathsCacheKey(input.cwd);
2755 yield* Cache.invalidate(repositoryPathsRefreshCache, cacheKey);
2756 yield* Cache.invalidate(repositoryPathsCache, cacheKey);
2757 yield* invalidateListRefsSnapshot(input.cwd).pipe(Effect.ignore);
2758 }),
2759 ),
2760 );
2761
25512762 return GitVcsDriver.GitVcsDriver.of({
25522763 execute,
25532764 status,
......@@ -2555,27 +2766,31 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function*
25552766 statusDetailsLocal,
25562767 statusDetailsRemote,
25572768 prepareCommitContext,
2558 commit,
2559 pushCurrentBranch,
2560 pullCurrentBranch,
2769 commit: (cwd, subject, body, options) =>
2770 withListRefsInvalidation(cwd, commit(cwd, subject, body, options)),
2771 pushCurrentBranch: (cwd, fallbackBranch, options) =>
2772 withListRefsInvalidation(cwd, pushCurrentBranch(cwd, fallbackBranch, options)),
2773 pullCurrentBranch: (cwd) => withListRefsInvalidation(cwd, pullCurrentBranch(cwd)),
25612774 readRangeContext,
25622775 getReviewDiffPreview,
25632776 readConfigValue,
25642777 listRefs,
2565 createWorktree,
2566 fetchPullRequestBranch,
2567 ensureRemote,
2778 createWorktree: (input) => withListRefsInvalidation(input.cwd, createWorktree(input)),
2779 fetchPullRequestBranch: (input) =>
2780 withListRefsInvalidation(input.cwd, fetchPullRequestBranch(input)),
2781 ensureRemote: (input) => withListRefsInvalidation(input.cwd, ensureRemote(input)),
25682782 resolvePrimaryRemoteName,
2569 fetchRemote,
2783 fetchRemote: (input) => withListRefsInvalidation(input.cwd, fetchRemote(input)),
25702784 resolveRemoteTrackingCommit,
2571 fetchRemoteBranch,
2572 fetchRemoteTrackingBranch,
2573 setBranchUpstream,
2574 removeWorktree,
2575 renameBranch,
2576 createRef,
2577 switchRef,
2578 initRepo,
2785 fetchRemoteBranch: (input) => withListRefsInvalidation(input.cwd, fetchRemoteBranch(input)),
2786 fetchRemoteTrackingBranch: (input) =>
2787 withListRefsInvalidation(input.cwd, fetchRemoteTrackingBranch(input)),
2788 setBranchUpstream: (input) => withListRefsInvalidation(input.cwd, setBranchUpstream(input)),
2789 removeWorktree: (input) => withListRefsInvalidation(input.cwd, removeWorktree(input)),
2790 renameBranch: (input) => withListRefsInvalidation(input.cwd, renameBranch(input)),
2791 createRef: (input) => withListRefsInvalidation(input.cwd, createRef(input)),
2792 switchRef: (input) => withListRefsInvalidation(input.cwd, switchRef(input)),
2793 initRepo: initRepoWithListRefsInvalidation,
25792794 listLocalBranchNames,
25802795 });
25812796});
apps/web/src/components/BranchToolbarBranchSelector.tsx+26-30
......@@ -24,6 +24,7 @@ import { useComposerDraftStore, type DraftId } from "../composerDraftStore";
2424import { writeTextToClipboard } from "../hooks/useCopyToClipboard";
2525import { readLocalApi } from "../localApi";
2626import { useOpenPrLink } from "../lib/openPullRequestLink";
27import { shouldLoadNextBranchPageAfterScroll } from "../state/paginatedBranches";
2728import { usePaginatedBranches } from "../state/queries";
2829import { useProject, useThread } from "../state/entities";
2930import { useEnvironmentQuery } from "../state/query";
......@@ -230,7 +231,7 @@ export function BranchToolbarBranchSelector({
230231 const refs = branchRefState.refs;
231232 const hasNextPage =
232233 branchRefState.data?.nextCursor !== null && branchRefState.data?.nextCursor !== undefined;
233 const isFetchingNextPage = branchRefState.isPending && branchRefState.data !== null;
234 const isFetchingNextPage = branchRefState.isFetchingNextPage;
234235 const isInitialBranchesLoadPending = branchRefState.isPending && branchRefState.data === null;
235236 const currentGitBranch =
236237 branchStatusQuery.data?.refName ?? refs.find((refName) => refName.current)?.name ?? null;
......@@ -505,19 +506,16 @@ export function BranchToolbarBranchSelector({
505506 // ---------------------------------------------------------------------------
506507 // Combobox / list plumbing
507508 // ---------------------------------------------------------------------------
508 const handleOpenChange = useCallback(
509 (open: boolean) => {
510 setIsBranchMenuOpen(open);
511 if (!open) {
512 setBranchQuery("");
513 return;
514 }
515 branchRefState.refresh();
516 },
517 [branchRefState.refresh],
518 );
519
520509 const branchListScrollElementRef = useRef<HTMLElement | null>(null);
510 const previousBranchListScrollTopRef = useRef<number | null>(null);
511 const handleOpenChange = useCallback((open: boolean) => {
512 previousBranchListScrollTopRef.current = null;
513 setIsBranchMenuOpen(open);
514 if (!open) {
515 setBranchQuery("");
516 }
517 }, []);
518
521519 const [showTopBranchScrollFade, setShowTopBranchScrollFade] = useState(false);
522520 const [showBottomBranchScrollFade, setShowBottomBranchScrollFade] = useState(false);
523521 const fetchNextBranchPage = useCallback(() => {
......@@ -528,18 +526,24 @@ export function BranchToolbarBranchSelector({
528526 branchRefState.loadNext();
529527 }, [branchRefState.loadNext, hasNextPage, isFetchingNextPage]);
530528 const maybeFetchNextBranchPage = useCallback(() => {
531 if (!isBranchMenuOpen || !hasNextPage || isFetchingNextPage) {
532 return;
533 }
534
535529 const scrollElement = branchListScrollElementRef.current;
536530 if (!scrollElement) {
537531 return;
538532 }
539533
540 const distanceFromBottom =
541 scrollElement.scrollHeight - scrollElement.scrollTop - scrollElement.clientHeight;
542 if (distanceFromBottom > 96) {
534 const previousScrollTop = previousBranchListScrollTopRef.current;
535 previousBranchListScrollTopRef.current = scrollElement.scrollTop;
536 if (
537 !isBranchMenuOpen ||
538 !hasNextPage ||
539 isFetchingNextPage ||
540 !shouldLoadNextBranchPageAfterScroll({
541 previousScrollTop,
542 scrollTop: scrollElement.scrollTop,
543 scrollHeight: scrollElement.scrollHeight,
544 clientHeight: scrollElement.clientHeight,
545 })
546 ) {
543547 return;
544548 }
545549
......@@ -589,10 +593,6 @@ export function BranchToolbarBranchSelector({
589593 void branchListRef.current?.scrollToOffset?.({ offset: 0, animated: false });
590594 }, [deferredTrimmedBranchQuery, isBranchMenuOpen]);
591595
592 useEffect(() => {
593 maybeFetchNextBranchPage();
594 }, [refs.length, maybeFetchNextBranchPage]);
595
596596 const triggerLabel = resolveBranchTriggerLabel({
597597 activeWorktreePath,
598598 effectiveEnvMode,
......@@ -791,14 +791,10 @@ export function BranchToolbarBranchSelector({
791791 renderItem={({ item, index }) => renderPickerItem(item, index)}
792792 estimatedItemSize={28}
793793 drawDistance={336}
794 onEndReached={() => {
795 if (hasNextPage && !isFetchingNextPage) {
796 fetchNextBranchPage();
797 }
798 }}
799794 onLayout={() => {
800795 updateBranchListScrollFades();
801 maybeFetchNextBranchPage();
796 previousBranchListScrollTopRef.current =
797 branchListScrollElementRef.current?.scrollTop ?? null;
802798 }}
803799 onScroll={() => {
804800 updateBranchListScrollFades();
apps/web/src/connection/storage.ts+14
......@@ -105,6 +105,8 @@ function persistenceError(
105105 | "save-server-config"
106106 | "load-vcs-refs"
107107 | "save-vcs-refs"
108 | "remove-vcs-refs"
109 | "clear-vcs-refs"
108110 | "clear-environment",
109111 cause: unknown,
110112) {
......@@ -619,6 +621,18 @@ export const connectionStorageLayer = Layer.effectContext(
619621 : persistenceError("save-vcs-refs", cause),
620622 ),
621623 ),
624 removeVcsRefs: (environmentId, cwd) =>
625 removeDatabaseValue(
626 database,
627 VCS_REFS_STORE_NAME,
628 vcsRefsCacheKey(environmentId, cwd),
629 ).pipe(Effect.mapError((cause) => persistenceError("remove-vcs-refs", cause))),
630 clearVcsRefs: (environmentId) =>
631 removeDatabaseValuesInRange(
632 database,
633 VCS_REFS_STORE_NAME,
634 IDBKeyRange.bound(`${environmentId}:`, `${environmentId}:\uffff`),
635 ).pipe(Effect.mapError((cause) => persistenceError("clear-vcs-refs", cause))),
622636 removeThread: (environmentId, threadId) =>
623637 removeDatabaseValue(
624638 database,
apps/web/src/state/paginatedBranches.test.ts created+111
......@@ -0,0 +1,111 @@
1import type { VcsListRefsResult } from "@t3tools/contracts";
2import { AsyncResult } from "effect/unstable/reactivity";
3import { describe, expect, it } from "vite-plus/test";
4
5import {
6 isPaginatedBranchesNextPagePending,
7 shouldLoadNextBranchPageAfterScroll,
8} from "./paginatedBranches";
9
10const FIRST_PAGE: VcsListRefsResult = {
11 refs: [],
12 isRepo: true,
13 hasPrimaryRemote: true,
14 nextCursor: 100,
15 totalCount: 150,
16};
17
18const LAST_PAGE: VcsListRefsResult = {
19 ...FIRST_PAGE,
20 nextCursor: null,
21};
22
23describe("paginated branch loading state", () => {
24 it("does not label a first-page background refresh as loading more refs", () => {
25 expect(
26 isPaginatedBranchesNextPagePending([
27 AsyncResult.success(FIRST_PAGE, {
28 waiting: true,
29 }),
30 ]),
31 ).toBe(false);
32 });
33
34 it("only reports loading more while a new cursor has no value", () => {
35 expect(
36 isPaginatedBranchesNextPagePending([
37 AsyncResult.success(FIRST_PAGE),
38 AsyncResult.initial<VcsListRefsResult>(true),
39 ]),
40 ).toBe(true);
41
42 expect(
43 isPaginatedBranchesNextPagePending([
44 AsyncResult.success(FIRST_PAGE),
45 AsyncResult.success(LAST_PAGE),
46 ]),
47 ).toBe(false);
48 });
49});
50
51describe("paginated branch scroll trigger", () => {
52 it("does not load from mount, layout, or an unchanged scroll position", () => {
53 expect(
54 shouldLoadNextBranchPageAfterScroll({
55 previousScrollTop: null,
56 scrollTop: 0,
57 scrollHeight: 224,
58 clientHeight: 224,
59 }),
60 ).toBe(false);
61
62 expect(
63 shouldLoadNextBranchPageAfterScroll({
64 previousScrollTop: 120,
65 scrollTop: 120,
66 scrollHeight: 344,
67 clientHeight: 224,
68 }),
69 ).toBe(false);
70 });
71
72 it("does not load when scrolling upward near the bottom", () => {
73 expect(
74 shouldLoadNextBranchPageAfterScroll({
75 previousScrollTop: 160,
76 scrollTop: 150,
77 scrollHeight: 450,
78 clientHeight: 224,
79 }),
80 ).toBe(false);
81 });
82
83 it("loads only after a downward scroll reaches the end threshold", () => {
84 expect(
85 shouldLoadNextBranchPageAfterScroll({
86 previousScrollTop: 0,
87 scrollTop: 100,
88 scrollHeight: 800,
89 clientHeight: 224,
90 }),
91 ).toBe(false);
92
93 expect(
94 shouldLoadNextBranchPageAfterScroll({
95 previousScrollTop: 100,
96 scrollTop: 500,
97 scrollHeight: 800,
98 clientHeight: 224,
99 }),
100 ).toBe(true);
101
102 expect(
103 shouldLoadNextBranchPageAfterScroll({
104 previousScrollTop: 499.5,
105 scrollTop: 500,
106 scrollHeight: 800,
107 clientHeight: 224,
108 }),
109 ).toBe(true);
110 });
111});
apps/web/src/state/paginatedBranches.ts created+36
......@@ -0,0 +1,36 @@
1import type { VcsListRefsResult } from "@t3tools/contracts";
2import * as Option from "effect/Option";
3import { AsyncResult } from "effect/unstable/reactivity";
4
5const DEFAULT_NEXT_PAGE_DISTANCE_PX = 96;
6
7export function isPaginatedBranchesNextPagePending<E>(
8 results: ReadonlyArray<AsyncResult.AsyncResult<VcsListRefsResult, E>>,
9): boolean {
10 const lastResult = results.at(-1);
11 return (
12 results.length > 1 &&
13 lastResult?.waiting === true &&
14 Option.isNone(AsyncResult.value(lastResult))
15 );
16}
17
18export function shouldLoadNextBranchPageAfterScroll({
19 previousScrollTop,
20 scrollTop,
21 scrollHeight,
22 clientHeight,
23 distanceThreshold = DEFAULT_NEXT_PAGE_DISTANCE_PX,
24}: {
25 previousScrollTop: number | null;
26 scrollTop: number;
27 scrollHeight: number;
28 clientHeight: number;
29 distanceThreshold?: number;
30}): boolean {
31 if (previousScrollTop === null || scrollTop <= previousScrollTop) {
32 return false;
33 }
34
35 return scrollHeight - scrollTop - clientHeight <= distanceThreshold;
36}
apps/web/src/state/queries.ts+3
......@@ -18,6 +18,7 @@ import { useCallback, useEffect, useMemo, useState } from "react";
1818
1919import { appAtomRegistry } from "../rpc/atomRegistry";
2020import { orchestrationEnvironment } from "./orchestration";
21import { isPaginatedBranchesNextPagePending } from "./paginatedBranches";
2122import { projectEnvironment } from "./projects";
2223import { useEnvironmentQuery } from "./query";
2324import { useEnvironmentThread } from "./threads";
......@@ -142,6 +143,7 @@ export function usePaginatedBranches(target: VcsRefTarget) {
142143 totalCount: Math.max(...values.map((value) => value.totalCount)),
143144 };
144145 const failed = results.find((result) => result._tag === "Failure");
146 const isFetchingNextPage = isPaginatedBranchesNextPagePending(results);
145147 const error =
146148 failed?._tag === "Failure"
147149 ? (() => {
......@@ -176,6 +178,7 @@ export function usePaginatedBranches(target: VcsRefTarget) {
176178 refs: data?.refs ?? EMPTY_REFS,
177179 error,
178180 isPending: results.some((result) => result.waiting),
181 isFetchingNextPage,
179182 refresh,
180183 loadNext,
181184 };
packages/client-runtime/src/connection/registry.test.ts+2
......@@ -251,6 +251,8 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* (
251251 saveServerConfig: () => Effect.void,
252252 loadVcsRefs: () => Effect.succeed(Option.none()),
253253 saveVcsRefs: () => Effect.void,
254 removeVcsRefs: () => Effect.void,
255 clearVcsRefs: () => Effect.void,
254256 clear: (environmentId) =>
255257 Ref.update(shellCache, (current) => {
256258 const next = new Map(current);
packages/client-runtime/src/platform/persistence.ts+14
......@@ -30,6 +30,8 @@ export class ConnectionPersistenceError extends Schema.TaggedErrorClass<Connecti
3030 "save-server-config",
3131 "load-vcs-refs",
3232 "save-vcs-refs",
33 "remove-vcs-refs",
34 "clear-vcs-refs",
3335 "clear-environment",
3436 ]),
3537 message: Schema.String,
......@@ -102,6 +104,18 @@ export class EnvironmentCacheStore extends Context.Service<
102104 cwd: string,
103105 refs: VcsListRefsResult,
104106 ) => Effect.Effect<void, ConnectionPersistenceError>;
107 readonly removeVcsRefs: (
108 environmentId: EnvironmentId,
109 cwd: string,
110 ) => Effect.Effect<void, ConnectionPersistenceError>;
111 /**
112 * Removes every persisted branch-list snapshot for an environment. Git ref
113 * mutations are repository-wide, and linked worktrees may have cached the
114 * same refs under different working-directory keys.
115 */
116 readonly clearVcsRefs: (
117 environmentId: EnvironmentId,
118 ) => Effect.Effect<void, ConnectionPersistenceError>;
105119 readonly clear: (
106120 environmentId: EnvironmentId,
107121 ) => Effect.Effect<void, ConnectionPersistenceError>;
packages/client-runtime/src/state/runtime.ts+41-3
......@@ -32,6 +32,17 @@ interface EnvironmentAtomOptions<Input, A, E, R> {
3232 }>;
3333}
3434
35interface EnvironmentCommandAtomOptions<Input, A, E, R> extends Omit<
36 EnvironmentAtomOptions<Input, A, E, R>,
37 "execute"
38> {
39 readonly execute: (
40 input: Input,
41 registry: AtomRegistry.AtomRegistry,
42 environmentId: EnvironmentIdType,
43 ) => Effect.Effect<A, E, R>;
44}
45
3546interface EnvironmentQueryAtomOptions<Input, A, E, R> extends EnvironmentAtomOptions<
3647 Input,
3748 A,
......@@ -520,13 +531,17 @@ export function createEnvironmentSubscriptionAtomFamily<R, ER, Input, A, E>(
520531
521532export function createEnvironmentCommand<R, ER, Input, A, E>(
522533 runtime: Atom.AtomRuntime<EnvironmentRegistry | R, ER>,
523 options: EnvironmentAtomOptions<Input, A, E, EnvironmentSupervisor | R>,
534 options: EnvironmentCommandAtomOptions<Input, A, E, EnvironmentSupervisor | R>,
524535) {
525536 return createRuntimeCommand(runtime, {
526537 label: options.label,
527538 ...(options.scheduler === undefined ? {} : { scheduler: options.scheduler }),
528539 ...(options.concurrency === undefined ? {} : { concurrency: options.concurrency }),
529 execute: (target) => runInEnvironment(target.environmentId, options.execute(target.input)),
540 execute: (target, registry) =>
541 runInEnvironment(
542 target.environmentId,
543 options.execute(target.input, registry, target.environmentId),
544 ),
530545 });
531546}
532547
......@@ -616,13 +631,36 @@ export function createEnvironmentRpcCommand<R, ER, TTag extends EnvironmentUnary
616631 readonly environmentId: EnvironmentIdType;
617632 readonly input: EnvironmentRpcInput<TTag>;
618633 }>;
634 readonly onSuccess?: (
635 target: {
636 readonly environmentId: EnvironmentIdType;
637 readonly input: EnvironmentRpcInput<TTag>;
638 },
639 registry: AtomRegistry.AtomRegistry,
640 ) => Effect.Effect<void, never, R>;
641 readonly onSettled?: (
642 target: {
643 readonly environmentId: EnvironmentIdType;
644 readonly input: EnvironmentRpcInput<TTag>;
645 },
646 registry: AtomRegistry.AtomRegistry,
647 ) => Effect.Effect<void, never, R>;
619648 },
620649) {
621650 return createEnvironmentCommand(runtime, {
622651 label: options.label,
623652 ...(options.scheduler === undefined ? {} : { scheduler: options.scheduler }),
624653 ...(options.concurrency === undefined ? {} : { concurrency: options.concurrency }),
625 execute: (input: EnvironmentRpcInput<TTag>) => request(options.tag, input),
654 execute: (input: EnvironmentRpcInput<TTag>, registry, environmentId) => {
655 const target = {
656 environmentId,
657 input,
658 };
659 return request(options.tag, input).pipe(
660 Effect.tap(() => options.onSuccess?.(target, registry) ?? Effect.void),
661 Effect.ensuring(options.onSettled?.(target, registry) ?? Effect.void),
662 );
663 },
626664 });
627665}
628666
packages/client-runtime/src/state/server.test.ts+4
......@@ -152,6 +152,8 @@ describe("server state projection", () => {
152152 saveServerConfig: (_environmentId, config) => Queue.offer(savedConfigs, config),
153153 loadVcsRefs: () => Effect.succeed(Option.none()),
154154 saveVcsRefs: () => Effect.void,
155 removeVcsRefs: () => Effect.void,
156 clearVcsRefs: () => Effect.void,
155157 clear: () => Effect.void,
156158 });
157159
......@@ -211,6 +213,8 @@ describe("server state projection", () => {
211213 saveServerConfig: (_environmentId, config) => Queue.offer(savedConfigs, config),
212214 loadVcsRefs: () => Effect.succeed(Option.none()),
213215 saveVcsRefs: () => Effect.void,
216 removeVcsRefs: () => Effect.void,
217 clearVcsRefs: () => Effect.void,
214218 clear: () => Effect.void,
215219 });
216220
packages/client-runtime/src/state/shell-sync.test.ts+6
......@@ -87,6 +87,8 @@ describe("environment shell synchronization", () => {
8787 saveServerConfig: () => Effect.void,
8888 loadVcsRefs: () => Effect.succeed(Option.none()),
8989 saveVcsRefs: () => Effect.void,
90 removeVcsRefs: () => Effect.void,
91 clearVcsRefs: () => Effect.void,
9092 clear: () => Effect.void,
9193 });
9294 // Cold cache with no HTTP snapshot available → falls back to the
......@@ -201,6 +203,8 @@ describe("environment shell synchronization", () => {
201203 saveServerConfig: () => Effect.void,
202204 loadVcsRefs: () => Effect.succeed(Option.none()),
203205 saveVcsRefs: () => Effect.void,
206 removeVcsRefs: () => Effect.void,
207 clearVcsRefs: () => Effect.void,
204208 clear: () => Effect.void,
205209 });
206210 const snapshotLoader = ShellSnapshotLoader.of({
......@@ -270,6 +274,8 @@ describe("environment shell synchronization", () => {
270274 saveServerConfig: () => Effect.void,
271275 loadVcsRefs: () => Effect.succeed(Option.none()),
272276 saveVcsRefs: () => Effect.void,
277 removeVcsRefs: () => Effect.void,
278 clearVcsRefs: () => Effect.void,
273279 clear: () => Effect.void,
274280 });
275281 const snapshotLoader = ShellSnapshotLoader.of({
packages/client-runtime/src/state/sourceControl.test.ts created+169
......@@ -0,0 +1,169 @@
1import {
2 EnvironmentId,
3 WS_METHODS,
4 type SourceControlPublishRepositoryResult,
5} from "@t3tools/contracts";
6import { describe, expect, it } from "@effect/vitest";
7import * as Effect from "effect/Effect";
8import * as Layer from "effect/Layer";
9import * as Option from "effect/Option";
10import * as SubscriptionRef from "effect/SubscriptionRef";
11import { AsyncResult, Atom, AtomRegistry } from "effect/unstable/reactivity";
12
13import {
14 AVAILABLE_CONNECTION_STATE,
15 PrimaryConnectionTarget,
16 type PreparedConnection,
17 type SupervisorConnectionState,
18} from "../connection/model.ts";
19import * as EnvironmentRegistry from "../connection/registry.ts";
20import * as EnvironmentSupervisor from "../connection/supervisor.ts";
21import * as Persistence from "../platform/persistence.ts";
22import { EnvironmentRpcUnavailableError } from "../rpc/client.ts";
23import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
24import type { RpcSession } from "../rpc/session.ts";
25import { createSourceControlEnvironmentAtoms } from "./sourceControl.ts";
26import { vcsRefsCacheStateAtom } from "./vcsRefInvalidation.ts";
27
28const TARGET = new PrimaryConnectionTarget({
29 environmentId: EnvironmentId.make("environment-1"),
30 label: "Test environment",
31 httpBaseUrl: "https://environment.example.test",
32 wsBaseUrl: "wss://environment.example.test",
33});
34
35const PUBLISH_RESULT: SourceControlPublishRepositoryResult = {
36 repository: {
37 provider: "github",
38 nameWithOwner: "t3tools/t3code",
39 url: "https://github.com/t3tools/t3code",
40 sshUrl: "git@github.com:t3tools/t3code.git",
41 },
42 remoteName: "origin",
43 remoteUrl: "git@github.com:t3tools/t3code.git",
44 branch: "main",
45 upstreamBranch: "origin/main",
46 status: "pushed",
47};
48
49function session(client: WsRpcProtocolClient): RpcSession {
50 return {
51 client,
52 initialConfig: Effect.never,
53 ready: Effect.void,
54 probe: Effect.void,
55 closed: Effect.never,
56 };
57}
58
59describe("source control environment atoms", () => {
60 it.effect("invalidates cached refs after successful and failed publishing", () =>
61 Effect.scoped(
62 Effect.gen(function* () {
63 const connectionState: SupervisorConnectionState = {
64 ...AVAILABLE_CONNECTION_STATE,
65 desired: true,
66 network: "online",
67 phase: "connected",
68 attempt: 1,
69 generation: 1,
70 };
71 let publishAttempts = 0;
72 const client = {
73 [WS_METHODS.sourceControlPublishRepository]: () => {
74 publishAttempts += 1;
75 return publishAttempts === 1
76 ? Effect.succeed(PUBLISH_RESULT)
77 : Effect.fail(
78 new EnvironmentRpcUnavailableError({
79 environmentId: TARGET.environmentId,
80 message: "push failed after adding the remote",
81 }),
82 );
83 },
84 } as unknown as WsRpcProtocolClient;
85 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
86 target: TARGET,
87 state: yield* SubscriptionRef.make(connectionState),
88 session: yield* SubscriptionRef.make(Option.some(session(client))),
89 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
90 connect: Effect.void,
91 disconnect: Effect.void,
92 retryNow: Effect.void,
93 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
94 const run: EnvironmentRegistry.EnvironmentRegistry["Service"]["run"] = (
95 _environmentId,
96 effect,
97 ) => Effect.provideService(effect, EnvironmentSupervisor.EnvironmentSupervisor, supervisor);
98 const environmentRegistry = EnvironmentRegistry.EnvironmentRegistry.of({
99 run,
100 } as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]);
101 const removed = new Array<string>();
102 const cache = Persistence.EnvironmentCacheStore.of({
103 loadShell: () => Effect.succeed(Option.none()),
104 saveShell: () => Effect.void,
105 loadThread: () => Effect.succeed(Option.none()),
106 saveThread: () => Effect.void,
107 removeThread: () => Effect.void,
108 loadServerConfig: () => Effect.succeed(Option.none()),
109 saveServerConfig: () => Effect.void,
110 loadVcsRefs: () => Effect.succeed(Option.none()),
111 saveVcsRefs: () => Effect.void,
112 removeVcsRefs: (environmentId, cwd) =>
113 Effect.sync(() => {
114 removed.push(`${environmentId}:${cwd}`);
115 }),
116 clearVcsRefs: (environmentId) =>
117 Effect.sync(() => {
118 removed.push(`${environmentId}:*`);
119 }),
120 clear: () => Effect.void,
121 });
122 const runtime = Atom.runtime(
123 Layer.merge(
124 Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
125 Layer.succeed(Persistence.EnvironmentCacheStore, cache),
126 ),
127 );
128 const atoms = createSourceControlEnvironmentAtoms(runtime);
129 const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
130 Effect.sync(() => registry.dispose()),
131 );
132 const state = vcsRefsCacheStateAtom({ environmentId: TARGET.environmentId });
133
134 expect(registry.get(state).revision).toBe(0);
135 const publishResult = yield* Effect.promise(() =>
136 atoms.publishRepository.run(registry, {
137 environmentId: TARGET.environmentId,
138 input: {
139 cwd: "/repo",
140 provider: "github",
141 repository: "t3tools/t3code",
142 visibility: "private",
143 },
144 }),
145 );
146
147 expect(AsyncResult.isSuccess(publishResult)).toBe(true);
148 expect(registry.get(state).revision).toBe(1);
149 expect(removed).toEqual([`${TARGET.environmentId}:*`]);
150
151 const failedPublish = yield* Effect.promise(() =>
152 atoms.publishRepository.run(registry, {
153 environmentId: TARGET.environmentId,
154 input: {
155 cwd: "/repo",
156 provider: "github",
157 repository: "t3tools/t3code",
158 visibility: "private",
159 },
160 }),
161 );
162
163 expect(AsyncResult.isFailure(failedPublish)).toBe(true);
164 expect(registry.get(state).revision).toBe(2);
165 expect(removed).toEqual([`${TARGET.environmentId}:*`, `${TARGET.environmentId}:*`]);
166 }),
167 ),
168 );
169});
packages/client-runtime/src/state/sourceControl.ts+8-1
......@@ -7,10 +7,12 @@ import {
77 createEnvironmentRpcQueryAtomFamily,
88} from "./runtime.ts";
99import type { EnvironmentRegistry } from "../connection/registry.ts";
10import { EnvironmentCacheStore } from "../platform/persistence.ts";
1011import { vcsCommandConcurrency, vcsCommandScheduler } from "./vcsCommandScheduler.ts";
12import { invalidateCachedVcsRefs } from "./vcsRefInvalidation.ts";
1113
1214export function createSourceControlEnvironmentAtoms<R, E>(
13 runtime: Atom.AtomRuntime<EnvironmentRegistry | R, E>,
15 runtime: Atom.AtomRuntime<EnvironmentRegistry | EnvironmentCacheStore | R, E>,
1416) {
1517 const commandScheduler = createAtomCommandScheduler();
1618 return {
......@@ -36,6 +38,11 @@ export function createSourceControlEnvironmentAtoms<R, E>(
3638 tag: WS_METHODS.sourceControlPublishRepository,
3739 scheduler: vcsCommandScheduler,
3840 concurrency: vcsCommandConcurrency,
41 onSettled: (target, registry) =>
42 invalidateCachedVcsRefs(registry, {
43 environmentId: target.environmentId,
44 cwd: target.input.cwd,
45 }),
3946 }),
4047 };
4148}
packages/client-runtime/src/state/threads-sync.test.ts+2
......@@ -217,6 +217,8 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
217217 saveServerConfig: () => Effect.void,
218218 loadVcsRefs: () => Effect.succeed(Option.none()),
219219 saveVcsRefs: () => Effect.void,
220 removeVcsRefs: () => Effect.void,
221 clearVcsRefs: () => Effect.void,
220222 clear: () => Effect.void,
221223 });
222224 const threadState = yield* makeEnvironmentThreadState(THREAD_ID).pipe(
packages/client-runtime/src/state/vcs.test.ts+432-20
......@@ -1,12 +1,19 @@
1import { EnvironmentId, WS_METHODS, type VcsListRefsResult } from "@t3tools/contracts";
1import {
2 EnvironmentId,
3 WS_METHODS,
4 type VcsListRefsInput,
5 type VcsListRefsResult,
6} from "@t3tools/contracts";
27import { describe, expect, it } from "@effect/vitest";
38import * as Effect from "effect/Effect";
49import * as Fiber from "effect/Fiber";
10import * as Layer from "effect/Layer";
511import * as Option from "effect/Option";
612import * as Ref from "effect/Ref";
713import * as Stream from "effect/Stream";
814import * as SubscriptionRef from "effect/SubscriptionRef";
915import * as TestClock from "effect/testing/TestClock";
16import { AsyncResult, Atom, AtomRegistry } from "effect/unstable/reactivity";
1017
1118import {
1219 AVAILABLE_CONNECTION_STATE,
......@@ -14,11 +21,23 @@ import {
1421 type PreparedConnection,
1522 type SupervisorConnectionState,
1623} from "../connection/model.ts";
24import * as EnvironmentRegistry from "../connection/registry.ts";
1725import * as EnvironmentSupervisor from "../connection/supervisor.ts";
1826import * as Persistence from "../platform/persistence.ts";
27import { EnvironmentRpcUnavailableError } from "../rpc/client.ts";
1928import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
2029import type { RpcSession } from "../rpc/session.ts";
21import { makeCachedVcsRefsChanges } from "./vcs.ts";
30
31import {
32 commitVcsRefsRefresh,
33 createVcsEnvironmentAtoms,
34 makeCachedVcsRefsChanges,
35} from "./vcs.ts";
36import {
37 invalidateCachedVcsRefs,
38 invalidateVcsRefs,
39 vcsRefsCacheStateAtom,
40} from "./vcsRefInvalidation.ts";
2241
2342const TARGET = new PrimaryConnectionTarget({
2443 environmentId: EnvironmentId.make("environment-1"),
......@@ -73,7 +92,10 @@ function session(client: WsRpcProtocolClient): RpcSession {
7392 };
7493}
7594
76function cacheWithRefs(refs: Option.Option<VcsListRefsResult>) {
95function cacheWithRefs(
96 refs: Option.Option<VcsListRefsResult>,
97 overrides: Partial<Persistence.EnvironmentCacheStore["Service"]> = {},
98) {
7799 return Persistence.EnvironmentCacheStore.of({
78100 loadShell: () => Effect.succeed(Option.none()),
79101 saveShell: () => Effect.void,
......@@ -84,11 +106,345 @@ function cacheWithRefs(refs: Option.Option<VcsListRefsResult>) {
84106 saveServerConfig: () => Effect.void,
85107 loadVcsRefs: () => Effect.succeed(refs),
86108 saveVcsRefs: () => Effect.void,
109 removeVcsRefs: () => Effect.void,
110 clearVcsRefs: () => Effect.void,
87111 clear: () => Effect.void,
112 ...overrides,
88113 });
89114}
90115
91116describe("cached VCS refs", () => {
117 it("invalidates all ref streams in the mutated environment", () => {
118 const registry = AtomRegistry.make();
119 const environment = {
120 environmentId: TARGET.environmentId,
121 };
122 const otherEnvironment = {
123 environmentId: EnvironmentId.make("environment-2"),
124 };
125
126 expect(registry.get(vcsRefsCacheStateAtom(environment))).toEqual({
127 revision: 0,
128 persistedCacheReadable: true,
129 });
130 expect(registry.get(vcsRefsCacheStateAtom(otherEnvironment))).toEqual({
131 revision: 0,
132 persistedCacheReadable: true,
133 });
134
135 invalidateVcsRefs(registry, environment);
136
137 expect(registry.get(vcsRefsCacheStateAtom(environment))).toEqual({
138 revision: 1,
139 persistedCacheReadable: true,
140 });
141 expect(registry.get(vcsRefsCacheStateAtom(otherEnvironment))).toEqual({
142 revision: 0,
143 persistedCacheReadable: true,
144 });
145 registry.dispose();
146 });
147
148 it.effect("preserves the caller's repository snapshot refresh policy", () =>
149 Effect.scoped(
150 Effect.gen(function* () {
151 const requests = yield* Ref.make<ReadonlyArray<VcsListRefsInput>>([]);
152 const client = {
153 [WS_METHODS.vcsListRefs]: (input: VcsListRefsInput) =>
154 Ref.update(requests, (current) => [...current, input]).pipe(Effect.as(LIVE_REFS)),
155 } as unknown as WsRpcProtocolClient;
156 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
157 target: TARGET,
158 state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE),
159 session: yield* SubscriptionRef.make(Option.some(session(client))),
160 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
161 connect: Effect.void,
162 disconnect: Effect.void,
163 retryNow: Effect.void,
164 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
165
166 yield* Stream.unwrap(
167 makeCachedVcsRefsChanges({
168 cwd: "/repo",
169 limit: 20,
170 query: "release",
171 refKind: "remote",
172 }).pipe(
173 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
174 Effect.provideService(Persistence.EnvironmentCacheStore, cacheWithRefs(Option.none())),
175 ),
176 ).pipe(Stream.runHead);
177
178 expect(yield* Ref.get(requests)).toEqual([
179 {
180 cwd: "/repo",
181 limit: 20,
182 query: "release",
183 refKind: "remote",
184 },
185 ]);
186
187 yield* Stream.unwrap(
188 makeCachedVcsRefsChanges({
189 cwd: "/repo",
190 limit: 20,
191 query: "release",
192 refKind: "remote",
193 refresh: true,
194 }).pipe(
195 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
196 Effect.provideService(Persistence.EnvironmentCacheStore, cacheWithRefs(Option.none())),
197 ),
198 ).pipe(Stream.runHead);
199
200 expect(yield* Ref.get(requests)).toEqual([
201 {
202 cwd: "/repo",
203 limit: 20,
204 query: "release",
205 refKind: "remote",
206 },
207 {
208 cwd: "/repo",
209 limit: 20,
210 query: "release",
211 refKind: "remote",
212 refresh: true,
213 },
214 ]);
215 }),
216 ),
217 );
218
219 it.effect("does not repersist a refresh superseded by ref invalidation", () =>
220 Effect.scoped(
221 Effect.gen(function* () {
222 const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
223 Effect.sync(() => registry.dispose()),
224 );
225 const saved = yield* Ref.make<ReadonlyArray<VcsListRefsResult>>([]);
226 const clears = yield* Ref.make(0);
227 const revisionsObservedDuringClear = yield* Ref.make<ReadonlyArray<number>>([]);
228 const cache = cacheWithRefs(Option.none(), {
229 saveVcsRefs: (_environmentId, _cwd, refs) =>
230 Ref.update(saved, (current) => [...current, refs]),
231 clearVcsRefs: () =>
232 Effect.all([
233 Ref.update(clears, (count) => count + 1),
234 Ref.update(revisionsObservedDuringClear, (current) => [
235 ...current,
236 registry.get(vcsRefsCacheStateAtom(TARGET)).revision,
237 ]),
238 ]).pipe(Effect.asVoid),
239 });
240
241 yield* invalidateCachedVcsRefs(registry, {
242 environmentId: TARGET.environmentId,
243 cwd: "/repo-worktree",
244 }).pipe(Effect.provideService(Persistence.EnvironmentCacheStore, cache));
245
246 expect(
247 yield* commitVcsRefsRefresh(registry, cache, {
248 environmentId: TARGET.environmentId,
249 cwd: "/repo",
250 refs: CACHED_REFS,
251 expectedRevision: 0,
252 persist: true,
253 }),
254 ).toBe(false);
255
256 expect(yield* Ref.get(saved)).toEqual([]);
257 expect(yield* Ref.get(clears)).toBe(1);
258 expect(yield* Ref.get(revisionsObservedDuringClear)).toEqual([0]);
259 expect(registry.get(vcsRefsCacheStateAtom(TARGET))).toEqual({
260 revision: 1,
261 persistedCacheReadable: true,
262 });
263
264 expect(
265 yield* commitVcsRefsRefresh(registry, cache, {
266 environmentId: TARGET.environmentId,
267 cwd: "/repo",
268 refs: LIVE_REFS,
269 expectedRevision: 1,
270 persist: true,
271 }),
272 ).toBe(true);
273 expect(yield* Ref.get(saved)).toEqual([LIVE_REFS]);
274 }),
275 ),
276 );
277
278 it.effect("invalidates persisted refs when ref-affecting commands settle", () =>
279 Effect.scoped(
280 Effect.gen(function* () {
281 const expectedError = new EnvironmentRpcUnavailableError({
282 environmentId: TARGET.environmentId,
283 message: "pull failed after fetching refs",
284 });
285 const client = {
286 [WS_METHODS.vcsPull]: () => Effect.fail(expectedError),
287 [WS_METHODS.vcsRefreshStatus]: () => Effect.void,
288 } as unknown as WsRpcProtocolClient;
289 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
290 target: TARGET,
291 state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE),
292 session: yield* SubscriptionRef.make(Option.some(session(client))),
293 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
294 connect: Effect.void,
295 disconnect: Effect.void,
296 retryNow: Effect.void,
297 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
298 const run: EnvironmentRegistry.EnvironmentRegistry["Service"]["run"] = (
299 _environmentId,
300 effect,
301 ) => Effect.provideService(effect, EnvironmentSupervisor.EnvironmentSupervisor, supervisor);
302 const environmentRegistry = EnvironmentRegistry.EnvironmentRegistry.of({
303 run,
304 } as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]);
305 const clears = yield* Ref.make(0);
306 const runtime = Atom.runtime(
307 Layer.merge(
308 Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
309 Layer.succeed(
310 Persistence.EnvironmentCacheStore,
311 cacheWithRefs(Option.none(), {
312 clearVcsRefs: () => Ref.update(clears, (count) => count + 1),
313 }),
314 ),
315 ),
316 );
317 const atoms = createVcsEnvironmentAtoms(runtime);
318 const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
319 Effect.sync(() => registry.dispose()),
320 );
321
322 const result = yield* Effect.promise(() =>
323 atoms.pull.run(registry, {
324 environmentId: TARGET.environmentId,
325 input: { cwd: "/repo" },
326 }),
327 );
328
329 expect(AsyncResult.isFailure(result)).toBe(true);
330 expect(yield* Ref.get(clears)).toBe(1);
331 expect(registry.get(vcsRefsCacheStateAtom(TARGET)).revision).toBe(1);
332
333 const refreshResult = yield* Effect.promise(() =>
334 atoms.refreshStatus.run(registry, {
335 environmentId: TARGET.environmentId,
336 input: { cwd: "/repo" },
337 }),
338 );
339
340 expect(AsyncResult.isSuccess(refreshResult)).toBe(true);
341 expect(yield* Ref.get(clears)).toBe(2);
342 expect(registry.get(vcsRefsCacheStateAtom(TARGET)).revision).toBe(2);
343 }),
344 ),
345 );
346
347 it.effect("suppresses persisted snapshots after an environment-wide clear fails", () =>
348 Effect.scoped(
349 Effect.gen(function* () {
350 const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
351 Effect.sync(() => registry.dispose()),
352 );
353 const persisted = yield* Ref.make<Option.Option<VcsListRefsResult>>(
354 Option.some(CACHED_REFS),
355 );
356 const loads = yield* Ref.make(0);
357 const clearAttempts = yield* Ref.make(0);
358 const cache = cacheWithRefs(Option.none(), {
359 loadVcsRefs: () =>
360 Ref.update(loads, (count) => count + 1).pipe(Effect.andThen(Ref.get(persisted))),
361 saveVcsRefs: (_environmentId, _cwd, refs) => Ref.set(persisted, Option.some(refs)),
362 clearVcsRefs: () =>
363 Ref.updateAndGet(clearAttempts, (count) => count + 1).pipe(
364 Effect.flatMap((attempt) =>
365 attempt === 1
366 ? Effect.fail(
367 new Persistence.ConnectionPersistenceError({
368 operation: "clear-vcs-refs",
369 message: "storage unavailable",
370 }),
371 )
372 : Ref.set(persisted, Option.none()),
373 ),
374 ),
375 });
376
377 yield* invalidateCachedVcsRefs(registry, {
378 environmentId: TARGET.environmentId,
379 cwd: "/repo",
380 }).pipe(Effect.provideService(Persistence.EnvironmentCacheStore, cache));
381
382 const state = registry.get(vcsRefsCacheStateAtom(TARGET));
383 expect(state).toEqual({
384 revision: 1,
385 persistedCacheReadable: false,
386 });
387
388 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
389 target: TARGET,
390 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
391 session: yield* SubscriptionRef.make(Option.none<RpcSession>()),
392 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
393 connect: Effect.void,
394 disconnect: Effect.void,
395 retryNow: Effect.void,
396 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
397 const refs = yield* Stream.unwrap(
398 makeCachedVcsRefsChanges(
399 { cwd: "/repo", limit: 100 },
400 state.revision,
401 registry,
402 state.persistedCacheReadable,
403 ).pipe(
404 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
405 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
406 ),
407 ).pipe(Stream.runHead, Effect.forkChild({ startImmediately: true }));
408
409 yield* Effect.yieldNow;
410 expect(yield* Ref.get(loads)).toBe(0);
411 yield* Fiber.interrupt(refs);
412 expect(registry.get(vcsRefsCacheStateAtom(TARGET))).toEqual(state);
413
414 expect(
415 yield* commitVcsRefsRefresh(registry, cache, {
416 environmentId: TARGET.environmentId,
417 cwd: "/repo",
418 refs: LIVE_REFS,
419 expectedRevision: state.revision,
420 persist: true,
421 }),
422 ).toBe(true);
423 const recoveredState = registry.get(vcsRefsCacheStateAtom(TARGET));
424 expect(recoveredState).toEqual({
425 revision: 1,
426 persistedCacheReadable: true,
427 });
428 expect(yield* Ref.get(clearAttempts)).toBe(2);
429
430 const recoveredRefs = yield* Stream.unwrap(
431 makeCachedVcsRefsChanges(
432 { cwd: "/repo", limit: 100 },
433 recoveredState.revision,
434 registry,
435 recoveredState.persistedCacheReadable,
436 ).pipe(
437 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
438 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
439 ),
440 ).pipe(Stream.runHead);
441
442 expect(Option.getOrThrow(recoveredRefs)).toEqual(LIVE_REFS);
443 expect(yield* Ref.get(loads)).toBe(1);
444 }),
445 ),
446 );
447
92448 it.effect("loads an unfiltered branch list without a connection", () =>
93449 Effect.scoped(
94450 Effect.gen(function* () {
......@@ -116,11 +472,12 @@ describe("cached VCS refs", () => {
116472 ),
117473 );
118474
119 it.effect("continues polling after a transient live failure", () =>
475 it.effect("retries a transient live failure while connected", () =>
120476 Effect.scoped(
121477 Effect.gen(function* () {
122478 const expectedError = new Error("Could not list Git refs.");
123479 const calls = yield* Ref.make(0);
480 const connectionState = yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE);
124481 const client = {
125482 [WS_METHODS.vcsListRefs]: () =>
126483 Ref.updateAndGet(calls, (count) => count + 1).pipe(
......@@ -131,7 +488,7 @@ describe("cached VCS refs", () => {
131488 } as unknown as WsRpcProtocolClient;
132489 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
133490 target: TARGET,
134 state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE),
491 state: connectionState,
135492 session: yield* SubscriptionRef.make(Option.some(session(client))),
136493 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
137494 connect: Effect.void,
......@@ -142,10 +499,7 @@ describe("cached VCS refs", () => {
142499 const result = Stream.unwrap(
143500 makeCachedVcsRefsChanges({ cwd: "/repo", limit: 100 }).pipe(
144501 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
145 Effect.provideService(
146 Persistence.EnvironmentCacheStore,
147 cacheWithRefs(Option.some(CACHED_REFS)),
148 ),
502 Effect.provideService(Persistence.EnvironmentCacheStore, cacheWithRefs(Option.none())),
149503 ),
150504 ).pipe(Stream.runHead);
151505 const fiber = yield* Effect.forkChild(result);
......@@ -155,22 +509,78 @@ describe("cached VCS refs", () => {
155509 }
156510 expect(yield* Ref.get(calls)).toBe(1);
157511
158 yield* TestClock.adjust("5 seconds");
512 yield* TestClock.adjust("1 second");
159513 expect(Option.getOrThrow(yield* Fiber.join(fiber))).toEqual(LIVE_REFS);
514 expect(yield* Ref.get(calls)).toBe(2);
160515 }).pipe(Effect.provide(TestClock.layer())),
161516 ),
162517 );
163518
164 it.effect("revalidates connected refs every five seconds", () =>
519 it.effect("cancels an in-flight refresh when the connection generation changes", () =>
165520 Effect.scoped(
166521 Effect.gen(function* () {
167522 const calls = yield* Ref.make(0);
523 const interruptions = yield* Ref.make(0);
524 const connectionState = yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE);
168525 const client = {
169526 [WS_METHODS.vcsListRefs]: () =>
170527 Ref.updateAndGet(calls, (count) => count + 1).pipe(
171 Effect.map((count) => (count === 1 ? CACHED_REFS : LIVE_REFS)),
528 Effect.flatMap((count) =>
529 count === 1
530 ? Effect.never.pipe(
531 Effect.onInterrupt(() =>
532 Ref.update(interruptions, (interruptions) => interruptions + 1),
533 ),
534 )
535 : Effect.succeed(LIVE_REFS),
536 ),
172537 ),
173538 } as unknown as WsRpcProtocolClient;
539 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
540 target: TARGET,
541 state: connectionState,
542 session: yield* SubscriptionRef.make(Option.some(session(client))),
543 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
544 connect: Effect.void,
545 disconnect: Effect.void,
546 retryNow: Effect.void,
547 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
548
549 const result = Stream.unwrap(
550 makeCachedVcsRefsChanges({ cwd: "/repo", limit: 100 }).pipe(
551 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
552 Effect.provideService(Persistence.EnvironmentCacheStore, cacheWithRefs(Option.none())),
553 ),
554 ).pipe(Stream.runHead);
555 const fiber = yield* Effect.forkChild(result);
556
557 for (let attempt = 0; attempt < 100 && (yield* Ref.get(calls)) < 1; attempt += 1) {
558 yield* Effect.yieldNow;
559 }
560 yield* SubscriptionRef.set(connectionState, AVAILABLE_CONNECTION_STATE);
561 for (let attempt = 0; attempt < 100 && (yield* Ref.get(interruptions)) < 1; attempt += 1) {
562 yield* Effect.yieldNow;
563 }
564 expect(yield* Ref.get(interruptions)).toBe(1);
565
566 yield* SubscriptionRef.set(connectionState, {
567 ...CONNECTED_CONNECTION_STATE,
568 generation: 2,
569 });
570 expect(Option.getOrThrow(yield* Fiber.join(fiber))).toEqual(LIVE_REFS);
571 expect(yield* Ref.get(calls)).toBe(2);
572 }),
573 ),
574 );
575
576 it.effect("does not poll refs while the connection remains stable", () =>
577 Effect.scoped(
578 Effect.gen(function* () {
579 const calls = yield* Ref.make(0);
580 const client = {
581 [WS_METHODS.vcsListRefs]: () =>
582 Ref.update(calls, (count) => count + 1).pipe(Effect.as(CACHED_REFS)),
583 } as unknown as WsRpcProtocolClient;
174584 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
175585 target: TARGET,
176586 state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE),
......@@ -180,26 +590,28 @@ describe("cached VCS refs", () => {
180590 disconnect: Effect.void,
181591 retryNow: Effect.void,
182592 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
183 const results = Stream.unwrap(
593 const stream = Stream.unwrap(
184594 makeCachedVcsRefsChanges({ cwd: "/repo", limit: 100 }).pipe(
185595 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
186596 Effect.provideService(Persistence.EnvironmentCacheStore, cacheWithRefs(Option.none())),
187597 ),
188 ).pipe(Stream.take(2), Stream.runCollect);
189 const fiber = yield* Effect.forkChild(results);
598 ).pipe(Stream.runDrain);
599 const fiber = yield* Effect.forkChild(stream);
190600
191601 for (let attempt = 0; attempt < 100 && (yield* Ref.get(calls)) < 1; attempt += 1) {
192602 yield* Effect.yieldNow;
193603 }
194604 expect(yield* Ref.get(calls)).toBe(1);
195605
196 yield* TestClock.adjust("5 seconds");
197 expect(Array.from(yield* Fiber.join(fiber))).toEqual([CACHED_REFS, LIVE_REFS]);
606 yield* TestClock.adjust("1 minute");
607 yield* Effect.yieldNow;
608 expect(yield* Ref.get(calls)).toBe(1);
609 yield* Fiber.interrupt(fiber);
198610 }).pipe(Effect.provide(TestClock.layer())),
199611 ),
200612 );
201613
202 it.effect("does not emit persisted refs before a live refresh", () =>
614 it.effect("emits persisted refs before a live refresh", () =>
203615 Effect.scoped(
204616 Effect.gen(function* () {
205617 const client = {
......@@ -223,9 +635,9 @@ describe("cached VCS refs", () => {
223635 cacheWithRefs(Option.some(CACHED_REFS)),
224636 ),
225637 ),
226 ).pipe(Stream.runHead);
638 ).pipe(Stream.take(2), Stream.runCollect);
227639
228 expect(Option.getOrThrow(refs)).toEqual(LIVE_REFS);
640 expect(refs).toEqual([CACHED_REFS, LIVE_REFS]);
229641 }),
230642 ),
231643 );
packages/client-runtime/src/state/vcs.ts+170-56
......@@ -6,12 +6,14 @@ import {
66 WS_METHODS,
77} from "@t3tools/contracts";
88import { applyGitStatusStreamEvent } from "@t3tools/shared/git";
9import * as Duration from "effect/Duration";
910import * as Effect from "effect/Effect";
1011import * as Option from "effect/Option";
1112import * as Result from "effect/Result";
13import * as Schedule from "effect/Schedule";
1214import * as Stream from "effect/Stream";
1315import * as SubscriptionRef from "effect/SubscriptionRef";
14import { Atom } from "effect/unstable/reactivity";
16import { Atom, AtomRegistry } from "effect/unstable/reactivity";
1517
1618import { createEnvironmentRpcCommand, createEnvironmentSubscriptionAtomFamily } from "./runtime.ts";
1719import type { EnvironmentRegistry } from "../connection/registry.ts";
......@@ -21,9 +23,19 @@ import { EnvironmentCacheStore } from "../platform/persistence.ts";
2123import { request, subscribe, type EnvironmentRpcInput } from "../rpc/client.ts";
2224import { followStreamInEnvironment } from "./runtime.ts";
2325import { vcsCommandConcurrency, vcsCommandScheduler } from "./vcsCommandScheduler.ts";
26import {
27 invalidateCachedVcsRefs,
28 vcsRefsCacheStateAtom,
29 withVcsRefsPersistenceLock,
30} from "./vcsRefInvalidation.ts";
2431
2532const OFFLINE_BRANCH_LIST_LIMIT = 100;
26const VCS_REFS_REVALIDATE_INTERVAL = "5 seconds";
33const VCS_REFS_IDLE_TTL_MS = 30_000;
34const VCS_REFS_RETRY_SCHEDULE = Schedule.exponential("1 second").pipe(
35 Schedule.modifyDelay(({ duration }) =>
36 Effect.succeed(Duration.min(duration, Duration.seconds(30))),
37 ),
38);
2739
2840function canUseVcsRefsCache(input: VcsListRefsInput): boolean {
2941 return (
......@@ -35,6 +47,66 @@ function canUseVcsRefsCache(input: VcsListRefsInput): boolean {
3547 );
3648}
3749
50export const commitVcsRefsRefresh = Effect.fn("CachedVcsRefsState.commitRefresh")(function* (
51 registry: AtomRegistry.AtomRegistry,
52 cache: EnvironmentCacheStore["Service"],
53 input: {
54 readonly environmentId: EnvironmentId;
55 readonly cwd: string;
56 readonly refs: VcsListRefsResult;
57 readonly expectedRevision: number;
58 readonly persist: boolean;
59 },
60) {
61 return yield* withVcsRefsPersistenceLock(
62 input.environmentId,
63 Effect.gen(function* () {
64 const stateAtom = vcsRefsCacheStateAtom({ environmentId: input.environmentId });
65 const state = registry.get(stateAtom);
66 if (state.revision !== input.expectedRevision) {
67 return false;
68 }
69 let persistedCacheReadable = state.persistedCacheReadable;
70 if (input.persist) {
71 if (!persistedCacheReadable) {
72 persistedCacheReadable = yield* cache.clearVcsRefs(input.environmentId).pipe(
73 Effect.as(true),
74 Effect.catch((error) =>
75 Effect.logWarning("Could not recover invalidated cached Git refs.").pipe(
76 Effect.annotateLogs({
77 environmentId: input.environmentId,
78 cwd: input.cwd,
79 ...safeErrorLogAttributes(error),
80 }),
81 Effect.as(false),
82 ),
83 ),
84 );
85 }
86 yield* cache.saveVcsRefs(input.environmentId, input.cwd, input.refs).pipe(
87 Effect.catch((error) =>
88 Effect.logWarning("Could not persist cached Git refs.").pipe(
89 Effect.annotateLogs({
90 environmentId: input.environmentId,
91 cwd: input.cwd,
92 ...safeErrorLogAttributes(error),
93 }),
94 ),
95 ),
96 );
97 if (persistedCacheReadable !== state.persistedCacheReadable) {
98 registry.update(stateAtom, (current) =>
99 current.revision === input.expectedRevision
100 ? { ...current, persistedCacheReadable }
101 : current,
102 );
103 }
104 }
105 return true;
106 }),
107 );
108});
109
38110/**
39111 * Retains the last unfiltered branch-list response for the new-task picker.
40112 * Filtered or paginated lists intentionally stay live-only: treating a
......@@ -43,54 +115,59 @@ function canUseVcsRefsCache(input: VcsListRefsInput): boolean {
43115 */
44116export const makeCachedVcsRefsChanges = Effect.fn("CachedVcsRefsState.makeChanges")(function* (
45117 input: VcsListRefsInput,
118 expectedRevision?: number,
119 registry?: AtomRegistry.AtomRegistry,
120 persistedCacheReadable = true,
46121) {
47122 const supervisor = yield* EnvironmentSupervisor;
48123 const cache = yield* EnvironmentCacheStore;
49124 const environmentId = supervisor.target.environmentId;
50125 const useCache = canUseVcsRefsCache(input);
51 const cached = useCache
52 ? yield* cache.loadVcsRefs(environmentId, input.cwd).pipe(
53 Effect.catch((error) =>
54 Effect.logWarning("Could not load cached Git refs.").pipe(
55 Effect.annotateLogs({
56 environmentId,
57 cwd: input.cwd,
58 ...safeErrorLogAttributes(error),
59 }),
60 Effect.as(Option.none<VcsListRefsResult>()),
126 const cached =
127 useCache && persistedCacheReadable
128 ? yield* cache.loadVcsRefs(environmentId, input.cwd).pipe(
129 Effect.catch((error) =>
130 Effect.logWarning("Could not load cached Git refs.").pipe(
131 Effect.annotateLogs({
132 environmentId,
133 cwd: input.cwd,
134 ...safeErrorLogAttributes(error),
135 }),
136 Effect.as(Option.none<VcsListRefsResult>()),
137 ),
61138 ),
62 ),
63 )
64 : Option.none<VcsListRefsResult>();
139 )
140 : Option.none<VcsListRefsResult>();
65141 const refresh = Effect.fn("CachedVcsRefsState.refresh")(function* () {
66142 const refs = yield* request(WS_METHODS.vcsListRefs, input).pipe(
67143 Effect.provideService(EnvironmentSupervisor, supervisor),
68144 );
69 if (useCache) {
70 yield* cache.saveVcsRefs(environmentId, input.cwd, refs).pipe(
71 Effect.catch((error) =>
72 Effect.logWarning("Could not persist cached Git refs.").pipe(
73 Effect.annotateLogs({
74 environmentId,
75 cwd: input.cwd,
76 ...safeErrorLogAttributes(error),
77 }),
78 ),
145 const persist = cache.saveVcsRefs(environmentId, input.cwd, refs).pipe(
146 Effect.catch((error) =>
147 Effect.logWarning("Could not persist cached Git refs.").pipe(
148 Effect.annotateLogs({
149 environmentId,
150 cwd: input.cwd,
151 ...safeErrorLogAttributes(error),
152 }),
79153 ),
80 );
154 ),
155 );
156 if (expectedRevision === undefined || registry === undefined) {
157 if (useCache) yield* persist;
158 return Option.some(refs);
81159 }
82 return refs;
160 const committed = yield* commitVcsRefsRefresh(registry, cache, {
161 environmentId,
162 cwd: input.cwd,
163 refs,
164 expectedRevision,
165 persist: useCache,
166 });
167 return committed ? Option.some(refs) : Option.none<VcsListRefsResult>();
83168 });
84169
85 const cachedRefs = Stream.fromEffect(
86 SubscriptionRef.get(supervisor.state).pipe(
87 Effect.flatMap((connection) =>
88 connection.phase === "connected"
89 ? Effect.succeed(Option.none<VcsListRefsResult>())
90 : Effect.succeed(cached),
91 ),
92 ),
93 ).pipe(
170 const cachedRefs = Stream.fromEffect(Effect.succeed(cached)).pipe(
94171 Stream.filterMap((refs) =>
95172 Option.match(refs, {
96173 onNone: () => Result.failVoid,
......@@ -107,24 +184,20 @@ export const makeCachedVcsRefsChanges = Effect.fn("CachedVcsRefsState.makeChange
107184 Stream.switchMap((generation) =>
108185 generation === null
109186 ? Stream.empty
110 : Stream.tick(VCS_REFS_REVALIDATE_INTERVAL).pipe(
111 Stream.mapEffect(
112 () =>
113 refresh().pipe(
114 Effect.map(Option.some),
115 Effect.catch((error) =>
116 Effect.logWarning("Could not refresh Git refs.").pipe(
117 Effect.annotateLogs({
118 environmentId,
119 cwd: input.cwd,
120 ...safeErrorLogAttributes(error),
121 }),
122 Effect.as(Option.none<VcsListRefsResult>()),
123 ),
124 ),
187 : Stream.fromEffect(
188 refresh().pipe(
189 Effect.tapError((error) =>
190 Effect.logWarning("Could not refresh Git refs.").pipe(
191 Effect.annotateLogs({
192 environmentId,
193 cwd: input.cwd,
194 ...safeErrorLogAttributes(error),
195 }),
125196 ),
126 { concurrency: 1 },
197 ),
127198 ),
199 ).pipe(
200 Stream.retry(VCS_REFS_RETRY_SCHEDULE),
128201 Stream.filterMap((refs) =>
129202 Option.match(refs, {
130203 onNone: () => Result.failVoid,
......@@ -138,8 +211,26 @@ export const makeCachedVcsRefsChanges = Effect.fn("CachedVcsRefsState.makeChange
138211 return Stream.concat(cachedRefs, refreshedRefs);
139212});
140213
141export function cachedVcsRefsChanges(environmentId: EnvironmentId, input: VcsListRefsInput) {
142 return followStreamInEnvironment(environmentId, Stream.unwrap(makeCachedVcsRefsChanges(input)));
214export function cachedVcsRefsChanges(
215 environmentId: EnvironmentId,
216 input: VcsListRefsInput,
217 expectedRevision: number,
218 persistedCacheReadable: boolean,
219) {
220 return followStreamInEnvironment(
221 environmentId,
222 Stream.unwrap(
223 Effect.gen(function* () {
224 const registry = yield* AtomRegistry.AtomRegistry;
225 return yield* makeCachedVcsRefsChanges(
226 input,
227 expectedRevision,
228 registry,
229 persistedCacheReadable,
230 );
231 }),
232 ),
233 );
143234}
144235
145236export function createVcsEnvironmentAtoms<R, E>(
......@@ -149,9 +240,17 @@ export function createVcsEnvironmentAtoms<R, E>(
149240 Atom.family((inputKey: string) => {
150241 const input = JSON.parse(inputKey) as VcsListRefsInput;
151242 return runtime
152 .atom(cachedVcsRefsChanges(environmentId, input))
243 .atom((get) => {
244 const state = get(vcsRefsCacheStateAtom({ environmentId }));
245 return cachedVcsRefsChanges(
246 environmentId,
247 input,
248 state.revision,
249 state.persistedCacheReadable,
250 );
251 })
153252 .pipe(
154 Atom.setIdleTTL(5 * 60_000),
253 Atom.setIdleTTL(VCS_REFS_IDLE_TTL_MS),
155254 Atom.withLabel(`environment-data:vcs:list-refs:${environmentId}:${inputKey}`),
156255 );
157256 }),
......@@ -160,6 +259,14 @@ export function createVcsEnvironmentAtoms<R, E>(
160259 readonly environmentId: EnvironmentId;
161260 readonly input: VcsListRefsInput;
162261 }) => listRefsByEnvironment(target.environmentId)(JSON.stringify(target.input));
262 const invalidateRefs = (
263 target: { readonly environmentId: EnvironmentId; readonly input: { readonly cwd: string } },
264 registry: AtomRegistry.AtomRegistry,
265 ) =>
266 invalidateCachedVcsRefs(registry, {
267 environmentId: target.environmentId,
268 cwd: target.input.cwd,
269 });
163270
164271 return {
165272 listRefs,
......@@ -181,42 +288,49 @@ export function createVcsEnvironmentAtoms<R, E>(
181288 tag: WS_METHODS.vcsPull,
182289 scheduler: vcsCommandScheduler,
183290 concurrency: vcsCommandConcurrency,
291 onSettled: invalidateRefs,
184292 }),
185293 refreshStatus: createEnvironmentRpcCommand(runtime, {
186294 label: "environment-data:vcs:refresh-status",
187295 tag: WS_METHODS.vcsRefreshStatus,
188296 scheduler: vcsCommandScheduler,
189297 concurrency: vcsCommandConcurrency,
298 onSettled: invalidateRefs,
190299 }),
191300 createWorktree: createEnvironmentRpcCommand(runtime, {
192301 label: "environment-data:vcs:create-worktree",
193302 tag: WS_METHODS.vcsCreateWorktree,
194303 scheduler: vcsCommandScheduler,
195304 concurrency: vcsCommandConcurrency,
305 onSettled: invalidateRefs,
196306 }),
197307 removeWorktree: createEnvironmentRpcCommand(runtime, {
198308 label: "environment-data:vcs:remove-worktree",
199309 tag: WS_METHODS.vcsRemoveWorktree,
200310 scheduler: vcsCommandScheduler,
201311 concurrency: vcsCommandConcurrency,
312 onSettled: invalidateRefs,
202313 }),
203314 createRef: createEnvironmentRpcCommand(runtime, {
204315 label: "environment-data:vcs:create-ref",
205316 tag: WS_METHODS.vcsCreateRef,
206317 scheduler: vcsCommandScheduler,
207318 concurrency: vcsCommandConcurrency,
319 onSettled: invalidateRefs,
208320 }),
209321 switchRef: createEnvironmentRpcCommand(runtime, {
210322 label: "environment-data:vcs:switch-ref",
211323 tag: WS_METHODS.vcsSwitchRef,
212324 scheduler: vcsCommandScheduler,
213325 concurrency: vcsCommandConcurrency,
326 onSettled: invalidateRefs,
214327 }),
215328 init: createEnvironmentRpcCommand(runtime, {
216329 label: "environment-data:vcs:init",
217330 tag: WS_METHODS.vcsInit,
218331 scheduler: vcsCommandScheduler,
219332 concurrency: vcsCommandConcurrency,
333 onSettled: invalidateRefs,
220334 }),
221335 };
222336}
packages/client-runtime/src/state/vcsAction.test.ts+159-4
......@@ -1,5 +1,6 @@
11import {
22 EnvironmentId,
3 WS_METHODS,
34 type GitActionProgressEvent,
45 type GitRunStackedActionResult,
56} from "@t3tools/contracts";
......@@ -7,11 +8,23 @@ import { describe, expect, it } from "@effect/vitest";
78import * as Cause from "effect/Cause";
89import * as Effect from "effect/Effect";
910import * as Layer from "effect/Layer";
11import * as Option from "effect/Option";
1012import * as Schema from "effect/Schema";
1113import * as Stream from "effect/Stream";
14import * as SubscriptionRef from "effect/SubscriptionRef";
1215import { AsyncResult, Atom, AtomRegistry } from "effect/unstable/reactivity";
1316
14import type { EnvironmentRegistry } from "../connection/registry.ts";
17import {
18 AVAILABLE_CONNECTION_STATE,
19 PrimaryConnectionTarget,
20 type PreparedConnection,
21 type SupervisorConnectionState,
22} from "../connection/model.ts";
23import * as EnvironmentRegistry from "../connection/registry.ts";
24import * as EnvironmentSupervisor from "../connection/supervisor.ts";
25import * as Persistence from "../platform/persistence.ts";
26import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
27import type { RpcSession } from "../rpc/session.ts";
1528import type { AtomCommandResult } from "./runtime.ts";
1629import {
1730 applyVcsActionProgressEvent,
......@@ -28,6 +41,7 @@ import {
2841 VcsActionTargetKeyParseError,
2942 VcsActionUnavailableError,
3043} from "./vcsAction.ts";
44import { vcsRefsCacheStateAtom } from "./vcsRefInvalidation.ts";
3145
3246const actionId = "action-123";
3347const action = "commit_push" as const;
......@@ -59,10 +73,44 @@ const result: GitRunStackedActionResult = {
5973 },
6074};
6175
76const target = new PrimaryConnectionTarget({
77 environmentId,
78 label: "Test environment",
79 httpBaseUrl: "https://environment.example.test",
80 wsBaseUrl: "wss://environment.example.test",
81});
82
83function session(client: WsRpcProtocolClient): RpcSession {
84 return {
85 client,
86 initialConfig: Effect.never,
87 ready: Effect.void,
88 probe: Effect.void,
89 closed: Effect.never,
90 };
91}
92
6293function progress<T extends GitActionProgressEvent>(event: T): T {
6394 return event;
6495}
6596
97function cacheStore(onClearVcsRefs: (environmentId: EnvironmentId) => void) {
98 return Persistence.EnvironmentCacheStore.of({
99 loadShell: () => Effect.succeed(Option.none()),
100 saveShell: () => Effect.void,
101 loadThread: () => Effect.succeed(Option.none()),
102 saveThread: () => Effect.void,
103 removeThread: () => Effect.void,
104 loadServerConfig: () => Effect.succeed(Option.none()),
105 saveServerConfig: () => Effect.void,
106 loadVcsRefs: () => Effect.succeed(Option.none()),
107 saveVcsRefs: () => Effect.void,
108 removeVcsRefs: () => Effect.void,
109 clearVcsRefs: (environmentId) => Effect.sync(() => onClearVcsRefs(environmentId)),
110 clear: () => Effect.void,
111 });
112}
113
66114describe("vcsActionState", () => {
67115 it("preserves malformed target key diagnostics and the native cause without copying the key", () => {
68116 const key = "not-json-with-credential=do-not-log";
......@@ -417,7 +465,7 @@ describe("vcsActionState", () => {
417465
418466 it("keys mutation ownership by environment and cwd", () => {
419467 const runtime = Atom.runtime(Layer.empty) as unknown as Atom.AtomRuntime<
420 EnvironmentRegistry,
468 EnvironmentRegistry.EnvironmentRegistry | Persistence.EnvironmentCacheStore,
421469 never
422470 >;
423471 const manager = createVcsActionManager(runtime);
......@@ -437,7 +485,7 @@ describe("vcsActionState", () => {
437485
438486 it("retains the incomplete target and operation when tracking is unavailable", async () => {
439487 const runtime = Atom.runtime(Layer.empty) as unknown as Atom.AtomRuntime<
440 EnvironmentRegistry,
488 EnvironmentRegistry.EnvironmentRegistry | Persistence.EnvironmentCacheStore,
441489 never
442490 >;
443491 const manager = createVcsActionManager(runtime);
......@@ -469,7 +517,7 @@ describe("vcsActionState", () => {
469517
470518 it("tracks finite mutations without letting an older completion clear newer state", async () => {
471519 const runtime = Atom.runtime(Layer.empty) as unknown as Atom.AtomRuntime<
472 EnvironmentRegistry,
520 EnvironmentRegistry.EnvironmentRegistry | Persistence.EnvironmentCacheStore,
473521 never
474522 >;
475523 const manager = createVcsActionManager(runtime);
......@@ -520,4 +568,111 @@ describe("vcsActionState", () => {
520568
521569 registry.dispose();
522570 });
571
572 it.effect("invalidates persisted refs after successful and failed stacked actions", () =>
573 Effect.scoped(
574 Effect.gen(function* () {
575 const connectionState: SupervisorConnectionState = {
576 ...AVAILABLE_CONNECTION_STATE,
577 desired: true,
578 network: "online",
579 phase: "connected",
580 attempt: 1,
581 generation: 1,
582 };
583 const targetKey = { environmentId, cwd };
584 const successfulActionId = "invalidate-success";
585 const failedActionId = "invalidate-failure";
586 const successfulTransportActionId = createVcsActionTransportId(
587 targetKey,
588 successfulActionId,
589 );
590 const failedTransportActionId = createVcsActionTransportId(targetKey, failedActionId);
591 const client = {
592 [WS_METHODS.gitRunStackedAction]: (input: { readonly actionId: string }) =>
593 input.actionId === successfulTransportActionId
594 ? Stream.make(
595 progress({
596 kind: "action_finished",
597 actionId: successfulTransportActionId,
598 cwd,
599 action,
600 result,
601 }),
602 )
603 : Stream.make(
604 progress({
605 kind: "action_failed",
606 actionId: failedTransportActionId,
607 cwd,
608 action,
609 phase: "push",
610 message: "push failed after creating the branch",
611 }),
612 ),
613 } as unknown as WsRpcProtocolClient;
614 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
615 target,
616 state: yield* SubscriptionRef.make(connectionState),
617 session: yield* SubscriptionRef.make(Option.some(session(client))),
618 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
619 connect: Effect.void,
620 disconnect: Effect.void,
621 retryNow: Effect.void,
622 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
623 const run: EnvironmentRegistry.EnvironmentRegistry["Service"]["run"] = (
624 _environmentId,
625 effect,
626 ) => Effect.provideService(effect, EnvironmentSupervisor.EnvironmentSupervisor, supervisor);
627 const runStream: EnvironmentRegistry.EnvironmentRegistry["Service"]["runStream"] = (
628 _environmentId,
629 stream,
630 ) => Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor);
631 const environmentRegistry = EnvironmentRegistry.EnvironmentRegistry.of({
632 run,
633 runStream,
634 } as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]);
635 const removed = new Array<string>();
636 const runtime = Atom.runtime(
637 Layer.merge(
638 Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
639 Layer.succeed(
640 Persistence.EnvironmentCacheStore,
641 cacheStore((removedEnvironmentId) => {
642 removed.push(`${removedEnvironmentId}:*`);
643 }),
644 ),
645 ),
646 );
647 const manager = createVcsActionManager(runtime);
648 const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
649 Effect.sync(() => registry.dispose()),
650 );
651 const state = vcsRefsCacheStateAtom({ environmentId });
652
653 expect(registry.get(state).revision).toBe(0);
654 const successfulResult = yield* Effect.promise(() =>
655 manager.runStackedAction(targetKey).run(registry, {
656 actionId: successfulActionId,
657 action,
658 }),
659 );
660
661 expect(AsyncResult.isSuccess(successfulResult)).toBe(true);
662 expect(registry.get(state).revision).toBe(1);
663 expect(removed).toEqual([`${environmentId}:*`]);
664
665 const failedResult = yield* Effect.promise(() =>
666 manager.runStackedAction(targetKey).run(registry, {
667 actionId: failedActionId,
668 action,
669 }),
670 );
671
672 expect(AsyncResult.isFailure(failedResult)).toBe(true);
673 expect(registry.get(state).revision).toBe(2);
674 expect(removed).toEqual([`${environmentId}:*`, `${environmentId}:*`]);
675 }),
676 ),
677 );
523678});
packages/client-runtime/src/state/vcsAction.ts+5-2
......@@ -16,6 +16,7 @@ import * as Stream from "effect/Stream";
1616import { AsyncResult, Atom, type AtomRegistry } from "effect/unstable/reactivity";
1717
1818import type { EnvironmentRegistry } from "../connection/registry.ts";
19import { EnvironmentCacheStore } from "../platform/persistence.ts";
1920import { runStream } from "../rpc/client.ts";
2021import {
2122 createRuntimeCommand,
......@@ -24,6 +25,7 @@ import {
2425 type AtomCommandResult,
2526} from "./runtime.ts";
2627import { vcsCommandScheduler } from "./vcsCommandScheduler.ts";
28import { invalidateCachedVcsRefs } from "./vcsRefInvalidation.ts";
2729
2830export const VcsActionOperation = Schema.Literals([
2931 "refresh_status",
......@@ -403,7 +405,7 @@ export function applyVcsActionProgressEvent(
403405}
404406
405407export function createVcsActionManager<R, E>(
406 runtime: Atom.AtomRuntime<EnvironmentRegistry | R, E>,
408 runtime: Atom.AtomRuntime<EnvironmentRegistry | EnvironmentCacheStore | R, E>,
407409) {
408410 const runStackedActionCommands = new Map<
409411 string,
......@@ -425,7 +427,7 @@ export function createVcsActionManager<R, E>(
425427 const target = targetKey === null ? null : parseVcsActionTargetKey(targetKey);
426428 const stateAtom = targetKey === null ? EMPTY_VCS_ACTION_ATOM : vcsActionStateAtom(targetKey);
427429 const command = createRuntimeCommand<
428 EnvironmentRegistry | R,
430 EnvironmentRegistry | EnvironmentCacheStore | R,
429431 E,
430432 RunVcsStackedActionInput,
431433 GitRunStackedActionResult,
......@@ -489,6 +491,7 @@ export function createVcsActionManager<R, E>(
489491 }),
490492 },
491493 ).pipe(
494 Effect.ensuring(invalidateCachedVcsRefs(registry, target)),
492495 Effect.tapError((error) =>
493496 Effect.sync(() => {
494497 const current = registry.get(stateAtom);
packages/client-runtime/src/state/vcsRefInvalidation.ts created+83
......@@ -0,0 +1,83 @@
1import type { EnvironmentId } from "@t3tools/contracts";
2import * as Effect from "effect/Effect";
3import * as PartitionedSemaphore from "effect/PartitionedSemaphore";
4import { Atom, type AtomRegistry } from "effect/unstable/reactivity";
5
6import { safeErrorLogAttributes } from "../errors/safeLog.ts";
7import { EnvironmentCacheStore } from "../platform/persistence.ts";
8
9export interface VcsRefsInvalidationTarget {
10 readonly environmentId: EnvironmentId;
11}
12
13export interface CachedVcsRefsInvalidationTarget extends VcsRefsInvalidationTarget {
14 readonly cwd: string;
15}
16
17export interface VcsRefsCacheState {
18 readonly revision: number;
19 readonly persistedCacheReadable: boolean;
20}
21
22const stateByEnvironment = Atom.family((environmentId: EnvironmentId) =>
23 Atom.make<VcsRefsCacheState>({
24 revision: 0,
25 persistedCacheReadable: true,
26 }).pipe(Atom.keepAlive, Atom.withLabel(`environment-data:vcs:list-refs-state:${environmentId}`)),
27);
28const persistenceLock = PartitionedSemaphore.makeUnsafe<EnvironmentId>({ permits: 1 });
29
30export function vcsRefsCacheStateAtom(target: VcsRefsInvalidationTarget) {
31 return stateByEnvironment(target.environmentId);
32}
33
34export function invalidateVcsRefs(
35 registry: AtomRegistry.AtomRegistry,
36 target: VcsRefsInvalidationTarget,
37 persistedCacheReadable?: boolean,
38): void {
39 registry.update(vcsRefsCacheStateAtom(target), (state) => ({
40 revision: state.revision + 1,
41 persistedCacheReadable: persistedCacheReadable ?? state.persistedCacheReadable,
42 }));
43}
44
45export function withVcsRefsPersistenceLock<A, E, R>(
46 environmentId: EnvironmentId,
47 effect: Effect.Effect<A, E, R>,
48): Effect.Effect<A, E, R> {
49 return persistenceLock.withPermit(environmentId)(effect);
50}
51
52/**
53 * Clears persisted snapshots and then restarts live ref streams while holding
54 * the same lock used by generation-checked refresh writes. Clearing first
55 * prevents the restarted stream from rehydrating the invalidated snapshot. A
56 * refresh that wins the lock is cleared afterward; one that loses observes the
57 * new revision and cannot repersist its pre-mutation result.
58 */
59export const invalidateCachedVcsRefs = Effect.fn("VcsRefsState.invalidateCached")(function* (
60 registry: AtomRegistry.AtomRegistry,
61 target: CachedVcsRefsInvalidationTarget,
62) {
63 const cache = yield* EnvironmentCacheStore;
64 yield* withVcsRefsPersistenceLock(
65 target.environmentId,
66 Effect.gen(function* () {
67 const persistedCacheReadable = yield* cache.clearVcsRefs(target.environmentId).pipe(
68 Effect.as(true),
69 Effect.catch((error) =>
70 Effect.logWarning("Could not remove invalidated cached Git refs.").pipe(
71 Effect.annotateLogs({
72 environmentId: target.environmentId,
73 cwd: target.cwd,
74 ...safeErrorLogAttributes(error),
75 }),
76 Effect.as(false),
77 ),
78 ),
79 );
80 invalidateVcsRefs(registry, target, persistedCacheReadable);
81 }),
82 );
83});
packages/contracts/src/git.ts+1
......@@ -127,6 +127,7 @@ export const VcsListRefsInput = Schema.Struct({
127127 cursor: Schema.optional(NonNegativeInt),
128128 includeMatchingRemoteRefs: Schema.optional(Schema.Boolean),
129129 refKind: Schema.optional(Schema.Literals(["all", "local", "remote"])),
130 refresh: Schema.optional(Schema.Boolean),
130131 limit: Schema.optional(
131132 PositiveInt.check(Schema.isLessThanOrEqualTo(GIT_LIST_BRANCHES_MAX_LIMIT)),
132133 ),