authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-09-09 16:33:25-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-09 16:33:25-07:00
log33242d0164d47c6935ba5c7725296539384de65d
tree99f5d68ae0d84de2d3cbba4244302438c2c6a0d9
parentde37964db26bd3260d253f138bfde2ac60fe254e
signature Signed by PGP key B5690EEEBB952194

fix(server): preserve recent PR reads across server restarts (#11007)

5 files changed, 275 insertions(+), 28 deletions(-)

apps/server/src/pullRequest/PullRequestReadCache.test.ts created+78
......@@ -0,0 +1,78 @@
1import { assert, it } from "@effect/vitest";
2import * as NodeServices from "@effect/platform-node/NodeServices";
3import { PullRequestOperationError } from "@t3tools/contracts";
4import * as Deferred from "effect/Deferred";
5import * as Effect from "effect/Effect";
6import * as Fiber from "effect/Fiber";
7import * as FileSystem from "effect/FileSystem";
8import * as Layer from "effect/Layer";
9import * as TestClock from "effect/testing/TestClock";
10import * as KeyValueStore from "effect/unstable/persistence/KeyValueStore";
11import * as Persistence from "effect/unstable/persistence/Persistence";
12import * as PullRequestReadCache from "./PullRequestReadCache.ts";
13
14const cacheLayer = (directory: string) =>
15 PullRequestReadCache.make.pipe(
16 Effect.provide(
17 Persistence.layerKvs.pipe(Layer.provideMerge(KeyValueStore.layerFileSystem(directory))),
18 ),
19 );
20
21it.layer(NodeServices.layer)("PR filesystem cache", (it) => {
22 it.effect("reuses files after restart and respects the original expiry", () =>
23 Effect.gen(function* () {
24 const fs = yield* FileSystem.FileSystem;
25 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pr-cache-" });
26 let reads = 0;
27 const lookup = Effect.sync(() => String(++reads));
28 const first = yield* cacheLayer(directory);
29 const key = "long/repository/key".repeat(100);
30 assert.strictEqual(yield* first.get(key, lookup), "1");
31 yield* TestClock.adjust("59 seconds");
32 const restarted = yield* cacheLayer(directory);
33 assert.strictEqual(yield* restarted.get(key, lookup), "1");
34 yield* TestClock.adjust("1 second");
35 assert.strictEqual(yield* restarted.get(key, lookup), "2");
36 assert.strictEqual(reads, 2);
37 assert.strictEqual((yield* fs.readDirectory(directory)).length, 1);
38 }),
39 );
40
41 it.effect("clears in-flight reads before a new service can reuse them", () =>
42 Effect.gen(function* () {
43 const fs = yield* FileSystem.FileSystem;
44 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pr-cache-" });
45 const started = yield* Deferred.make<void>();
46 const release = yield* Deferred.make<void>();
47 const cache = yield* cacheLayer(directory);
48 const read = yield* cache
49 .get(
50 "summary",
51 Deferred.succeed(started, undefined).pipe(
52 Effect.andThen(Deferred.await(release)),
53 Effect.as("old"),
54 ),
55 )
56 .pipe(Effect.forkChild);
57 yield* Deferred.await(started);
58 const invalidate = yield* cache.invalidate.pipe(Effect.forkChild({ startImmediately: true }));
59 yield* Deferred.succeed(release, undefined);
60 yield* Fiber.join(read);
61 yield* Fiber.join(invalidate);
62 const restarted = yield* cacheLayer(directory);
63 assert.strictEqual(yield* restarted.get("summary", Effect.succeed("new")), "new");
64 }),
65 );
66
67 it.effect("does not persist failed GitHub reads", () =>
68 Effect.gen(function* () {
69 const fs = yield* FileSystem.FileSystem;
70 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pr-cache-" });
71 const cache = yield* cacheLayer(directory);
72 const error = new PullRequestOperationError({ operation: "summary", detail: "unavailable" });
73 yield* cache.get("summary", Effect.fail(error)).pipe(Effect.flip);
74 const restarted = yield* cacheLayer(directory);
75 assert.strictEqual(yield* restarted.get("summary", Effect.succeed("recovered")), "recovered");
76 }),
77 );
78});
apps/server/src/pullRequest/PullRequestReadCache.ts created+125
......@@ -0,0 +1,125 @@
1import * as Cache from "effect/Cache";
2import * as Clock from "effect/Clock";
3import * as Equal from "effect/Equal";
4import * as Hash from "effect/Hash";
5import { PullRequestOperationError, PullRequestUnavailableError } from "@t3tools/contracts";
6import * as Crypto from "effect/Crypto";
7import * as Encoding from "effect/Encoding";
8import * as Option from "effect/Option";
9import * as Context from "effect/Context";
10import * as Duration from "effect/Duration";
11import * as Effect from "effect/Effect";
12import * as Exit from "effect/Exit";
13import * as Layer from "effect/Layer";
14import * as Path from "effect/Path";
15import * as Schema from "effect/Schema";
16import * as Semaphore from "effect/Semaphore";
17import * as KeyValueStore from "effect/unstable/persistence/KeyValueStore";
18import * as Persistable from "effect/unstable/persistence/Persistable";
19import * as PersistedCache from "effect/unstable/persistence/PersistedCache";
20import * as Persistence from "effect/unstable/persistence/Persistence";
21import { ServerConfig } from "../config.ts";
22
23const CONCURRENT_READS = 512;
24type ReadError = PullRequestOperationError | PullRequestUnavailableError;
25
26class Read extends Persistable.Class<{
27 payload: { key: string; lookup: Effect.Effect<string, ReadError> };
28}>()("PullRequestRead", {
29 primaryKey: ({ key }) => key,
30 success: Schema.Struct({ payload: Schema.String, expiresAt: Schema.Finite }),
31 error: Schema.Union([PullRequestOperationError, PullRequestUnavailableError]),
32}) {
33 [Equal.symbol](that: unknown): boolean {
34 return that instanceof Read && that.key === this.key;
35 }
36 [Hash.symbol](): number {
37 return Hash.string(this.key);
38 }
39}
40
41export class PullRequestReadCache extends Context.Service<
42 PullRequestReadCache,
43 {
44 readonly get: (
45 key: string,
46 lookup: Effect.Effect<string, ReadError>,
47 ) => Effect.Effect<string, ReadError>;
48 readonly invalidate: Effect.Effect<void>;
49 }
50>()("t3/pullRequest/PullRequestReadCache") {}
51
52export const make = Effect.gen(function* () {
53 const backing = yield* KeyValueStore.KeyValueStore;
54 const crypto = yield* Crypto.Crypto;
55 const clock = yield* Clock.Clock;
56 let enabled = true;
57 const lock = yield* Semaphore.make(CONCURRENT_READS);
58 const timeToLive: Persistable.TimeToLiveFn<Read> = (exit) =>
59 Exit.isSuccess(exit)
60 ? Duration.millis(Math.max(0, exit.value.expiresAt - clock.currentTimeMillisUnsafe()))
61 : Duration.zero;
62 const cache = yield* PersistedCache.make(
63 (request: Read) =>
64 request.lookup.pipe(
65 Effect.map((payload) => ({ payload, expiresAt: clock.currentTimeMillisUnsafe() + 60_000 })),
66 ),
67 {
68 storeId: "pr-v2",
69 timeToLive,
70 inMemoryTTL: timeToLive,
71 inMemoryCapacity: CONCURRENT_READS,
72 },
73 );
74 return PullRequestReadCache.of({
75 get: Effect.fn("PullRequestReadCache.get")(function* (key, lookup) {
76 if (!enabled) return yield* lookup;
77 const digest = yield* crypto
78 .digest("SHA-256", new TextEncoder().encode(key))
79 .pipe(Effect.option);
80 if (Option.isNone(digest)) return yield* lookup;
81 const read = yield* Effect.cached(lookup);
82 return yield* cache
83 .get(new Read({ key: Encoding.encodeHex(digest.value), lookup: read }))
84 .pipe(
85 Effect.map((result) => result.payload),
86 Effect.catchTags({
87 PersistenceError: () => read,
88 SchemaError: () => read,
89 }),
90 Effect.uninterruptible,
91 lock.withPermits(1),
92 );
93 }),
94 // Let existing reads finish before clearing, so they cannot repopulate stale entries.
95 invalidate: Cache.invalidateAll(cache.inMemory).pipe(
96 Effect.andThen(backing.clear),
97 Effect.catch(() => {
98 enabled = false;
99 return Effect.logWarning("PR cache disabled after clearing failed");
100 }),
101 lock.withPermits(CONCURRENT_READS),
102 ),
103 });
104});
105
106export const layer = Layer.unwrap(
107 Effect.gen(function* () {
108 const config = yield* ServerConfig;
109 const path = yield* Path.Path;
110 return Layer.effect(PullRequestReadCache, make).pipe(
111 Layer.provide(Persistence.layerKvs),
112 Layer.provide(
113 KeyValueStore.layerFileSystem(
114 path.join(config.providerStatusCacheDir, "pull-requests"),
115 ).pipe(
116 Layer.catch(() =>
117 Layer.effectDiscard(
118 Effect.logWarning("PR cache directory unavailable; using memory cache"),
119 ).pipe(Layer.provideMerge(KeyValueStore.layerMemory)),
120 ),
121 ),
122 ),
123 );
124 }),
125);
apps/server/src/pullRequest/PullRequestService.test.ts+9
......@@ -1,3 +1,6 @@
1import * as NodeServices from "@effect/platform-node/NodeServices";
2import * as KeyValueStore from "effect/unstable/persistence/KeyValueStore";
3import * as Persistence from "effect/unstable/persistence/Persistence";
14import { assert, it } from "@effect/vitest";
25import * as Deferred from "effect/Deferred";
36import * as Effect from "effect/Effect";
......@@ -24,6 +27,7 @@ import {
2427} from "./PullRequestProvider.ts";
2528import { PullRequestProviderRegistry, fromProviders } from "./PullRequestProviderRegistry.ts";
2629import * as PullRequestService from "./PullRequestService.ts";
30import * as PullRequestReadCache from "./PullRequestReadCache.ts";
2731
2832function project(input: {
2933 readonly id: string;
......@@ -198,6 +202,11 @@ function makeService(input: {
198202 }),
199203 }),
200204 SourceControlRateLimit.layer,
205 Layer.effect(PullRequestReadCache.PullRequestReadCache, PullRequestReadCache.make).pipe(
206 Layer.provide(Persistence.layerKvs),
207 Layer.provide(KeyValueStore.layerMemory),
208 Layer.provide(NodeServices.layer),
209 ),
201210 ),
202211 ),
203212 );
apps/server/src/pullRequest/PullRequestService.ts+61-28
......@@ -11,6 +11,7 @@ import * as Effect from "effect/Effect";
1111import * as Exit from "effect/Exit";
1212import * as Layer from "effect/Layer";
1313import * as PubSub from "effect/PubSub";
14import * as Option from "effect/Option";
1415import * as Schema from "effect/Schema";
1516import type * as Scope from "effect/Scope";
1617import * as Stream from "effect/Stream";
......@@ -50,8 +51,8 @@ import {
5051 type PullRequestLabelCandidateList,
5152 type PullRequestLabelChangeInput,
5253 type PullRequestSubmitReviewInput,
53 type PullRequestStack,
54 type PullRequestSummary,
54 PullRequestStack,
55 PullRequestSummary,
5556 type PullRequestThreadReplyInput,
5657 type PullRequestThreadResolutionInput,
5758 type PullRequestThreadCommentsInput,
......@@ -71,6 +72,7 @@ import {
7172 type PullRequestProviderApi,
7273 PullRequestProviderError,
7374} from "./PullRequestProvider.ts";
75import * as PullRequestReadCache from "./PullRequestReadCache.ts";
7476import { PullRequestProviderRegistry } from "./PullRequestProviderRegistry.ts";
7577
7678export interface PullRequestMergeEvent extends PullRequestRef {
......@@ -114,7 +116,6 @@ const REPOSITORY_SEARCH_CHUNK = 100;
114116 * `invalidate` rather than a flag on the read, so an ordinary read can never opt out.
115117 */
116118const LIST_CACHE_TTL = Duration.seconds(30);
117const SUMMARY_CACHE_TTL = Duration.seconds(60);
118119const DETAIL_CACHE_TTL = Duration.seconds(15);
119120const DIFF_CACHE_TTL = Duration.seconds(60);
120121/** A commit is content-addressed, so its own diff cannot change under its key. */
......@@ -535,6 +536,7 @@ export const make = Effect.gen(function* () {
535536 const projections = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery;
536537 const sourceControlProviders = yield* SourceControlProviderRegistry.SourceControlProviderRegistry;
537538 const rateLimits = yield* SourceControlRateLimit.SourceControlRateLimit;
539 const readCache = yield* PullRequestReadCache.PullRequestReadCache;
538540
539541 const refineUnknownProjectKinds = (
540542 projects: ReadonlyArray<OrchestrationProjectShell>,
......@@ -2333,18 +2335,49 @@ export const make = Effect.gen(function* () {
23332335 };
23342336 };
23352337
2336 const summaryCache = yield* Cache.makeWith(
2337 (key: string) => {
2338 return summaryUncached(refOfCacheKey(key));
2339 },
2340 {
2341 capacity: DETAIL_CACHE_CAPACITY,
2342 timeToLive: (exit) => (Exit.isSuccess(exit) ? SUMMARY_CACHE_TTL : Duration.zero),
2343 },
2344 );
2338 const persistedRead = Effect.fn("PullRequestService.persistedRead")(function* <A>(
2339 input: PullRequestRef,
2340 operation: string,
2341 codec: Schema.Codec<A, string>,
2342 read: Effect.Effect<A, PullRequestError>,
2343 ) {
2344 const project = yield* requireProject(input);
2345 const key = [
2346 operation,
2347 project.api.kind,
2348 project.host.toLowerCase(),
2349 project.repository.toLowerCase(),
2350 project.project.id,
2351 project.project.workspaceRoot,
2352 String(input.number),
2353 ]
2354 .map(encodeURIComponent)
2355 .join(":");
2356 const lookup = yield* Effect.cached(read);
2357 const encodedRead = lookup.pipe(
2358 Effect.flatMap((value) =>
2359 Schema.encodeEffect(codec)(value).pipe(
2360 Effect.mapError(
2361 (cause) =>
2362 new PullRequestOperationError({
2363 operation: "cache",
2364 detail: "Could not encode PR cache data.",
2365 cause,
2366 }),
2367 ),
2368 ),
2369 ),
2370 );
2371 const payload = yield* readCache.get(key, encodedRead);
2372 const decoded = yield* Schema.decodeUnknownEffect(codec)(payload).pipe(Effect.option);
2373 return Option.isSome(decoded) ? decoded.value : yield* lookup;
2374 });
2375 const summaryCodec = Schema.fromJsonString(PullRequestSummary);
2376 const stackCodec = Schema.fromJsonString(Schema.NullOr(PullRequestStack));
2377
23452378 const summary: PullRequestService["Service"]["summary"] = (input, options) => {
23462379 const key = refCacheKey(input);
2347 const cached = Cache.get(summaryCache, key);
2380 const cached = persistedRead(input, "summary", summaryCodec, summaryUncached(input));
23482381 const held = lastGoodSummary.peek(key);
23492382 return held !== undefined &&
23502383 (options?.recoverTransientFailure !== false || held.state === "merged")
......@@ -2356,18 +2389,13 @@ export const make = Effect.gen(function* () {
23562389 );
23572390 };
23582391
2359 const stackCache = yield* Cache.makeWith(
2360 (key: string) => {
2361 const [referenceKey, includeDetails] = JSON.parse(key) as [string, boolean];
2362 return stackUncached(refOfCacheKey(referenceKey), { includeDetails });
2363 },
2364 {
2365 capacity: DETAIL_CACHE_CAPACITY,
2366 timeToLive: (exit) => (Exit.isSuccess(exit) ? SUMMARY_CACHE_TTL : Duration.zero),
2367 },
2368 );
23692392 const stack: PullRequestService["Service"]["stack"] = (input, options) =>
2370 Cache.get(stackCache, JSON.stringify([refCacheKey(input), options?.includeDetails !== false]));
2393 persistedRead(
2394 input,
2395 `stack:${options?.includeDetails !== false}`,
2396 stackCodec,
2397 stackUncached(input, options),
2398 );
23712399
23722400 // Keys serialize positionally and parse back in the lookup, so the cache is the only holder
23732401 // of in-flight state: concurrent identical reads coalesce on the key into one host request.
......@@ -2642,7 +2670,7 @@ export const make = Effect.gen(function* () {
26422670 const invalidate: PullRequestService["Service"]["invalidate"] = (input) => {
26432671 const reference = input.reference;
26442672 if (reference !== undefined) {
2645 return Effect.sync(() => bumpRefEpoch(reference));
2673 return readCache.invalidate.pipe(Effect.andThen(Effect.sync(() => bumpRefEpoch(reference))));
26462674 }
26472675 return Effect.sync(() => {
26482676 listingsEpoch = ++epochCounter;
......@@ -2652,7 +2680,9 @@ export const make = Effect.gen(function* () {
26522680
26532681 const refreshAfterTurn: PullRequestService["Service"]["refreshAfterTurn"] = Effect.suspend(() => {
26542682 turnRefreshEpoch = listingsEpoch = ++epochCounter;
2655 return SubscriptionRef.set(pullRequestRefreshes, turnRefreshEpoch);
2683 return readCache.invalidate.pipe(
2684 Effect.andThen(SubscriptionRef.set(pullRequestRefreshes, turnRefreshEpoch)),
2685 );
26562686 });
26572687
26582688 // A mutation's own client re-reads right after it, and every other client's next read must
......@@ -2663,7 +2693,9 @@ export const make = Effect.gen(function* () {
26632693 method: (input: I) => Effect.Effect<void, PullRequestError>,
26642694 ): ((input: I) => Effect.Effect<void, PullRequestError>) =>
26652695 (input) =>
2666 method(input).pipe(
2696 readCache.invalidate.pipe(
2697 Effect.andThen(method(input)),
2698 Effect.ensuring(readCache.invalidate),
26672699 Effect.tap(() =>
26682700 Effect.sync(() => {
26692701 bumpRefEpoch(input);
......@@ -2674,7 +2706,8 @@ export const make = Effect.gen(function* () {
26742706 const runActionAndInvalidate: PullRequestService["Service"]["runAction"] = Effect.fn(
26752707 "PullRequestService.runActionAndInvalidate",
26762708 )(function* (input) {
2677 const repository = yield* runAction(input);
2709 yield* readCache.invalidate;
2710 const repository = yield* runAction(input).pipe(Effect.ensuring(readCache.invalidate));
26782711 bumpRefEpoch({ ...input, repository });
26792712 listingsEpoch = ++epochCounter;
26802713 if (input.action === "merge") {
apps/server/src/server.ts+2
......@@ -93,6 +93,7 @@ import * as VcsStatusBroadcaster from "./vcs/VcsStatusBroadcaster.ts";
9393import * as GitWorkflowService from "./git/GitWorkflowService.ts";
9494import * as ReviewService from "./review/ReviewService.ts";
9595import * as SourceControlProviderRegistry from "./sourceControl/SourceControlProviderRegistry.ts";
96import * as PullRequestReadCache from "./pullRequest/PullRequestReadCache.ts";
9697import * as SourceControlRateLimit from "./sourceControl/SourceControlRateLimit.ts";
9798import * as SourceControlRepositoryService from "./sourceControl/SourceControlRepositoryService.ts";
9899import * as ProjectSetupScriptRunner from "./project/ProjectSetupScriptRunner.ts";
......@@ -318,6 +319,7 @@ const SourceControlProviderRegistryLayerLive = SourceControlProviderRegistry.lay
318319
319320const PullRequestServiceLive = PullRequestService.layer.pipe(
320321 Layer.provide(PullRequestProviderRegistry.layer),
322 Layer.provide(PullRequestReadCache.layer),
321323 Layer.provide(SourceControlProviderRegistryLayerLive),
322324 Layer.provide(SourceControlRateLimit.layer),
323325);