Commit a5efb7cb6b9e80b9543bc3e0aabfea37d3b459c5 | the-knife

author	Julius Marminge <julius0216@outlook.com> 2026-06-16 10:54:00-07:00
committer	GitHub <noreply@github.com> 2026-06-16 10:54:00-07:00
log	a5efb7cb6b9e80b9543bc3e0aabfea37d3b459c5
tree	dea81b01ef04943b45317f5dc3600a4dbb452db9
parent	8f9911f79dcfa6890c6f14f9c712c236e8453cea
signature	Signed by PGP key B5690EEEBB952194

Parallelize VCS status refresh and status reads (#3112)

3 files changed, 140 insertions(+), 30 deletions(-)

apps
- server
  - src
    git
    GitManager.ts
    vcs
    VcsStatusBroadcaster.test.ts
    VcsStatusBroadcaster.ts

apps/server/src/git/GitManager.ts+3-1

 @@ -1361,7 +1361,9 @@ export const makeGitManager = Effect.fn("makeGitManager")(function* () {
     },
   );
   const status: GitManagerShape["status"] = Effect.fn("status")(function* (input) {
-    const [local, remote] = yield* Effect.all([localStatus(input), remoteStatus(input)]);
+    const [local, remote] = yield* Effect.all([localStatus(input), remoteStatus(input)], {
+      concurrency: "unbounded",
+    });
     return mergeGitStatusParts(local, remote);
   });
   const invalidateLocalStatus: GitManagerShape["invalidateLocalStatus"] = Effect.fn(

apps/server/src/vcs/VcsStatusBroadcaster.test.ts+66

...	@@ -17,6 +17,7 @@ import type {	...	@@ -17,6 +17,7 @@ import type {
17	VcsStatusResult,	17	VcsStatusResult,
18	VcsStatusStreamEvent,	18	VcsStatusStreamEvent,
19	} from "@t3tools/contracts";	19	} from "@t3tools/contracts";
		20	import { GitManagerError } from "@t3tools/contracts";
20		21
21	import * as VcsStatusBroadcaster from "./VcsStatusBroadcaster.ts";	22	import * as VcsStatusBroadcaster from "./VcsStatusBroadcaster.ts";
22	import * as GitWorkflowService from "../git/GitWorkflowService.ts";	23	import * as GitWorkflowService from "../git/GitWorkflowService.ts";
...	@@ -149,6 +150,71 @@ describe("VcsStatusBroadcaster", () => {	...	@@ -149,6 +150,71 @@ describe("VcsStatusBroadcaster", () => {
149	}).pipe(Effect.provide(makeTestLayer(state)));	150	}).pipe(Effect.provide(makeTestLayer(state)));
150	});	151	});
151		152
		153	it.effect("keeps the cached snapshot unchanged when a refresh branch fails", () => {
		154	const state = {
		155	currentLocalStatus: baseLocalStatus,
		156	currentRemoteStatus: baseRemoteStatus,
		157	localStatusCalls: 0,
		158	remoteStatusCalls: 0,
		159	localInvalidationCalls: 0,
		160	remoteInvalidationCalls: 0,
		161	failRemoteStatus: false,
		162	};
		163	const testLayer = VcsStatusBroadcaster.layer.pipe(
		164	Layer.provideMerge(NodeServices.layer),
		165	Layer.provide(
		166	Layer.mock(GitWorkflowService.GitWorkflowService)({
		167	localStatus: () =>
		168	Effect.sync(() => {
		169	state.localStatusCalls += 1;
		170	return state.currentLocalStatus;
		171	}),
		172	remoteStatus: () =>
		173	Effect.suspend(() => {
		174	state.remoteStatusCalls += 1;
		175	return state.failRemoteStatus
		176	? Effect.fail(
		177	new GitManagerError({
		178	operation: "VcsStatusBroadcaster.test",
		179	detail: "remote status failed",
		180	}),
		181	)
		182	: Effect.succeed(state.currentRemoteStatus);
		183	}),
		184	invalidateLocalStatus: () =>
		185	Effect.sync(() => {
		186	state.localInvalidationCalls += 1;
		187	}),
		188	invalidateRemoteStatus: () =>
		189	Effect.sync(() => {
		190	state.remoteInvalidationCalls += 1;
		191	}),
		192	}),
		193	),
		194	);
		195
		196	return Effect.gen(function* () {
		197	const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;
		198	yield* broadcaster.getStatus({ cwd: "/repo" });
		199
		200	state.currentLocalStatus = {
		201	...baseLocalStatus,
		202	refName: "feature/partial-refresh",
		203	};
		204	state.currentRemoteStatus = {
		205	...baseRemoteStatus,
		206	aheadCount: 3,
		207	};
		208	state.failRemoteStatus = true;
		209
		210	const refreshExit = yield* broadcaster.refreshStatus("/repo").pipe(Effect.exit);
		211	const cached = yield* broadcaster.getStatus({ cwd: "/repo" });
		212
		213	assert.isTrue(Exit.isFailure(refreshExit));
		214	assert.deepStrictEqual(cached, baseStatus);
		215	}).pipe(Effect.provide(testLayer));
		216	});
		217
152	it.effect("refreshes only the cached local snapshot when requested", () => {	218	it.effect("refreshes only the cached local snapshot when requested", () => {
153	const state = {	219	const state = {
154	currentLocalStatus: baseLocalStatus,	220	currentLocalStatus: baseLocalStatus,

apps/server/src/vcs/VcsStatusBroadcaster.ts+71-29

 @@ -179,18 +179,53 @@ export const layer = Layer.effect(
       },
     );
-    const loadLocalStatus = Effect.fn("VcsStatusBroadcaster.loadLocalStatus")(function* (
+    const updateCachedStatus = Effect.fn("VcsStatusBroadcaster.updateCachedStatus")(function* (
       cwd: string,
+      local: VcsStatusLocalResult,
+      remote: VcsStatusRemoteResult | null,
+      options?: { publish?: boolean },
     ) {
-      const local = yield* workflow.localStatus({ cwd });
+      const nextLocal = {
-      return yield* updateCachedLocalStatus(cwd, local);
+        fingerprint: fingerprintStatusPart(local),
+        value: local,
+      } satisfies CachedValue<VcsStatusLocalResult>;
+      const nextRemote = {
+        fingerprint: fingerprintStatusPart(remote),
+        value: remote,
+      } satisfies CachedValue<VcsStatusRemoteResult | null>;
+      const shouldPublish = yield* Ref.modify(cacheRef, (cache) => {
+        const previous = cache.get(cwd) ?? { local: null, remote: null };
+        const nextCache = new Map(cache);
+        nextCache.set(cwd, {
+          local: nextLocal,
+          remote: nextRemote,
+        });
+        return [
+          previous.local?.fingerprint !== nextLocal.fingerprint ||
+            previous.remote?.fingerprint !== nextRemote.fingerprint,
+          nextCache,
+        ] as const;
+      });
+      if (options?.publish && shouldPublish) {
+        yield* PubSub.publish(changesPubSub, {
+          cwd,
+          event: {
+            _tag: "snapshot",
+            local,
+            remote,
+          },
+        });
+      }
+      return mergeGitStatusParts(local, remote);
     });
-    const loadRemoteStatus = Effect.fn("VcsStatusBroadcaster.loadRemoteStatus")(function* (
+    const loadLocalStatus = Effect.fn("VcsStatusBroadcaster.loadLocalStatus")(function* (
       cwd: string,
     ) {
-      const remote = yield* workflow.remoteStatus({ cwd });
+      const local = yield* workflow.localStatus({ cwd });
-      return yield* updateCachedRemoteStatus(cwd, remote);
+      return yield* updateCachedLocalStatus(cwd, local);
     });
     const getOrLoadLocalStatus = Effect.fn("VcsStatusBroadcaster.getOrLoadLocalStatus")(function* (
 @@ -203,36 +238,39 @@ export const layer = Layer.effect(
       return yield* loadLocalStatus(cwd);
     });
-    const getOrLoadRemoteStatus = Effect.fn("VcsStatusBroadcaster.getOrLoadRemoteStatus")(
-      function* (cwd: string) {
-        const cached = yield* getCachedStatus(cwd);
-        if (cached?.remote) {
-          return cached.remote.value;
-        }
-        return yield* loadRemoteStatus(cwd);
-      },
-    );
     const withFileSystem = Effect.provideService(FileSystem.FileSystem, fs);
     const getStatus: VcsStatusBroadcasterShape["getStatus"] = Effect.fn(
       "VcsStatusBroadcaster.getStatus",
     )(function* (input) {
       const cwd = yield* withFileSystem(normalizeCwd(input.cwd));
-      const [local, remote] = yield* Effect.all([
+      const cached = yield* getCachedStatus(cwd);
-        getOrLoadLocalStatus(cwd),
+      if (cached?.local && cached.remote) {
-        getOrLoadRemoteStatus(cwd),
+        return mergeGitStatusParts(cached.local.value, cached.remote.value);
-      ]);
+      }
-      return mergeGitStatusParts(local, remote);
+      const [local, remote] = yield* Effect.all(
+        [
+          cached?.local ? Effect.succeed(cached.local.value) : workflow.localStatus({ cwd }),
+          cached?.remote ? Effect.succeed(cached.remote.value) : workflow.remoteStatus({ cwd }),
+        ],
+        { concurrency: "unbounded" },
+      );
+      return yield* updateCachedStatus(cwd, local, remote);
     });
+    const refreshLocalStatusCore = Effect.fn("VcsStatusBroadcaster.refreshLocalStatusCore")(
+      function* (cwd: string) {
+        yield* workflow.invalidateLocalStatus(cwd);
+        const local = yield* workflow.localStatus({ cwd });
+        return yield* updateCachedLocalStatus(cwd, local, { publish: true });
+      },
+    );
     const refreshLocalStatus: VcsStatusBroadcasterShape["refreshLocalStatus"] = Effect.fn(
       "VcsStatusBroadcaster.refreshLocalStatus",
     )(function* (rawCwd) {
       const cwd = yield* withFileSystem(normalizeCwd(rawCwd));
-      yield* workflow.invalidateLocalStatus(cwd);
+      return yield* refreshLocalStatusCore(cwd);
-      const local = yield* workflow.localStatus({ cwd });
-      return yield* updateCachedLocalStatus(cwd, local, { publish: true });
     });
     const refreshRemoteStatus = Effect.fn("VcsStatusBroadcaster.refreshRemoteStatus")(function* (
 @@ -247,11 +285,15 @@ export const layer = Layer.effect(
       "VcsStatusBroadcaster.refreshStatus",
     )(function* (rawCwd) {
       const cwd = yield* withFileSystem(normalizeCwd(rawCwd));
-      const [local, remote] = yield* Effect.all([
+      yield* Effect.all(
-        refreshLocalStatus(cwd),
+        [workflow.invalidateLocalStatus(cwd), workflow.invalidateRemoteStatus(cwd)],
-        refreshRemoteStatus(cwd),
+        { concurrency: "unbounded", discard: true },
-      ]);
+      );
-      return mergeGitStatusParts(local, remote);
+      const [local, remote] = yield* Effect.all(
+        [workflow.localStatus({ cwd }), workflow.remoteStatus({ cwd })],
+        { concurrency: "unbounded" },
+      );
+      return yield* updateCachedStatus(cwd, local, remote, { publish: true });
     });
     const makeRemoteRefreshLoop = (