fix(server): release completed PR sync sweep data from the read cache (#25285)

Author: juliusmarminge

apps/server/src/pullRequest/PullRequestReadCache.test.ts

@@ -1,3 +1,5 @@
+import * as NodeV8 from "node:v8";
+import * as NodeVM from "node:vm";
 import { assert, it } from "@effect/vitest";
 import * as NodeServices from "@effect/platform-node/NodeServices";
 import { PullRequestOperationError } from "@t3tools/contracts";
@@ -8,6 +10,9 @@ import * as FileSystem from "effect/FileSystem";
 import * as KeyValueStore from "effect/persistence/KeyValueStore";
 import * as PullRequestReadCache from "./PullRequestReadCache.ts";
 
+NodeV8.setFlagsFromString("--expose-gc");
+const collectGarbage = NodeVM.runInNewContext("gc") as () => void;
+
 const cacheLayer = (directory: string) =>
   PullRequestReadCache.make.pipe(Effect.provide(KeyValueStore.layerFileSystem(directory)));
 
@@ -147,6 +152,32 @@ it.layer(NodeServices.layer)("PR filesystem cache", (it) => {
     }).pipe(Effect.provide(KeyValueStore.layerMemory)),
   );
 
+  it.effect("does not keep a caller's data alive after its reads and invalidations", () =>
+    Effect.gen(function* () {
+      const cache = yield* PullRequestReadCache.make;
+      const sweeps: Array<WeakRef<object>> = [];
+      // Like a sync sweep: an Effect.fn made per run that closes over that run's data.
+      const sweep = Effect.fn("sweep")(function* (index: number) {
+        const groups = new Map<string, number>(
+          Array.from({ length: 1_000 }, (_, id) => [`group-${id}`, id]),
+        );
+        sweeps.push(new WeakRef(groups));
+        const read = Effect.fn("read")(function* (pr: string) {
+          const size = groups.size;
+          yield* cache.invalidate(pr);
+          const answer = yield* cache.get(`${pr}-${index}`, Effect.succeed("answer"), [pr]);
+          return size + answer.length;
+        });
+        yield* Effect.forEach(["pr-1", "pr-2"], read, { discard: true });
+      });
+      for (let index = 0; index < 5; index++) yield* sweep(index);
+      // WeakRef targets stay alive until the current job ends, so yield before collecting.
+      yield* Effect.yieldNow;
+      collectGarbage();
+      assert.strictEqual(sweeps.filter((sweepData) => sweepData.deref() !== undefined).length, 0);
+    }).pipe(Effect.provide(KeyValueStore.layerMemory)),
+  );
+
   it.effect("cancels abandoned reads without blocking invalidation", () =>
     Effect.gen(function* () {
       const cache = yield* PullRequestReadCache.make;

apps/server/src/pullRequest/PullRequestReadCache.ts

@@ -21,6 +21,7 @@ import * as Persistable from "effect/persistence/Persistable";
 import * as PersistedCache from "effect/persistence/PersistedCache";
 import * as Persistence from "effect/persistence/Persistence";
 import * as ServerConfig from "../config.ts";
+import { detachStackFrame } from "../git/detachStackFrame.ts";
 import { forkParked } from "../serverActivation.ts";
 
 const CONCURRENT_READS = 512;
@@ -124,6 +125,8 @@ export const make = Effect.gen(function* () {
     }),
     { capacity: CONCURRENT_READS, timeToLive: () => Duration.zero },
   );
+  // A cache entry keeps the fiber that filled it, whose lazy stack frames would keep the caller's
+  // data (e.g. a whole sync sweep) alive until eviction.
   return PullRequestReadCache.of({
     get: Effect.fn("PullRequestReadCache.get")(function* (key, lookup, scopes = []) {
       if (!enabled) return yield* lookup;
@@ -151,7 +154,7 @@ export const make = Effect.gen(function* () {
         }),
         lock.withPermits(1),
       );
-    }),
+    }, detachStackFrame),
     invalidate: Effect.fn("PullRequestReadCache.invalidate")(
       function* (scope) {
         const now = clock.currentTimeMillisUnsafe();
@@ -170,6 +173,7 @@ export const make = Effect.gen(function* () {
       }),
       Effect.uninterruptible,
       lock.withPermits(CONCURRENT_READS),
+      detachStackFrame,
     ),
   });
 });