authorgravatar for 62337003+Bil0000@users.noreply.github.comBilal Bakr <62337003+Bil0000@users.noreply.github.com> 2026-09-16 01:30:58+03:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-15 15:30:58-07:00
logf4600d77dd7c2fa9f10e8f4500882e427fcc7e26
tree0cd5bab04d8947624e943639009a70e2fbd9cd64
parentb84f63bb13f13b1d0ee8537c4d244d6f1efdb987
signature Signed by PGP key B5690EEEBB952194

fix: reduce GitHub quota use with sharing enabled (#11888)

Co-authored-by: Julius Marminge <julius0216@outlook.com>

20 files changed, 1020 insertions(+), 273 deletions(-)

apps/server/integration/OrchestrationEngineHarness.integration.ts+1-1
...@@ -352,7 +352,7 @@ export const makeOrchestrationIntegrationHarness = (...@@ -352,7 +352,7 @@ export const makeOrchestrationIntegrationHarness = (
352 Layer.provideMerge(runtimeServicesLayer),352 Layer.provideMerge(runtimeServicesLayer),
353 Layer.provideMerge(353 Layer.provideMerge(
354 Layer.mock(PullRequestService.PullRequestService)({354 Layer.mock(PullRequestService.PullRequestService)({
355 refreshAfterTurn: Effect.void,355 refreshAfterTurn: () => Effect.void,
356 }),356 }),
357 ),357 ),
358 Layer.provideMerge(358 Layer.provideMerge(
apps/server/src/orchestration/Layers/CheckpointReactor.test.ts+1-1
...@@ -336,7 +336,7 @@ describe("CheckpointReactor", () => {...@@ -336,7 +336,7 @@ describe("CheckpointReactor", () => {
336 prefix: "t3-checkpoint-reactor-test-",336 prefix: "t3-checkpoint-reactor-test-",
337 });337 });
338 const pullRequestRefreshes: number[] = [];338 const pullRequestRefreshes: number[] = [];
339 const refreshAfterTurn = Effect.sync(() => void pullRequestRefreshes.push(1));339 const refreshAfterTurn = () => Effect.sync(() => void pullRequestRefreshes.push(1));
340 const vcsStatusBroadcasterLayer = Layer.succeed(VcsStatusBroadcaster, {340 const vcsStatusBroadcasterLayer = Layer.succeed(VcsStatusBroadcaster, {
341 getStatus: () => Effect.die("getStatus should not be called in this test"),341 getStatus: () => Effect.die("getStatus should not be called in this test"),
342 refreshLocalStatus: (cwd: string) =>342 refreshLocalStatus: (cwd: string) =>
apps/server/src/orchestration/Layers/CheckpointReactor.ts+1-1
...@@ -886,7 +886,7 @@ const make = Effect.gen(function* () {...@@ -886,7 +886,7 @@ const make = Effect.gen(function* () {
886 (startedTurnId === undefined && !thread.session?.activeTurnId))886 (startedTurnId === undefined && !thread.session?.activeTurnId))
887 ) {887 ) {
888 pending.delete(event.threadId);888 pending.delete(event.threadId);
889 yield* pullRequests.refreshAfterTurn;889 yield* pullRequests.refreshAfterTurn(thread.projectId);
890 }890 }
891 if (891 if (
892 event.type === "turn.aborted" &&892 event.type === "turn.aborted" &&
apps/server/src/pullRequest/GitHubPullRequestCli.test.ts+2
...@@ -9,6 +9,7 @@ import { ChildProcessSpawner } from "effect/unstable/process";...@@ -9,6 +9,7 @@ import { ChildProcessSpawner } from "effect/unstable/process";
99
10import * as GitHubCli from "../sourceControl/GitHubCli.ts";10import * as GitHubCli from "../sourceControl/GitHubCli.ts";
11import * as VcsProcess from "../vcs/VcsProcess.ts";11import * as VcsProcess from "../vcs/VcsProcess.ts";
12import * as SourceControlRateLimit from "../sourceControl/SourceControlRateLimit.ts";
12import * as GitHubGraphQlBudget from "../sourceControl/githubGraphQlBudget.ts";13import * as GitHubGraphQlBudget from "../sourceControl/githubGraphQlBudget.ts";
13import * as GitHubPullRequestCli from "./GitHubPullRequestCli.ts";14import * as GitHubPullRequestCli from "./GitHubPullRequestCli.ts";
14import { BASE_COMPARISON_GRAPHQL_QUERY } from "./gitHubPullRequestJson.ts";15import { BASE_COMPARISON_GRAPHQL_QUERY } from "./gitHubPullRequestJson.ts";
...@@ -202,6 +203,7 @@ it.effect(...@@ -202,6 +203,7 @@ it.effect(
202 let activeToken = "broad-credential";203 let activeToken = "broad-credential";
203 const commands: VcsProcess.VcsProcessInput[] = [];204 const commands: VcsProcess.VcsProcessInput[] = [];
204 const github = yield* GitHubCli.make.pipe(205 const github = yield* GitHubCli.make.pipe(
206 Effect.provide(Layer.merge(GitHubGraphQlBudget.layer, SourceControlRateLimit.layer)),
205 Effect.provideService(VcsProcess.VcsProcess, {207 Effect.provideService(VcsProcess.VcsProcess, {
206 run: (input) =>208 run: (input) =>
207 Effect.sync(() => {209 Effect.sync(() => {
apps/server/src/pullRequest/GitHubPullRequestCli.ts+1
...@@ -1106,6 +1106,7 @@ export const make = Effect.gen(function* () {...@@ -1106,6 +1106,7 @@ export const make = Effect.gen(function* () {
1106 captureVerifiedCredential(input).pipe(1106 captureVerifiedCredential(input).pipe(
1107 Effect.flatMap(({ host, token, accountId, viewer, credentialFingerprint }) =>1107 Effect.flatMap(({ host, token, accountId, viewer, credentialFingerprint }) =>
1108 use({ accountId, viewer, credentialFingerprint }).pipe(1108 use({ accountId, viewer, credentialFingerprint }).pipe(
1109 Effect.provideService(SourceControlRateLimit.CredentialScope, credentialFingerprint),
1109 Effect.provideService(GitHubCli.PinnedGitHubCredential, {1110 Effect.provideService(GitHubCli.PinnedGitHubCredential, {
1110 host,1111 host,
1111 token,1112 token,
apps/server/src/pullRequest/GitHubPullRequestProvider.ts+5-1
...@@ -109,7 +109,11 @@ export function gitHubProviderFailure(...@@ -109,7 +109,11 @@ export function gitHubProviderFailure(
109): PullRequestProviderFailure {109): PullRequestProviderFailure {
110 if (error._tag === "GitHubCliUnavailableError") return { reason: "missing-tool" };110 if (error._tag === "GitHubCliUnavailableError") return { reason: "missing-tool" };
111 if (error._tag === "GitHubCliAuthenticationError") return { reason: "unauthenticated" };111 if (error._tag === "GitHubCliAuthenticationError") return { reason: "unauthenticated" };
112 if (error._tag === "GitHubCliRateLimitError") return { reason: "rate-limited" };112 if (error._tag === "GitHubCliRateLimitError")
113 return {
114 reason: "rate-limited",
115 ...(error.retryAt === undefined ? {} : { retryAt: error.retryAt }),
116 };
113 if (error._tag === "SourceControlRateLimitPausedError") {117 if (error._tag === "SourceControlRateLimitPausedError") {
114 return { reason: "rate-limited", retryAt: error.retryAt };118 return { reason: "rate-limited", retryAt: error.retryAt };
115 }119 }
apps/server/src/pullRequest/PullRequestReadCache.test.ts+175-9
...@@ -5,18 +5,12 @@ import * as Deferred from "effect/Deferred";...@@ -5,18 +5,12 @@ import * as Deferred from "effect/Deferred";
5import * as Effect from "effect/Effect";5import * as Effect from "effect/Effect";
6import * as Fiber from "effect/Fiber";6import * as Fiber from "effect/Fiber";
7import * as FileSystem from "effect/FileSystem";7import * as FileSystem from "effect/FileSystem";
8import * as Layer from "effect/Layer";
9import * as TestClock from "effect/testing/TestClock";8import * as TestClock from "effect/testing/TestClock";
10import * as KeyValueStore from "effect/unstable/persistence/KeyValueStore";9import * as KeyValueStore from "effect/unstable/persistence/KeyValueStore";
11import * as Persistence from "effect/unstable/persistence/Persistence";
12import * as PullRequestReadCache from "./PullRequestReadCache.ts";10import * as PullRequestReadCache from "./PullRequestReadCache.ts";
1311
14const cacheLayer = (directory: string) =>12const cacheLayer = (directory: string) =>
15 PullRequestReadCache.make.pipe(13 PullRequestReadCache.make.pipe(Effect.provide(KeyValueStore.layerFileSystem(directory)));
16 Effect.provide(
17 Persistence.layerKvs.pipe(Layer.provideMerge(KeyValueStore.layerFileSystem(directory))),
18 ),
19 );
2014
21it.layer(NodeServices.layer)("PR filesystem cache", (it) => {15it.layer(NodeServices.layer)("PR filesystem cache", (it) => {
22 it.effect("reuses files after restart and respects the original expiry", () =>16 it.effect("reuses files after restart and respects the original expiry", () =>
...@@ -52,15 +46,106 @@ it.layer(NodeServices.layer)("PR filesystem cache", (it) => {...@@ -52,15 +46,106 @@ it.layer(NodeServices.layer)("PR filesystem cache", (it) => {
52 Effect.andThen(Deferred.await(release)),46 Effect.andThen(Deferred.await(release)),
53 Effect.as("old"),47 Effect.as("old"),
54 ),48 ),
49 ["pr"],
55 )50 )
56 .pipe(Effect.forkChild);51 .pipe(Effect.forkChild);
57 yield* Deferred.await(started);52 yield* Deferred.await(started);
58 const invalidate = yield* cache.invalidate.pipe(Effect.forkChild({ startImmediately: true }));53 const invalidate = yield* cache
54 .invalidate("pr")
55 .pipe(Effect.forkChild({ startImmediately: true }));
59 yield* Deferred.succeed(release, undefined);56 yield* Deferred.succeed(release, undefined);
60 yield* Fiber.join(read);57 yield* Fiber.join(read);
61 yield* Fiber.join(invalidate);58 yield* Fiber.join(invalidate);
62 const restarted = yield* cacheLayer(directory);59 const restarted = yield* cacheLayer(directory);
63 assert.strictEqual(yield* restarted.get("summary", Effect.succeed("new")), "new");60 assert.strictEqual(yield* restarted.get("summary", Effect.succeed("new"), ["pr"]), "new");
61 }),
62 );
63
64 it.effect("invalidates only the changed scope across restarts and coalesces its next reads", () =>
65 Effect.gen(function* () {
66 const fs = yield* FileSystem.FileSystem;
67 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pr-cache-" });
68 let reads = 0;
69 const lookup = Effect.sync(() => String(++reads));
70 const cache = yield* cacheLayer(directory);
71 yield* cache.get("first", lookup, ["project", "pr-1"]);
72 yield* cache.get("second", lookup, ["project", "pr-2"]);
73 yield* cache.get("third", lookup, ["other-project", "pr-3"]);
74 yield* cache.invalidate("pr-1");
75 const restarted = yield* cacheLayer(directory);
76 const answers = yield* Effect.all(
77 Array.from({ length: 10 }, () => restarted.get("first", lookup, ["project", "pr-1"])),
78 { concurrency: 10 },
79 );
80 assert.deepStrictEqual(answers, Array(10).fill("4"));
81 assert.strictEqual(yield* restarted.get("second", lookup, ["project", "pr-2"]), "2");
82 yield* restarted.invalidate("project");
83 const again = yield* cacheLayer(directory);
84 assert.strictEqual(yield* again.get("second", lookup, ["project", "pr-2"]), "5");
85 assert.strictEqual(yield* again.get("third", lookup, ["other-project", "pr-3"]), "3");
86 assert.strictEqual(reads, 5);
87 const files = (yield* fs.readDirectory(directory)).length;
88 for (let index = 0; index < 3; index++) {
89 yield* again.invalidate("pr-1");
90 yield* again.get("first", lookup, ["project", "pr-1"]);
91 }
92 assert.strictEqual((yield* fs.readDirectory(directory)).length, files);
93 }),
94 );
95
96 it.effect("shares a pending refresh without blocking an unrelated cached PR", () =>
97 Effect.gen(function* () {
98 const cache = yield* PullRequestReadCache.make;
99 yield* cache.get("first", Effect.succeed("old"), ["pr-1"]);
100 yield* cache.get("second", Effect.succeed("warm"), ["pr-2"]);
101 yield* cache.invalidate("pr-1");
102 const started = yield* Deferred.make<void>();
103 const release = yield* Deferred.make<void>();
104 let reads = 0;
105 const refresh = cache.get(
106 "first",
107 Effect.gen(function* () {
108 reads++;
109 yield* Deferred.succeed(started, undefined);
110 yield* Deferred.await(release);
111 return "fresh";
112 }),
113 ["pr-1"],
114 );
115 const pending = yield* Effect.all(
116 Array.from({ length: 10 }, () => refresh),
117 {
118 concurrency: 10,
119 },
120 ).pipe(Effect.forkChild);
121 yield* Deferred.await(started);
122 assert.strictEqual(yield* cache.get("second", Effect.die("cache miss"), ["pr-2"]), "warm");
123 yield* Deferred.succeed(release, undefined);
124 assert.deepStrictEqual(yield* Fiber.join(pending), Array(10).fill("fresh"));
125 assert.strictEqual(reads, 1);
126 }).pipe(Effect.provide(KeyValueStore.layerMemory)),
127 );
128
129 it.effect("compacts expired scope records without discarding fresh PR data", () =>
130 Effect.gen(function* () {
131 const fs = yield* FileSystem.FileSystem;
132 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pr-cache-" });
133 const cache = yield* cacheLayer(directory);
134 yield* cache.get("summary", Effect.succeed("old"), ["pr"]);
135 yield* cache.invalidate("pr");
136 for (let index = 0; index < 100; index++) yield* cache.invalidate(`pr-${index}`);
137 assert.strictEqual((yield* fs.readDirectory(directory)).length, 2);
138 const before = (yield* fs.stat(`${directory}/revisions`)).size;
139 yield* TestClock.adjust("59 seconds");
140 assert.strictEqual(yield* cache.get("summary", Effect.succeed("fresh"), ["pr"]), "fresh");
141 yield* TestClock.adjust("1 second");
142 yield* cache.invalidate("other-pr");
143 assert.isTrue((yield* fs.stat(`${directory}/revisions`)).size < before);
144 const restarted = yield* cacheLayer(directory);
145 assert.strictEqual(
146 yield* restarted.get("summary", Effect.die("cache miss"), ["pr"]),
147 "fresh",
148 );
64 }),149 }),
65 );150 );
66151
...@@ -75,4 +160,85 @@ it.layer(NodeServices.layer)("PR filesystem cache", (it) => {...@@ -75,4 +160,85 @@ it.layer(NodeServices.layer)("PR filesystem cache", (it) => {
75 assert.strictEqual(yield* restarted.get("summary", Effect.succeed("recovered")), "recovered");160 assert.strictEqual(yield* restarted.get("summary", Effect.succeed("recovered")), "recovered");
76 }),161 }),
77 );162 );
163
164 it.effect("resumes caching after a failed scope read", () =>
165 Effect.gen(function* () {
166 const backing = yield* KeyValueStore.KeyValueStore;
167 let fail = true;
168 let reads = 0;
169 const cache = yield* PullRequestReadCache.make.pipe(
170 Effect.provideService(KeyValueStore.KeyValueStore, {
171 ...backing,
172 get: (key) =>
173 Effect.suspend(() => {
174 if (!fail) return backing.get(key);
175 fail = false;
176 return Effect.fail(
177 new KeyValueStore.KeyValueStoreError({ method: "get", message: "unavailable" }),
178 );
179 }),
180 }),
181 );
182 const read = cache.get(
183 "summary",
184 Effect.sync(() => String(++reads)),
185 ["pr"],
186 );
187 assert.strictEqual(yield* read, "1");
188 assert.strictEqual(yield* read, "2");
189 assert.strictEqual(yield* read, "2");
190 }).pipe(Effect.provide(KeyValueStore.layerMemory)),
191 );
192
193 it.effect("cancels abandoned reads without blocking invalidation", () =>
194 Effect.gen(function* () {
195 const cache = yield* PullRequestReadCache.make;
196 const started = yield* Deferred.make<void>();
197 const read = yield* cache
198 .get("summary", Deferred.succeed(started, undefined).pipe(Effect.andThen(Effect.never)), [
199 "pr",
200 ])
201 .pipe(Effect.forkChild);
202 yield* Deferred.await(started);
203 yield* Fiber.interrupt(read);
204 yield* cache.invalidate("pr");
205 assert.strictEqual(yield* cache.get("summary", Effect.succeed("fresh"), ["pr"]), "fresh");
206 }).pipe(Effect.provide(KeyValueStore.layerMemory)),
207 );
208
209 it.effect(
210 "finishes the in-memory revision update when invalidation is canceled after writing",
211 () =>
212 Effect.gen(function* () {
213 const backing = yield* KeyValueStore.KeyValueStore;
214 const written = yield* Deferred.make<void>();
215 const release = yield* Deferred.make<void>();
216 const cache = yield* PullRequestReadCache.make.pipe(
217 Effect.provideService(KeyValueStore.KeyValueStore, {
218 ...backing,
219 set: (key, value) =>
220 backing
221 .set(key, value)
222 .pipe(
223 Effect.andThen(
224 key === "revisions"
225 ? Deferred.succeed(written, undefined).pipe(
226 Effect.andThen(Deferred.await(release)),
227 )
228 : Effect.void,
229 ),
230 ),
231 }),
232 );
233 yield* cache.get("summary", Effect.succeed("old"), ["pr"]);
234 const invalidation = yield* cache.invalidate("pr").pipe(Effect.forkChild);
235 yield* Deferred.await(written);
236 const interrupt = yield* Fiber.interrupt(invalidation).pipe(
237 Effect.forkChild({ startImmediately: true }),
238 );
239 yield* Deferred.succeed(release, undefined);
240 yield* Fiber.join(interrupt);
241 assert.strictEqual(yield* cache.get("summary", Effect.succeed("fresh"), ["pr"]), "fresh");
242 }).pipe(Effect.provide(KeyValueStore.layerMemory)),
243 );
78});244});
apps/server/src/pullRequest/PullRequestReadCache.ts+92-34
...@@ -5,7 +5,6 @@ import * as Hash from "effect/Hash";...@@ -5,7 +5,6 @@ import * as Hash from "effect/Hash";
5import { PullRequestOperationError, PullRequestUnavailableError } from "@t3tools/contracts";5import { PullRequestOperationError, PullRequestUnavailableError } from "@t3tools/contracts";
6import * as Crypto from "effect/Crypto";6import * as Crypto from "effect/Crypto";
7import * as Encoding from "effect/Encoding";7import * as Encoding from "effect/Encoding";
8import * as Option from "effect/Option";
9import * as Context from "effect/Context";8import * as Context from "effect/Context";
10import * as Duration from "effect/Duration";9import * as Duration from "effect/Duration";
11import * as Effect from "effect/Effect";10import * as Effect from "effect/Effect";
...@@ -22,19 +21,35 @@ import { ServerConfig } from "../config.ts";...@@ -22,19 +21,35 @@ import { ServerConfig } from "../config.ts";
2221
23const CONCURRENT_READS = 512;22const CONCURRENT_READS = 512;
24type ReadError = PullRequestOperationError | PullRequestUnavailableError;23type ReadError = PullRequestOperationError | PullRequestUnavailableError;
2524const revisionCodec = Schema.fromJsonString(
25 Schema.Record(
26 Schema.String,
27 Schema.Struct({ revision: Schema.String, expiresAt: Schema.Finite }),
28 ),
29);
26class Read extends Persistable.Class<{30class Read extends Persistable.Class<{
27 payload: { key: string; lookup: Effect.Effect<string, ReadError> };31 payload: { key: string; revision: string; lookup: Effect.Effect<string, ReadError> };
28}>()("PullRequestRead", {32}>()("PullRequestRead", {
29 primaryKey: ({ key }) => key,33 primaryKey: ({ key }) => key,
30 success: Schema.Struct({ payload: Schema.String, expiresAt: Schema.Finite }),34 success: Schema.Struct({
35 payload: Schema.String,
36 expiresAt: Schema.Finite,
37 revision: Schema.optionalKey(Schema.String),
38 }),
31 error: Schema.Union([PullRequestOperationError, PullRequestUnavailableError]),39 error: Schema.Union([PullRequestOperationError, PullRequestUnavailableError]),
32}) {40}) {
41 matchesRevision(revision: string | undefined): boolean {
42 const stored = revision?.split(":") ?? [];
43 return this.revision
44 .split(":")
45 .every((value, index) => value === "" || value === stored[index]);
46 }
47
33 [Equal.symbol](that: unknown): boolean {48 [Equal.symbol](that: unknown): boolean {
34 return that instanceof Read && that.key === this.key;49 return that instanceof Read && that.key === this.key && that.revision === this.revision;
35 }50 }
36 [Hash.symbol](): number {51 [Hash.symbol](): number {
37 return Hash.string(this.key);52 return Hash.string(`${this.key}:${this.revision}`);
38 }53 }
39}54}
4055
...@@ -44,8 +59,9 @@ export class PullRequestReadCache extends Context.Service<...@@ -44,8 +59,9 @@ export class PullRequestReadCache extends Context.Service<
44 readonly get: (59 readonly get: (
45 key: string,60 key: string,
46 lookup: Effect.Effect<string, ReadError>,61 lookup: Effect.Effect<string, ReadError>,
62 scopes?: ReadonlyArray<string>,
47 ) => Effect.Effect<string, ReadError>;63 ) => Effect.Effect<string, ReadError>;
48 readonly invalidate: Effect.Effect<void>;64 readonly invalidate: (scope: string) => Effect.Effect<void>;
49 }65 }
50>()("t3/pullRequest/PullRequestReadCache") {}66>()("t3/pullRequest/PullRequestReadCache") {}
5167
...@@ -55,6 +71,18 @@ export const make = Effect.gen(function* () {...@@ -55,6 +71,18 @@ export const make = Effect.gen(function* () {
55 const clock = yield* Clock.Clock;71 const clock = yield* Clock.Clock;
56 let enabled = true;72 let enabled = true;
57 const lock = yield* Semaphore.make(CONCURRENT_READS);73 const lock = yield* Semaphore.make(CONCURRENT_READS);
74 const digest = (key: string) =>
75 crypto.digest("SHA-256", new TextEncoder().encode(key)).pipe(Effect.map(Encoding.encodeHex));
76 const revisions = yield* Cache.makeWith(
77 () =>
78 backing
79 .get("revisions")
80 .pipe(Effect.flatMap((raw) => Schema.decodeUnknownEffect(revisionCodec)(raw ?? "{}"))),
81 {
82 capacity: 1,
83 timeToLive: (exit) => (Exit.isSuccess(exit) ? Duration.infinity : Duration.zero),
84 },
85 );
58 const timeToLive: Persistable.TimeToLiveFn<Read> = (exit) =>86 const timeToLive: Persistable.TimeToLiveFn<Read> = (exit) =>
59 Exit.isSuccess(exit)87 Exit.isSuccess(exit)
60 ? Duration.millis(Math.max(0, exit.value.expiresAt - clock.currentTimeMillisUnsafe()))88 ? Duration.millis(Math.max(0, exit.value.expiresAt - clock.currentTimeMillisUnsafe()))
...@@ -62,7 +90,11 @@ export const make = Effect.gen(function* () {...@@ -62,7 +90,11 @@ export const make = Effect.gen(function* () {
62 const cache = yield* PersistedCache.make(90 const cache = yield* PersistedCache.make(
63 (request: Read) =>91 (request: Read) =>
64 request.lookup.pipe(92 request.lookup.pipe(
65 Effect.map((payload) => ({ payload, expiresAt: clock.currentTimeMillisUnsafe() + 60_000 })),93 Effect.map((payload) => ({
94 payload,
95 expiresAt: clock.currentTimeMillisUnsafe() + 60_000,
96 revision: request.revision,
97 })),
66 ),98 ),
67 {99 {
68 storeId: "pr-v2",100 storeId: "pr-v2",
...@@ -70,36 +102,63 @@ export const make = Effect.gen(function* () {...@@ -70,36 +102,63 @@ export const make = Effect.gen(function* () {
70 inMemoryTTL: timeToLive,102 inMemoryTTL: timeToLive,
71 inMemoryCapacity: CONCURRENT_READS,103 inMemoryCapacity: CONCURRENT_READS,
72 },104 },
105 ).pipe(Effect.provide(Persistence.layerKvs));
106 const refreshes = yield* Cache.makeWith(
107 Effect.fn("PullRequestReadCache.refresh")(function* (request: Read) {
108 const stored = yield* cache.get(request);
109 if (request.matchesRevision(stored.revision)) return stored;
110 yield* cache.invalidate(request);
111 return yield* cache.get(request);
112 }),
113 { capacity: CONCURRENT_READS, timeToLive: () => Duration.zero },
73 );114 );
74 return PullRequestReadCache.of({115 return PullRequestReadCache.of({
75 get: Effect.fn("PullRequestReadCache.get")(function* (key, lookup) {116 get: Effect.fn("PullRequestReadCache.get")(function* (key, lookup, scopes = []) {
76 if (!enabled) return yield* lookup;117 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);118 const read = yield* Effect.cached(lookup);
82 return yield* cache119 return yield* Effect.gen(function* () {
83 .get(new Read({ key: Encoding.encodeHex(digest.value), lookup: read }))120 const current = yield* Cache.get(revisions, undefined);
84 .pipe(121 const now = clock.currentTimeMillisUnsafe();
85 Effect.map((result) => result.payload),122 const revision = scopes
86 Effect.catchTags({123 .map((scope) => {
87 PersistenceError: () => read,124 const value = current[scope];
88 SchemaError: () => read,125 return value !== undefined && value.expiresAt > now ? value.revision : "";
89 }),126 })
90 Effect.uninterruptible,127 .join(":");
91 lock.withPermits(1),128 const request = new Read({ key: yield* digest(key), revision, lookup: read });
92 );129 const stored = yield* cache.get(request);
130 return (
131 request.matchesRevision(stored.revision) ? stored : yield* Cache.get(refreshes, request)
132 ).payload;
133 }).pipe(
134 Effect.catchTags({
135 PlatformError: () => read,
136 KeyValueStoreError: () => read,
137 PersistenceError: () => read,
138 SchemaError: () => read,
139 }),
140 lock.withPermits(1),
141 );
93 }),142 }),
94 // Let existing reads finish before clearing, so they cannot repopulate stale entries.143 invalidate: (scope) =>
95 invalidate: Cache.invalidateAll(cache.inMemory).pipe(144 Effect.gen(function* () {
96 Effect.andThen(backing.clear),145 const now = clock.currentTimeMillisUnsafe();
97 Effect.catch(() => {146 const current = yield* Cache.get(revisions, undefined);
98 enabled = false;147 const next = Object.fromEntries(
99 return Effect.logWarning("PR cache disabled after clearing failed");148 Object.entries(current).filter(([, value]) => value.expiresAt > now),
100 }),149 );
101 lock.withPermits(CONCURRENT_READS),150 next[scope] = { revision: yield* crypto.randomUUIDv4, expiresAt: now + 60_000 };
102 ),151 const encoded = yield* Schema.encodeEffect(revisionCodec)(next);
152 yield* backing.set("revisions", encoded);
153 yield* Cache.set(revisions, undefined, next);
154 }).pipe(
155 Effect.catch(() => {
156 enabled = false;
157 return Effect.logWarning("PR cache disabled after clearing failed");
158 }),
159 Effect.uninterruptible,
160 lock.withPermits(CONCURRENT_READS),
161 ),
103 });162 });
104});163});
105164
...@@ -108,7 +167,6 @@ export const layer = Layer.unwrap(...@@ -108,7 +167,6 @@ export const layer = Layer.unwrap(
108 const config = yield* ServerConfig;167 const config = yield* ServerConfig;
109 const path = yield* Path.Path;168 const path = yield* Path.Path;
110 return Layer.effect(PullRequestReadCache, make).pipe(169 return Layer.effect(PullRequestReadCache, make).pipe(
111 Layer.provide(Persistence.layerKvs),
112 Layer.provide(170 Layer.provide(
113 KeyValueStore.layerFileSystem(171 KeyValueStore.layerFileSystem(
114 path.join(config.providerStatusCacheDir, "pull-requests"),172 path.join(config.providerStatusCacheDir, "pull-requests"),
apps/server/src/pullRequest/PullRequestService.test.ts+227-98
...@@ -1,6 +1,5 @@...@@ -1,6 +1,5 @@
1import * as NodeServices from "@effect/platform-node/NodeServices";1import * as NodeServices from "@effect/platform-node/NodeServices";
2import * as KeyValueStore from "effect/unstable/persistence/KeyValueStore";2import * as KeyValueStore from "effect/unstable/persistence/KeyValueStore";
3import * as Persistence from "effect/unstable/persistence/Persistence";
4import { assert, it } from "@effect/vitest";3import { assert, it } from "@effect/vitest";
5import * as Deferred from "effect/Deferred";4import * as Deferred from "effect/Deferred";
6import * as Effect from "effect/Effect";5import * as Effect from "effect/Effect";
...@@ -204,7 +203,6 @@ function makeService(input: {...@@ -204,7 +203,6 @@ function makeService(input: {
204 }),203 }),
205 SourceControlRateLimit.layer,204 SourceControlRateLimit.layer,
206 Layer.effect(PullRequestReadCache.PullRequestReadCache, PullRequestReadCache.make).pipe(205 Layer.effect(PullRequestReadCache.PullRequestReadCache, PullRequestReadCache.make).pipe(
207 Layer.provide(Persistence.layerKvs),
208 Layer.provide(KeyValueStore.layerMemory),206 Layer.provide(KeyValueStore.layerMemory),
209 Layer.provide(NodeServices.layer),207 Layer.provide(NodeServices.layer),
210 ),208 ),
...@@ -3117,6 +3115,107 @@ it.effect("a listing narrowed to some projects is its own cache entry", () =>...@@ -3117,6 +3115,107 @@ it.effect("a listing narrowed to some projects is its own cache entry", () =>
3117 }),3115 }),
3118);3116);
31193117
3118it.effect("keeps unrelated PRs warm after a mutation, explicit refresh, and project turn", () =>
3119 Effect.gen(function* () {
3120 const calls: string[] = [];
3121 const service = yield* makeService({
3122 projects: [
3123 project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" }),
3124 project({ id: "p2", title: "docs", workspaceRoot: "/b", repository: "acme/docs" }),
3125 ],
3126 providers: [
3127 fakeProvider("github", {
3128 getChangeRequest: (input) =>
3129 Effect.sync(() => {
3130 calls.push(`${input.repository}/${input.number}`);
3131 return { ...hostedChangeRequest("body"), number: input.number };
3132 }),
3133 }),
3134 ],
3135 });
3136 const refs = [
3137 { projectId: "p1" as ProjectId, repository: "acme/web", number: 1 },
3138 { projectId: "p1" as ProjectId, repository: "acme/web", number: 2 },
3139 { projectId: "p2" as ProjectId, repository: "acme/docs", number: 3 },
3140 ];
3141 const readAll = Effect.forEach(refs, (ref) => service.summary({ ...ref, allowStale: false }));
3142 yield* readAll;
3143 yield* service.invalidate({ reference: { ...refs[0]!, host: "github.com" } });
3144 yield* readAll;
3145 assert.deepStrictEqual(calls, ["acme/web/1", "acme/web/2", "acme/docs/3", "acme/web/1"]);
3146 yield* service.comment({ ...refs[0]!, body: "hello" });
3147 yield* readAll;
3148 assert.deepStrictEqual(calls.slice(4), ["acme/web/1"]);
3149 yield* service.refreshAfterTurn("p1" as ProjectId);
3150 yield* readAll;
3151 assert.deepStrictEqual(calls.slice(5), ["acme/web/1", "acme/web/2"]);
3152 }),
3153);
3154
3155it.effect(
3156 "keeps matching PR numbers on different hosts separate and refreshes the serving project",
3157 () =>
3158 Effect.gen(function* () {
3159 const hosts: string[] = [];
3160 const service = yield* makeService({
3161 projects: [
3162 project({ id: "p1", title: "public", workspaceRoot: "/a", repository: "acme/web" }),
3163 project({
3164 id: "p2",
3165 title: "enterprise",
3166 workspaceRoot: "/b",
3167 repository: "acme/web",
3168 host: "enterprise.test",
3169 }),
3170 ],
3171 providers: [
3172 fakeProvider("github", {
3173 getChangeRequest: (input) =>
3174 Effect.sync(() => {
3175 hosts.push(input.host);
3176 return hostedChangeRequest("body");
3177 }),
3178 }),
3179 ],
3180 });
3181 const own = { projectId: "p1" as ProjectId, repository: "acme/web", number: 1 };
3182 const other = { ...own, host: "enterprise.test" };
3183 const readBoth = Effect.all([service.summary(own), service.summary(other)]);
3184 yield* readBoth;
3185 yield* service.invalidate({ reference: { ...own, host: "github.com" } });
3186 yield* readBoth;
3187 assert.deepStrictEqual(hosts, ["github.com", "enterprise.test", "github.com"]);
3188 yield* service.invalidate({ reference: own });
3189 yield* readBoth;
3190 assert.deepStrictEqual(hosts.slice(3), ["github.com"]);
3191 yield* service.refreshAfterTurn("p2" as ProjectId);
3192 yield* readBoth;
3193 assert.deepStrictEqual(hosts.slice(4), ["enterprise.test"]);
3194 }),
3195);
3196
3197it.effect("does not revive old summaries when project epochs are evicted", () =>
3198 Effect.gen(function* () {
3199 let title = "old";
3200 const service = yield* makeService({
3201 projects: [project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" })],
3202 providers: [
3203 fakeProvider("github", {
3204 getChangeRequest: () => Effect.succeed({ ...hostedChangeRequest("body"), title }),
3205 }),
3206 ],
3207 });
3208 const ref = { projectId: "p1" as ProjectId, repository: "acme/web", number: 1 };
3209 assert.strictEqual((yield* service.summary(ref))?.title, "old");
3210 title = "new";
3211 yield* service.refreshAfterTurn(ref.projectId);
3212 assert.strictEqual((yield* service.summary(ref))?.title, "new");
3213 for (let index = 0; index < 2048; index++)
3214 yield* service.refreshAfterTurn(`project-${index}` as ProjectId);
3215 assert.strictEqual((yield* service.summary(ref))?.title, "new");
3216 }),
3217);
3218
3120it.effect("explicit and turn invalidations make the next listing ask the host again", () =>3219it.effect("explicit and turn invalidations make the next listing ask the host again", () =>
3121 Effect.gen(function* () {3220 Effect.gen(function* () {
3122 let hostCalls = 0;3221 let hostCalls = 0;
...@@ -3148,7 +3247,7 @@ it.effect("explicit and turn invalidations make the next listing ask the host ag...@@ -3148,7 +3247,7 @@ it.effect("explicit and turn invalidations make the next listing ask the host ag
3148 yield* service.invalidate({ reference });3247 yield* service.invalidate({ reference });
3149 yield* service.list({ state: "open" });3248 yield* service.list({ state: "open" });
3150 assert.strictEqual(hostCalls, 2);3249 assert.strictEqual(hostCalls, 2);
3151 yield* service.refreshAfterTurn;3250 yield* service.refreshAfterTurn("p1" as ProjectId);
3152 const refresh = Option.getOrThrow(yield* Stream.runHead(service.subscribeRefreshes));3251 const refresh = Option.getOrThrow(yield* Stream.runHead(service.subscribeRefreshes));
3153 yield* service.list({ state: "open" });3252 yield* service.list({ state: "open" });
3154 assert.isAbove(refresh, 0);3253 assert.isAbove(refresh, 0);
...@@ -3551,7 +3650,7 @@ it.effect(...@@ -3551,7 +3650,7 @@ it.effect(
3551 yield* service.listStats({ refs: [ref(1), ref(2), ref(3)] });3650 yield* service.listStats({ refs: [ref(1), ref(2), ref(3)] });
3552 assert.deepStrictEqual(asked, [[1, 2], [3], [2], [1, 2, 3]]);3651 assert.deepStrictEqual(asked, [[1, 2], [3], [2], [1, 2, 3]]);
35533652
3554 yield* service.refreshAfterTurn;3653 yield* service.refreshAfterTurn("p1" as ProjectId);
3555 yield* service.listStats({ refs: [ref(1)] });3654 yield* service.listStats({ refs: [ref(1)] });
3556 assert.deepStrictEqual(asked, [[1, 2], [3], [2], [1, 2, 3], [1]]);3655 assert.deepStrictEqual(asked, [[1, 2], [3], [2], [1, 2, 3], [1]]);
35573656
...@@ -3768,9 +3867,9 @@ it.effect("shares linked summaries and reuses them for display without asking th...@@ -3768,9 +3867,9 @@ it.effect("shares linked summaries and reuses them for display without asking th
3768 }),3867 }),
3769);3868);
37703869
3771it.effect("keeps routed summaries and details separate when the GitHub account changes", () =>3870it.effect("keeps routed reads separate when the GitHub account changes", () =>
3772 Effect.gen(function* () {3871 Effect.gen(function* () {
3773 for (const operation of ["summary", "detail"] as const) {3872 for (const operation of ["summary", "detail", "diff"] as const) {
3774 let failing = false;3873 let failing = false;
3775 let calls = 0;3874 let calls = 0;
3776 const read = () =>3875 const read = () =>
...@@ -3785,16 +3884,27 @@ it.effect("keeps routed summaries and details separate when the GitHub account c...@@ -3785,16 +3884,27 @@ it.effect("keeps routed summaries and details separate when the GitHub account c
3785 project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" }),3884 project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" }),
3786 ],3885 ],
3787 providers: [3886 providers: [
3788 fakeProvider("github", { getChangeRequestSummary: read, getChangeRequest: read }),3887 fakeProvider("github", {
3888 getChangeRequestSummary: read,
3889 getChangeRequest: read,
3890 getDiff: () =>
3891 read().pipe(
3892 Effect.as({ patch: "private patch", truncated: false, nextCursor: null }),
3893 ),
3894 }),
3789 ],3895 ],
3790 });3896 });
3897 const readOperation = (input: Parameters<typeof service.diff>[0]) =>
3898 Effect.gen(function* () {
3899 yield* service[operation](input);
3900 });
3791 const reference = { projectId: "p1" as ProjectId, repository: "acme/web", number: 1 };3901 const reference = { projectId: "p1" as ProjectId, repository: "acme/web", number: 1 };
3792 yield* service[operation]({ ...reference, expectedAccountId: "101" });3902 yield* readOperation({ ...reference, expectedAccountId: "101" });
3793 failing = true;3903 failing = true;
37943904
3795 for (const allowStale of [false, true]) {3905 for (const allowStale of [false, true]) {
3796 const error = yield* Effect.flip(3906 const error = yield* Effect.flip(
3797 service[operation]({ ...reference, expectedAccountId: "202", allowStale }),3907 readOperation({ ...reference, expectedAccountId: "202", allowStale }),
3798 );3908 );
3799 assert.strictEqual(error._tag, "PullRequestOperationError");3909 assert.strictEqual(error._tag, "PullRequestOperationError");
3800 }3910 }
...@@ -3805,7 +3915,7 @@ it.effect("keeps routed summaries and details separate when the GitHub account c...@@ -3805,7 +3915,7 @@ it.effect("keeps routed summaries and details separate when the GitHub account c
38053915
3806it.effect("isolates routed caches for two credentials belonging to the same account", () =>3916it.effect("isolates routed caches for two credentials belonging to the same account", () =>
3807 Effect.gen(function* () {3917 Effect.gen(function* () {
3808 for (const operation of ["summary", "detail"] as const) {3918 for (const operation of ["summary", "detail", "diff"] as const) {
3809 let credential = "broad";3919 let credential = "broad";
3810 let calls = 0;3920 let calls = 0;
3811 const read = () =>3921 const read = () =>
...@@ -3831,9 +3941,17 @@ it.effect("isolates routed caches for two credentials belonging to the same acco...@@ -3831,9 +3941,17 @@ it.effect("isolates routed caches for two credentials belonging to the same acco
3831 ),3941 ),
3832 getChangeRequest: read,3942 getChangeRequest: read,
3833 getChangeRequestSummary: read,3943 getChangeRequestSummary: read,
3944 getDiff: () =>
3945 read().pipe(
3946 Effect.as({ patch: "private patch", truncated: false, nextCursor: null }),
3947 ),
3834 }),3948 }),
3835 ],3949 ],
3836 });3950 });
3951 const readOperation = (input: Parameters<typeof service.diff>[0]) =>
3952 Effect.gen(function* () {
3953 yield* service[operation](input);
3954 });
3837 const reference = {3955 const reference = {
3838 projectId: "p1" as ProjectId,3956 projectId: "p1" as ProjectId,
3839 repository: "acme/web",3957 repository: "acme/web",
...@@ -3841,14 +3959,11 @@ it.effect("isolates routed caches for two credentials belonging to the same acco...@@ -3841,14 +3959,11 @@ it.effect("isolates routed caches for two credentials belonging to the same acco
3841 host: "github.com",3959 host: "github.com",
3842 expectedAccountId: "101",3960 expectedAccountId: "101",
3843 };3961 };
3844 yield* service.withRoutingCredential(reference, service[operation](reference));3962 yield* service.withRoutingCredential(reference, readOperation(reference));
3845 credential = "restricted";3963 credential = "restricted";
3846 for (const allowStale of [false, true]) {3964 for (const allowStale of [false, true]) {
3847 const error = yield* Effect.flip(3965 const error = yield* Effect.flip(
3848 service.withRoutingCredential(3966 service.withRoutingCredential(reference, readOperation({ ...reference, allowStale })),
3849 reference,
3850 service[operation]({ ...reference, allowStale }),
3851 ),
3852 );3967 );
3853 assert.strictEqual(error._tag, "PullRequestOperationError");3968 assert.strictEqual(error._tag, "PullRequestOperationError");
3854 }3969 }
...@@ -4354,90 +4469,104 @@ it.effect('resolves an author filter of "me" to the viewer before narrowing a ho...@@ -4354,90 +4469,104 @@ it.effect('resolves an author filter of "me" to the viewer before narrowing a ho
4354 }),4469 }),
4355);4470);
43564471
4357it.effect("authorizes stack rebases independently of whether the selected layer is behind", () =>4472for (const crossHost of [false, true]) {
4358 Effect.gen(function* () {4473 it.effect(
4359 let taken = 0;4474 `authorizes stack rebases and refreshes sibling layers (cross-host: ${crossHost})`,
4360 let summaryReads = 0;4475 () =>
4361 let mutationFails = false;4476 Effect.gen(function* () {
4362 let stackRebase = true;4477 let taken = 0;
4363 let stackActions = true;4478 let summaryReads = 0;
4364 const capabilities = {4479 let mutationFails = false;
4365 diff: true,4480 let stackRebase = true;
4366 comment: true,4481 let stackActions = true;
4367 actions: ["update-branch"] as const,4482 const capabilities = {
4368 mergeMethods: ["merge"] as const,4483 diff: true,
4369 updateMethods: ["rebase"] as const,4484 comment: true,
4370 get stackActions() {4485 actions: ["update-branch"] as const,
4371 return stackActions;4486 mergeMethods: ["merge"] as const,
4372 },4487 updateMethods: ["rebase"] as const,
4373 search: true,4488 get stackActions() {
4374 reactions: true,4489 return stackActions;
4375 review: FULL_REVIEW,4490 },
4376 reviewers: FULL_REVIEWERS,4491 search: true,
4377 };4492 reactions: true,
4378 const service = yield* makeService({4493 review: FULL_REVIEW,
4379 projects: [project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" })],4494 reviewers: FULL_REVIEWERS,
4380 providers: [4495 };
4381 fakeProvider("github", {4496 const service = yield* makeService({
4382 capabilities,4497 projects: [
4383 getViewerPermissions: () =>4498 project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" }),
4384 Effect.succeed({4499 project({
4385 actions: [],4500 id: "p2",
4386 stackRebase,4501 title: "enterprise",
4387 comment: true,4502 workspaceRoot: "/b",
4388 resolve: false,4503 repository: "acme/web",
4389 verdicts: [],4504 host: "enterprise.test",
4390 requestReviewers: false,
4391 }),
4392 getChangeRequestSummary: () =>
4393 Effect.sync(() => {
4394 summaryReads++;
4395 return changeRequest(8, "2026-07-01T00:00:00Z");
4396 }),4505 }),
4397 runAction: () =>4506 ],
4398 Effect.gen(function* () {4507 providers: [
4399 taken++;4508 fakeProvider("github", {
4400 if (mutationFails) return yield* requestFailed;4509 capabilities,
4510 getViewerPermissions: () =>
4511 Effect.succeed({
4512 actions: [],
4513 stackRebase,
4514 comment: true,
4515 resolve: false,
4516 verdicts: [],
4517 requestReviewers: false,
4518 }),
4519 getChangeRequestSummary: () =>
4520 Effect.sync(() => {
4521 summaryReads++;
4522 return changeRequest(8, "2026-07-01T00:00:00Z");
4523 }),
4524 runAction: () =>
4525 Effect.gen(function* () {
4526 taken++;
4527 if (mutationFails) return yield* requestFailed;
4528 }),
4401 }),4529 }),
4402 }),4530 ],
4403 ],4531 });
4404 });4532 const input = {
4405 const input = {4533 ...(crossHost ? { host: "enterprise.test" } : {}),
4406 projectId: "p1" as ProjectId,4534 projectId: "p1" as ProjectId,
4407 repository: "acme/web",4535 repository: "acme/web",
4408 number: 3,4536 number: 3,
4409 action: "update-branch" as const,4537 action: "update-branch" as const,
4410 updateMethod: "rebase" as const,4538 updateMethod: "rebase" as const,
4411 stackNumber: 50,4539 stackNumber: 50,
4412 expectedStackHeads: [{ number: 3, headSha: "ccc" }],4540 expectedStackHeads: [{ number: 3, headSha: "ccc" }],
4413 };4541 };
4414 yield* service.runAction(input);4542 yield* service.runAction(input);
4415 assert.strictEqual(taken, 1);4543 assert.strictEqual(taken, 1);
4416 const unrelated = { ...input, number: 8 };4544 const unrelated = { ...input, number: 8 };
4417 yield* service.summary(unrelated);4545 yield* service.summary(unrelated);
4418 assert.strictEqual(summaryReads, 1);4546 assert.strictEqual(summaryReads, 1);
4419 stackRebase = false;4547 stackRebase = false;
4420 assert.strictEqual(4548 assert.strictEqual(
4421 (yield* Effect.flip(service.runAction(input)))._tag,4549 (yield* Effect.flip(service.runAction(input)))._tag,
4422 "PullRequestOperationError",4550 "PullRequestOperationError",
4423 );4551 );
4424 stackRebase = true;4552 stackRebase = true;
4425 stackActions = false;4553 stackActions = false;
4426 assert.strictEqual(4554 assert.strictEqual(
4427 (yield* Effect.flip(service.runAction(input)))._tag,4555 (yield* Effect.flip(service.runAction(input)))._tag,
4428 "PullRequestOperationError",4556 "PullRequestOperationError",
4429 );4557 );
4430 assert.strictEqual(taken, 1);4558 assert.strictEqual(taken, 1);
4431 yield* service.summary(unrelated);4559 yield* service.summary(unrelated);
4432 assert.strictEqual(summaryReads, 1);4560 assert.strictEqual(summaryReads, 1);
4433 stackActions = true;4561 stackActions = true;
4434 mutationFails = true;4562 mutationFails = true;
4435 yield* Effect.flip(service.runAction(input));4563 yield* Effect.flip(service.runAction(input));
4436 assert.strictEqual(taken, 2);4564 assert.strictEqual(taken, 2);
4437 yield* service.summary(unrelated);4565 yield* service.summary(unrelated);
4438 assert.strictEqual(summaryReads, 2);4566 assert.strictEqual(summaryReads, 2);
4439 }),4567 }),
4440);4568 );
4569}
44414570
4442it.effect("refuses a way of updating a branch that the host or the viewer does not allow", () =>4571it.effect("refuses a way of updating a branch that the host or the viewer does not allow", () =>
4443 Effect.gen(function* () {4572 Effect.gen(function* () {
...@@ -4773,7 +4902,7 @@ it.effect("forgets the cached detail after a rewrite or terminal turn", () =>...@@ -4773,7 +4902,7 @@ it.effect("forgets the cached detail after a rewrite or terminal turn", () =>
4773 yield* service.detail(reference);4902 yield* service.detail(reference);
4774 assert.strictEqual(coreCalls, 2);4903 assert.strictEqual(coreCalls, 2);
47754904
4776 yield* service.refreshAfterTurn;4905 yield* service.refreshAfterTurn("p1" as ProjectId);
4777 yield* service.detail(reference);4906 yield* service.detail(reference);
4778 assert.strictEqual(coreCalls, 3);4907 assert.strictEqual(coreCalls, 3);
4779 }),4908 }),
apps/server/src/pullRequest/PullRequestService.ts+99-60
...@@ -24,6 +24,7 @@ import {...@@ -24,6 +24,7 @@ import {
24 pullRequestProviderRequirement,24 pullRequestProviderRequirement,
25 resolvePullRequestAuthorFilter,25 resolvePullRequestAuthorFilter,
26 type OrchestrationProjectShell,26 type OrchestrationProjectShell,
27 type ProjectId,
27 type PullRequestAction,28 type PullRequestAction,
28 type PullRequestActionInput,29 type PullRequestActionInput,
29 type PullRequestActivity,30 type PullRequestActivity,
...@@ -67,6 +68,7 @@ import {...@@ -67,6 +68,7 @@ import {
67} from "@t3tools/contracts";68} from "@t3tools/contracts";
68import { detectSourceControlProviderFromRemoteUrl } from "@t3tools/shared/sourceControl";69import { detectSourceControlProviderFromRemoteUrl } from "@t3tools/shared/sourceControl";
6970
71import { AllowGitHubReserve } from "../sourceControl/GitHubCli.ts";
70import * as ProjectionSnapshotQuery from "../orchestration/Services/ProjectionSnapshotQuery.ts";72import * as ProjectionSnapshotQuery from "../orchestration/Services/ProjectionSnapshotQuery.ts";
71import * as SourceControlProviderRegistry from "../sourceControl/SourceControlProviderRegistry.ts";73import * as SourceControlProviderRegistry from "../sourceControl/SourceControlProviderRegistry.ts";
72import * as SourceControlRateLimit from "../sourceControl/SourceControlRateLimit.ts";74import * as SourceControlRateLimit from "../sourceControl/SourceControlRateLimit.ts";
...@@ -186,7 +188,7 @@ export class PullRequestService extends Context.Service<...@@ -186,7 +188,7 @@ export class PullRequestService extends Context.Service<
186 Scope.Scope188 Scope.Scope
187 >;189 >;
188 readonly subscribeRefreshes: Stream.Stream<number>;190 readonly subscribeRefreshes: Stream.Stream<number>;
189 readonly refreshAfterTurn: Effect.Effect<void>;191 readonly refreshAfterTurn: (projectId: ProjectId) => Effect.Effect<void>;
190 readonly detail: (input: PullRequestRef) => Effect.Effect<PullRequestDetail, PullRequestError>;192 readonly detail: (input: PullRequestRef) => Effect.Effect<PullRequestDetail, PullRequestError>;
191 readonly activity: (193 readonly activity: (
192 input: PullRequestRef,194 input: PullRequestRef,
...@@ -464,6 +466,7 @@ function withRateLimitBackoff(...@@ -464,6 +466,7 @@ function withRateLimitBackoff(
464 ),466 ),
465 Effect.flatMap((lease) =>467 Effect.flatMap((lease) =>
466 effect.pipe(468 effect.pipe(
469 Effect.provideService(AllowGitHubReserve, allowPaused),
467 Effect.tap(() => limits.recordSuccess({ ...key, lease })),470 Effect.tap(() => limits.recordSuccess({ ...key, lease })),
468 Effect.tapError((error) =>471 Effect.tapError((error) =>
469 error.reason === "rate-limited"472 error.reason === "rate-limited"
...@@ -794,6 +797,18 @@ export const make = Effect.gen(function* () {...@@ -794,6 +797,18 @@ export const make = Effect.gen(function* () {
794 }),797 }),
795 );798 );
796799
800 const canonicalRef = Effect.fn("PullRequestService.canonicalRef")(function* <
801 I extends PullRequestRef,
802 >(input: I) {
803 const project = yield* requireProject(input);
804 return {
805 ...input,
806 projectId: project.project.id,
807 host: project.host,
808 repository: project.repository,
809 };
810 });
811
797 /**812 /**
798 * What the signed-in account may do with this change request, asked of the host itself. Every813 * What the signed-in account may do with this change request, asked of the host itself. Every
799 * write goes through it: the page hides what a viewer may not do, and a request that arrived814 * write goes through it: the page hides what a viewer may not do, and a request that arrived
...@@ -1783,7 +1798,11 @@ export const make = Effect.gen(function* () {...@@ -1783,7 +1798,11 @@ export const make = Effect.gen(function* () {
1783 .pipe(1798 .pipe(
1784 // Once the authorized provider action starts, a failure may leave partial1799 // Once the authorized provider action starts, a failure may leave partial
1785 // remote updates. Validation and permission failures above changed nothing.1800 // remote updates. Validation and permission failures above changed nothing.
1786 Effect.ensuring(input.stackNumber === undefined ? Effect.void : refreshAfterTurn),1801 Effect.ensuring(
1802 input.stackNumber === undefined
1803 ? Effect.void
1804 : refreshAfterTurn(project.project.id),
1805 ),
1787 Effect.mapError(toPullRequestError("runAction")),1806 Effect.mapError(toPullRequestError("runAction")),
1788 Effect.as(1807 Effect.as(
1789 project.api.kind === "azure-devops"1808 project.api.kind === "azure-devops"
...@@ -2416,13 +2435,22 @@ export const make = Effect.gen(function* () {...@@ -2416,13 +2435,22 @@ export const make = Effect.gen(function* () {
2416 // scope re-entering `refEpochs` after eviction can never mint a key an old entry still has.2435 // scope re-entering `refEpochs` after eviction can never mint a key an old entry still has.
2417 let epochCounter = 0;2436 let epochCounter = 0;
2418 let listingsEpoch = 0;2437 let listingsEpoch = 0;
2419 let turnRefreshEpoch = 0;
2420 const refEpochs = new Map<string, number>();2438 const refEpochs = new Map<string, number>();
2439 const projectEpochs = new Map<ProjectId, number>();
2440 let projectEpochFloor = 0;
2421 const REF_EPOCH_CAPACITY = 2_048;2441 const REF_EPOCH_CAPACITY = 2_048;
2422 const refScope = (ref: PullRequestRef) =>2442 const refScope = (ref: PullRequestRef) =>
2423 `${ref.projectId} ${ref.host?.toLowerCase() ?? ""} ${ref.repository.toLowerCase()} ${ref.number}`;2443 JSON.stringify([
2444 ref.projectId,
2445 ref.host?.toLowerCase() ?? "",
2446 ref.repository.toLowerCase(),
2447 ref.number,
2448 ]);
2424 const refEpoch = (ref: PullRequestRef) =>2449 const refEpoch = (ref: PullRequestRef) =>
2425 Math.max(turnRefreshEpoch, refEpochs.get(refScope(ref)) ?? 0);2450 Math.max(
2451 projectEpochs.get(ref.projectId) ?? projectEpochFloor,
2452 refEpochs.get(refScope(ref)) ?? 0,
2453 );
2426 // Keys carry the reference back out of the cache loader, so the slot layout is shared with2454 // Keys carry the reference back out of the cache loader, so the slot layout is shared with
2427 // `refOfCacheKey` rather than read positionally at every loader.2455 // `refOfCacheKey` rather than read positionally at every loader.
2428 const refCacheKey = (ref: CredentialRef) =>2456 const refCacheKey = (ref: CredentialRef) =>
...@@ -2524,7 +2552,10 @@ export const make = Effect.gen(function* () {...@@ -2524,7 +2552,10 @@ export const make = Effect.gen(function* () {
2524 ),2552 ),
2525 ),2553 ),
2526 );2554 );
2527 const payload = yield* readCache.get(key, encodedRead);2555 const payload = yield* readCache.get(key, encodedRead, [
2556 `project:${input.projectId}`,
2557 refScope(input),
2558 ]);
2528 const decoded = yield* Schema.decodeUnknownEffect(codec)(payload).pipe(Effect.option);2559 const decoded = yield* Schema.decodeUnknownEffect(codec)(payload).pipe(Effect.option);
2529 return Option.isSome(decoded) ? decoded.value : yield* lookup;2560 return Option.isSome(decoded) ? decoded.value : yield* lookup;
2530 });2561 });
...@@ -2723,20 +2754,9 @@ export const make = Effect.gen(function* () {...@@ -2723,20 +2754,9 @@ export const make = Effect.gen(function* () {
27232754
2724 const diffCache = yield* Cache.makeWith(2755 const diffCache = yield* Cache.makeWith(
2725 (key: string) => {2756 (key: string) => {
2726 const [, projectId, host, repository, number, cursor, commit] = JSON.parse(key) as [2757 const [reference, cursor, commit] = JSON.parse(key) as [string, string | null, string | null];
2727 number,
2728 string,
2729 string | null,
2730 string,
2731 number,
2732 string | null,
2733 string | null,
2734 ];
2735 return diffUncached({2758 return diffUncached({
2736 projectId,2759 ...refOfCacheKey(reference),
2737 ...(host === null ? {} : { host }),
2738 repository,
2739 number,
2740 ...(cursor === null ? {} : { cursor }),2760 ...(cursor === null ? {} : { cursor }),
2741 ...(commit === null ? {} : { commit }),2761 ...(commit === null ? {} : { commit }),
2742 } as PullRequestDiffInput);2762 } as PullRequestDiffInput);
...@@ -2745,18 +2765,14 @@ export const make = Effect.gen(function* () {...@@ -2745,18 +2765,14 @@ export const make = Effect.gen(function* () {
2745 capacity: DIFF_CACHE_CAPACITY,2765 capacity: DIFF_CACHE_CAPACITY,
2746 timeToLive: (exit, key) => {2766 timeToLive: (exit, key) => {
2747 if (!Exit.isSuccess(exit)) return Duration.zero;2767 if (!Exit.isSuccess(exit)) return Duration.zero;
2748 const commit = (JSON.parse(key) as ReadonlyArray<unknown>)[6];2768 const commit = (JSON.parse(key) as ReadonlyArray<unknown>)[2];
2749 return commit === null ? DIFF_CACHE_TTL : COMMIT_DIFF_CACHE_TTL;2769 return commit === null ? DIFF_CACHE_TTL : COMMIT_DIFF_CACHE_TTL;
2750 },2770 },
2751 },2771 },
2752 );2772 );
2753 const diff: PullRequestService["Service"]["diff"] = (input) => {2773 const diff: PullRequestService["Service"]["diff"] = (input) => {
2754 const key = JSON.stringify([2774 const key = JSON.stringify([
2755 refEpoch(input),2775 refCacheKey(input),
2756 input.projectId,
2757 input.host?.toLowerCase() ?? null,
2758 input.repository.toLowerCase(),
2759 input.number,
2760 input.cursor ?? null,2776 input.cursor ?? null,
2761 input.commit ?? null,2777 input.commit ?? null,
2762 input.commit === undefined2778 input.commit === undefined
...@@ -2826,7 +2842,14 @@ export const make = Effect.gen(function* () {...@@ -2826,7 +2842,14 @@ export const make = Effect.gen(function* () {
2826 const invalidate: PullRequestService["Service"]["invalidate"] = (input) => {2842 const invalidate: PullRequestService["Service"]["invalidate"] = (input) => {
2827 const reference = input.reference;2843 const reference = input.reference;
2828 if (reference !== undefined) {2844 if (reference !== undefined) {
2829 return readCache.invalidate.pipe(Effect.andThen(Effect.sync(() => bumpRefEpoch(reference))));2845 return canonicalRef(reference).pipe(
2846 Effect.flatMap((ref) =>
2847 readCache
2848 .invalidate(refScope(ref))
2849 .pipe(Effect.andThen(Effect.sync(() => bumpRefEpoch(ref)))),
2850 ),
2851 Effect.ignore,
2852 );
2830 }2853 }
2831 return Effect.sync(() => {2854 return Effect.sync(() => {
2832 listingsEpoch = ++epochCounter;2855 listingsEpoch = ++epochCounter;
...@@ -2834,12 +2857,22 @@ export const make = Effect.gen(function* () {...@@ -2834,12 +2857,22 @@ export const make = Effect.gen(function* () {
2834 }).pipe(Effect.andThen(Cache.invalidateAll(viewerFlights)));2857 }).pipe(Effect.andThen(Cache.invalidateAll(viewerFlights)));
2835 };2858 };
28362859
2837 const refreshAfterTurn: PullRequestService["Service"]["refreshAfterTurn"] = Effect.suspend(() => {2860 const refreshAfterTurn: PullRequestService["Service"]["refreshAfterTurn"] = (projectId) =>
2838 turnRefreshEpoch = listingsEpoch = ++epochCounter;2861 Effect.suspend(() => {
2839 return readCache.invalidate.pipe(2862 listingsEpoch = ++epochCounter;
2840 Effect.andThen(SubscriptionRef.set(pullRequestRefreshes, turnRefreshEpoch)),2863 projectEpochs.delete(projectId);
2841 );2864 if (projectEpochs.size >= REF_EPOCH_CAPACITY) {
2842 });2865 const oldest = projectEpochs.keys().next().value;
2866 if (oldest !== undefined) {
2867 projectEpochFloor = projectEpochs.get(oldest)!;
2868 projectEpochs.delete(oldest);
2869 }
2870 }
2871 projectEpochs.set(projectId, listingsEpoch);
2872 return readCache
2873 .invalidate(`project:${projectId}`)
2874 .pipe(Effect.andThen(SubscriptionRef.set(pullRequestRefreshes, listingsEpoch)));
2875 });
28432876
2844 // A mutation's own client re-reads right after it, and every other client's next read must2877 // A mutation's own client re-reads right after it, and every other client's next read must
2845 // see the action too — so a write forgets the change request it touched and the listings its2878 // see the action too — so a write forgets the change request it touched and the listings its
...@@ -2849,22 +2882,28 @@ export const make = Effect.gen(function* () {...@@ -2849,22 +2882,28 @@ export const make = Effect.gen(function* () {
2849 method: (input: I) => Effect.Effect<void, PullRequestError>,2882 method: (input: I) => Effect.Effect<void, PullRequestError>,
2850 ): ((input: I) => Effect.Effect<void, PullRequestError>) =>2883 ): ((input: I) => Effect.Effect<void, PullRequestError>) =>
2851 (input) =>2884 (input) =>
2852 readCache.invalidate.pipe(2885 Effect.gen(function* () {
2853 Effect.andThen(method(input)),2886 const ref = yield* canonicalRef(input);
2854 Effect.ensuring(readCache.invalidate),2887 yield* readCache.invalidate(refScope(ref)).pipe(
2855 Effect.tap(() =>2888 Effect.andThen(method(input)),
2856 Effect.sync(() => {2889 Effect.ensuring(readCache.invalidate(refScope(ref))),
2857 bumpRefEpoch(input);2890 Effect.tap(() =>
2858 listingsEpoch = ++epochCounter;2891 Effect.sync(() => {
2859 }),2892 bumpRefEpoch(ref);
2860 ),2893 listingsEpoch = ++epochCounter;
2861 );2894 }),
2895 ),
2896 );
2897 });
2862 const runActionAndInvalidate: PullRequestService["Service"]["runAction"] = Effect.fn(2898 const runActionAndInvalidate: PullRequestService["Service"]["runAction"] = Effect.fn(
2863 "PullRequestService.runActionAndInvalidate",2899 "PullRequestService.runActionAndInvalidate",
2864 )(function* (input) {2900 )(function* (input) {
2865 yield* readCache.invalidate;2901 const ref = yield* canonicalRef(input);
2866 const repository = yield* runAction(input).pipe(Effect.ensuring(readCache.invalidate));2902 yield* readCache.invalidate(refScope(ref));
2867 bumpRefEpoch({ ...input, repository });2903 const repository = yield* runAction(input).pipe(
2904 Effect.ensuring(readCache.invalidate(refScope(ref))),
2905 );
2906 bumpRefEpoch({ ...ref, repository });
2868 listingsEpoch = ++epochCounter;2907 listingsEpoch = ++epochCounter;
2869 if (input.action === "merge") {2908 if (input.action === "merge") {
2870 // A successful merge action can merely enqueue the PR or enable auto-merge.2909 // A successful merge action can merely enqueue the PR or enable auto-merge.
...@@ -2890,26 +2929,26 @@ export const make = Effect.gen(function* () {...@@ -2890,26 +2929,26 @@ export const make = Effect.gen(function* () {
2890 read: (input: I, ...args: Args) => Effect.Effect<A, E>,2929 read: (input: I, ...args: Args) => Effect.Effect<A, E>,
2891 ) =>2930 ) =>
2892 (input: I, ...args: Args) =>2931 (input: I, ...args: Args) =>
2893 routingCredential.pipe(2932 Effect.gen(function* () {
2894 Effect.flatMap((credential) =>2933 const ref = yield* canonicalRef(input);
2895 read(2934 const credential = yield* routingCredential;
2896 credential === null2935 return yield* read(
2897 ? input2936 credential === null
2898 : {2937 ? ref
2899 ...input,2938 : { ...ref, [credentialNamespace]: credential.credentialFingerprint },
2900 [credentialNamespace]: credential.credentialFingerprint,2939 ...args,
2901 },2940 );
2902 ...args,2941 });
2903 ),
2904 ),
2905 );
29062942
2907 return PullRequestService.of({2943 return PullRequestService.of({
2908 routing,2944 routing,
2909 routingIdentity,2945 routingIdentity,
2910 withRoutingCredential,2946 withRoutingCredential,
2911 list,2947 list,
2912 listStats,2948 listStats: (input) =>
2949 Effect.forEach(input.refs, (ref) => canonicalRef(ref).pipe(Effect.option)).pipe(
2950 Effect.flatMap((refs) => listStats({ ...input, refs: refs.flatMap(Option.toArray) })),
2951 ),
2913 summary: credentialCached(summary),2952 summary: credentialCached(summary),
2914 stack: credentialCached(stack),2953 stack: credentialCached(stack),
2915 subscribeMerges: PubSub.subscribe(mergedPullRequests).pipe(2954 subscribeMerges: PubSub.subscribe(mergedPullRequests).pipe(
...@@ -2922,7 +2961,7 @@ export const make = Effect.gen(function* () {...@@ -2922,7 +2961,7 @@ export const make = Effect.gen(function* () {
2922 detail: credentialCached(detail),2961 detail: credentialCached(detail),
2923 activity: credentialCached(activity),2962 activity: credentialCached(activity),
2924 threadComments,2963 threadComments,
2925 diff,2964 diff: credentialCached(diff),
2926 diffFileContents,2965 diffFileContents,
2927 runAction: runActionAndInvalidate,2966 runAction: runActionAndInvalidate,
2928 update: invalidatedByMutation(update),2967 update: invalidatedByMutation(update),
apps/server/src/sourceControl/GitHubCli.test.ts+112-1
...@@ -1,5 +1,8 @@...@@ -1,5 +1,8 @@
1import { assert, it, afterEach, describe, expect, vi } from "@effect/vitest";1import { assert, it, afterEach, describe, expect, vi } from "@effect/vitest";
2import * as Cache from "effect/Cache";2import * as Cache from "effect/Cache";
3import * as TestClock from "effect/testing/TestClock";
4import * as Clock from "effect/Clock";
5import * as DateTime from "effect/DateTime";
3import * as Effect from "effect/Effect";6import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";7import * as Layer from "effect/Layer";
5import * as PlatformError from "effect/PlatformError";8import * as PlatformError from "effect/PlatformError";
...@@ -10,6 +13,8 @@ import { VcsProcessExitError, VcsProcessSpawnError } from "@t3tools/contracts";...@@ -10,6 +13,8 @@ import { VcsProcessExitError, VcsProcessSpawnError } from "@t3tools/contracts";
1013
11import * as VcsProcess from "../vcs/VcsProcess.ts";14import * as VcsProcess from "../vcs/VcsProcess.ts";
12import * as GitHubCli from "./GitHubCli.ts";15import * as GitHubCli from "./GitHubCli.ts";
16import * as GitHubGraphQlBudget from "./githubGraphQlBudget.ts";
17import * as SourceControlRateLimit from "./SourceControlRateLimit.ts";
1318
14const encodeGitHubCliError = Schema.encodeEffect(Schema.fromJsonString(GitHubCli.GitHubCliError));19const encodeGitHubCliError = Schema.encodeEffect(Schema.fromJsonString(GitHubCli.GitHubCliError));
1520
...@@ -21,12 +26,18 @@ const processOutput = (stdout: string): VcsProcess.VcsProcessOutput => ({...@@ -21,12 +26,18 @@ const processOutput = (stdout: string): VcsProcess.VcsProcessOutput => ({
21 stderrTruncated: false,26 stderrTruncated: false,
22});27});
2328
29const quotaOutput = (remaining = 5000, resetAt = "2099-01-01T00:00:00Z") =>
30 processOutput(
31 JSON.stringify({ data: { rateLimit: { cost: 1, limit: 5000, remaining, resetAt } } }),
32 );
33
24const mockRun = vi.fn<VcsProcess.VcsProcess["Service"]["run"]>();34const mockRun = vi.fn<VcsProcess.VcsProcess["Service"]["run"]>();
2535
26const layer = GitHubCli.layer.pipe(36const layer = GitHubCli.layer.pipe(
27 Layer.provide(37 Layer.provide(
28 Layer.mock(VcsProcess.VcsProcess)({38 Layer.mock(VcsProcess.VcsProcess)({
29 run: mockRun,39 run: (input) =>
40 input.args[1] === "rate_limit" ? Effect.succeed(quotaOutput()) : mockRun(input),
30 }),41 }),
31 ),42 ),
32);43);
...@@ -35,7 +46,98 @@ afterEach(() => {...@@ -35,7 +46,98 @@ afterEach(() => {
35 mockRun.mockReset();46 mockRun.mockReset();
36});47});
3748
49it.effect("shares quota checks, preserves the reserve, and resumes after reset", () =>
50 Effect.gen(function* () {
51 let probes = 0;
52 const commands: string[] = [];
53 let remaining = 501;
54 let resetAt = DateTime.formatIso(
55 DateTime.makeUnsafe((yield* Clock.currentTimeMillis) + 60_000),
56 );
57 const gh = yield* GitHubCli.make.pipe(
58 Effect.provideService(VcsProcess.VcsProcess, {
59 run: (input) =>
60 Effect.sync(() => {
61 if (input.args[1] === "rate_limit") {
62 probes++;
63 assert.strictEqual(input.args[3], "enterprise.test");
64 return quotaOutput(remaining, resetAt);
65 }
66 commands.push(input.args.slice(0, 2).join(" "));
67 return processOutput("[]");
68 }),
69 }),
70 );
71 const read = (command: string) =>
72 gh.execute({
73 cwd: "/repo",
74 args:
75 command === "repo"
76 ? ["repo", "view", "enterprise.test/acme/web", "--json", "name"]
77 : ["pr", command, "--repo=enterprise.test/acme/web", "--json", "number"],
78 });
79 yield* read("list");
80 const failure = yield* read("view").pipe(Effect.flip);
81 assert.strictEqual(failure._tag, "GitHubCliRateLimitError");
82 assert.strictEqual(probes, 1);
83 assert.deepStrictEqual(commands, ["pr list"]);
84 yield* read("view").pipe(Effect.provideService(GitHubCli.AllowGitHubReserve, true));
85 yield* gh.execute({ cwd: "/repo", args: ["pr", "merge", "1"] });
86 assert.deepStrictEqual(commands, ["pr list", "pr view", "pr merge"]);
87 remaining = 0;
88 yield* TestClock.adjust("30 seconds");
89 yield* read("repo").pipe(Effect.flip);
90 assert.strictEqual(probes, 2);
91 yield* TestClock.adjust("30 seconds");
92 remaining = 5000;
93 resetAt = DateTime.formatIso(DateTime.makeUnsafe((yield* Clock.currentTimeMillis) + 60_000));
94 yield* Effect.all([read("list"), read("repo")], { concurrency: 2 });
95 assert.strictEqual(probes, 3);
96 assert.deepStrictEqual(commands.slice(3).toSorted(), ["pr list", "repo view"]);
97 }).pipe(Effect.provide(Layer.merge(GitHubGraphQlBudget.layer, SourceControlRateLimit.layer))),
98);
99
38describe("GitHubCli.layer", () => {100describe("GitHubCli.layer", () => {
101 it.effect("shares the registry budget with CLI reads through nested layer providers", () =>
102 Effect.gen(function* () {
103 const budget = yield* GitHubGraphQlBudget.GitHubGraphQlBudget;
104 const gh = yield* GitHubCli.GitHubCli;
105 yield* budget.observe("github.com", quotaOutput(0).stdout);
106 const error = yield* gh.execute({ cwd: "/repo", args: ["pr", "list"] }).pipe(Effect.flip);
107 assert.strictEqual(error._tag, "GitHubCliRateLimitError");
108 expect(mockRun).not.toHaveBeenCalled();
109 }).pipe(Effect.provide(layer.pipe(Layer.provide(GitHubGraphQlBudget.layer)))),
110 );
111
112 it.effect("keeps quota snapshots separate for verified credentials on the same host", () =>
113 Effect.gen(function* () {
114 let reads = 0;
115 const gh = yield* GitHubCli.make.pipe(
116 Effect.provideService(VcsProcess.VcsProcess, {
117 run: (input) =>
118 Effect.sync(() => {
119 if (input.args[1] === "rate_limit")
120 return quotaOutput(input.env?.GH_TOKEN === "empty" ? 0 : 5000);
121 reads++;
122 return processOutput("[]");
123 }),
124 }),
125 );
126 const read = (token: string) =>
127 gh.execute({ cwd: "/repo", args: ["pr", "list", "--repo", "github.com/acme/web"] }).pipe(
128 Effect.provideService(GitHubCli.PinnedGitHubCredential, {
129 host: "github.com",
130 token: Redacted.make(token),
131 credentialFingerprint: token,
132 }),
133 );
134 yield* read("empty").pipe(Effect.flip);
135 yield* read("healthy");
136 yield* read("empty").pipe(Effect.flip);
137 assert.strictEqual(reads, 1);
138 }).pipe(Effect.provide(Layer.merge(GitHubGraphQlBudget.layer, SourceControlRateLimit.layer))),
139 );
140
39 it.effect("pins concurrent cached commands to their own verified credentials", () =>141 it.effect("pins concurrent cached commands to their own verified credentials", () =>
40 Effect.gen(function* () {142 Effect.gen(function* () {
41 mockRun.mockImplementation((input) =>143 mockRun.mockImplementation((input) =>
...@@ -523,6 +625,15 @@ describe("GitHubCli.layer", () => {...@@ -523,6 +625,15 @@ describe("GitHubCli.layer", () => {
523 assert.include(error.detail, "gh api rate_limit");625 assert.include(error.detail, "gh api rate_limit");
524 assert.strictEqual(error.cause, cause);626 assert.strictEqual(error.cause, cause);
525 assert.notInclude(error.message, "user ID");627 assert.notInclude(error.message, "user ID");
628 const paused = yield* gh
629 .execute({ cwd: "/other-repo", args: ["pr", "list"] })
630 .pipe(Effect.flip);
631 assert.strictEqual(paused._tag, "GitHubCliRateLimitError");
632 expect(mockRun).toHaveBeenCalledTimes(1);
633 yield* TestClock.adjust("30 seconds");
634 mockRun.mockReturnValueOnce(Effect.succeed(processOutput("[]")));
635 yield* gh.execute({ cwd: "/other-repo", args: ["pr", "list"] });
636 expect(mockRun).toHaveBeenCalledTimes(2);
526 }).pipe(Effect.provide(layer)),637 }).pipe(Effect.provide(layer)),
527 );638 );
528});639});
apps/server/src/sourceControl/GitHubCli.ts+115-4
...@@ -1,3 +1,6 @@...@@ -1,3 +1,6 @@
1import * as Cache from "effect/Cache";
2import * as Duration from "effect/Duration";
3import * as Exit from "effect/Exit";
1import * as Context from "effect/Context";4import * as Context from "effect/Context";
2import * as DateTime from "effect/DateTime";5import * as DateTime from "effect/DateTime";
3import * as Effect from "effect/Effect";6import * as Effect from "effect/Effect";
...@@ -15,6 +18,8 @@ import {...@@ -15,6 +18,8 @@ import {
15} from "@t3tools/contracts";18} from "@t3tools/contracts";
1619
17import * as VcsProcess from "../vcs/VcsProcess.ts";20import * as VcsProcess from "../vcs/VcsProcess.ts";
21import * as GitHubGraphQlBudget from "./githubGraphQlBudget.ts";
22import * as SourceControlRateLimit from "./SourceControlRateLimit.ts";
18import {23import {
19 decodeGitHubPullRequestJson,24 decodeGitHubPullRequestJson,
20 decodeGitHubPullRequestListJson,25 decodeGitHubPullRequestListJson,
...@@ -30,7 +35,12 @@ export const PinnedGitHubCredential = Context.Reference<{...@@ -30,7 +35,12 @@ export const PinnedGitHubCredential = Context.Reference<{
30 readonly credentialFingerprint: string;35 readonly credentialFingerprint: string;
31} | null>("t3/sourceControl/PinnedGitHubCredential", { defaultValue: () => null });36} | null>("t3/sourceControl/PinnedGitHubCredential", { defaultValue: () => null });
3237
33function targetsVerifiedHost(args: ReadonlyArray<string>, host: string): boolean {38export const AllowGitHubReserve = Context.Reference<boolean>(
39 "t3/sourceControl/AllowGitHubReserve",
40 { defaultValue: () => false },
41);
42
43function commandHosts(args: ReadonlyArray<string>): Array<string | null> {
34 const hosts: Array<string | null> = [];44 const hosts: Array<string | null> = [];
35 const repositoryHost = (repository: string | undefined) => {45 const repositoryHost = (repository: string | undefined) => {
36 if (repository === undefined) return null;46 if (repository === undefined) return null;
...@@ -54,6 +64,11 @@ function targetsVerifiedHost(args: ReadonlyArray<string>, host: string): boolean...@@ -54,6 +64,11 @@ function targetsVerifiedHost(args: ReadonlyArray<string>, host: string): boolean
54 else if (arg.startsWith("-R")) hosts.push(repositoryHost(arg.slice(2)));64 else if (arg.startsWith("-R")) hosts.push(repositoryHost(arg.slice(2)));
55 else if (/^https?:\/\//i.test(arg)) hosts.push(repositoryHost(arg));65 else if (/^https?:\/\//i.test(arg)) hosts.push(repositoryHost(arg));
56 }66 }
67 return hosts;
68}
69
70function targetsVerifiedHost(args: ReadonlyArray<string>, host: string): boolean {
71 const hosts = commandHosts(args);
57 return hosts.length > 0 && hosts.every((target) => target === host);72 return hosts.length > 0 && hosts.every((target) => target === host);
58}73}
5974
...@@ -91,7 +106,7 @@ export class GitHubCliAuthenticationError extends Schema.TaggedError<GitHubCliAu...@@ -91,7 +106,7 @@ export class GitHubCliAuthenticationError extends Schema.TaggedError<GitHubCliAu
91106
92export class GitHubCliRateLimitError extends Schema.TaggedError<GitHubCliRateLimitError>()(107export class GitHubCliRateLimitError extends Schema.TaggedError<GitHubCliRateLimitError>()(
93 "GitHubCliRateLimitError",108 "GitHubCliRateLimitError",
94 gitHubCliFailureFields,109 { ...gitHubCliFailureFields, retryAt: Schema.optionalKey(Schema.Finite) },
95) {110) {
96 get detail(): string {111 get detail(): string {
97 return "GitHub API rate limit exceeded. Run `gh api rate_limit` to inspect the quota and reset time.";112 return "GitHub API rate limit exceeded. Run `gh api rate_limit` to inspect the quota and reset time.";
...@@ -274,17 +289,21 @@ export class GitHubCli extends Context.Service<...@@ -274,17 +289,21 @@ export class GitHubCli extends Context.Service<
274 readonly stdin?: string;289 readonly stdin?: string;
275 readonly env?: NodeJS.ProcessEnv;290 readonly env?: NodeJS.ProcessEnv;
276 readonly maxOutputBytes?: number;291 readonly maxOutputBytes?: number;
292 readonly rateLimitHost?: string;
293 readonly allowReserve?: boolean;
277 }) => Effect.Effect<VcsProcess.VcsProcessOutput, GitHubCliError>;294 }) => Effect.Effect<VcsProcess.VcsProcessOutput, GitHubCliError>;
278295
279 readonly listOpenPullRequests: (input: {296 readonly listOpenPullRequests: (input: {
280 readonly cwd: string;297 readonly cwd: string;
281 readonly headSelector: string;298 readonly headSelector: string;
282 readonly limit?: number;299 readonly limit?: number;
300 readonly rateLimitHost?: string;
283 }) => Effect.Effect<ReadonlyArray<GitHubPullRequestSummary>, GitHubCliError>;301 }) => Effect.Effect<ReadonlyArray<GitHubPullRequestSummary>, GitHubCliError>;
284302
285 readonly getPullRequest: (input: {303 readonly getPullRequest: (input: {
286 readonly cwd: string;304 readonly cwd: string;
287 readonly reference: string;305 readonly reference: string;
306 readonly rateLimitHost?: string;
288 }) => Effect.Effect<GitHubPullRequestSummary, GitHubCliError>;307 }) => Effect.Effect<GitHubPullRequestSummary, GitHubCliError>;
289308
290 readonly getRepositoryCloneUrls: (input: {309 readonly getRepositoryCloneUrls: (input: {
...@@ -308,6 +327,7 @@ export class GitHubCli extends Context.Service<...@@ -308,6 +327,7 @@ export class GitHubCli extends Context.Service<
308327
309 readonly getDefaultBranch: (input: {328 readonly getDefaultBranch: (input: {
310 readonly cwd: string;329 readonly cwd: string;
330 readonly rateLimitHost?: string;
311 }) => Effect.Effect<string | null, GitHubCliError>;331 }) => Effect.Effect<string | null, GitHubCliError>;
312332
313 readonly checkoutPullRequest: (input: {333 readonly checkoutPullRequest: (input: {
...@@ -377,8 +397,10 @@ function deriveRepositoryCloneUrlsFromCreateOutput(...@@ -377,8 +397,10 @@ function deriveRepositoryCloneUrlsFromCreateOutput(
377/** @public Service construction is part of the canonical Effect module API. */397/** @public Service construction is part of the canonical Effect module API. */
378export const make = Effect.gen(function* () {398export const make = Effect.gen(function* () {
379 const process = yield* VcsProcess.VcsProcess;399 const process = yield* VcsProcess.VcsProcess;
400 const budget = yield* GitHubGraphQlBudget.GitHubGraphQlBudget;
401 const limits = yield* SourceControlRateLimit.SourceControlRateLimit;
380402
381 const execute: GitHubCli["Service"]["execute"] = Effect.fn("GitHubCli.execute")(403 const executeRaw: GitHubCli["Service"]["execute"] = Effect.fn("GitHubCli.executeRaw")(
382 function* (input) {404 function* (input) {
383 const credential = yield* PinnedGitHubCredential;405 const credential = yield* PinnedGitHubCredential;
384 if (credential !== null && !targetsVerifiedHost(input.args, credential.host)) {406 if (credential !== null && !targetsVerifiedHost(input.args, credential.host)) {
...@@ -416,11 +438,94 @@ export const make = Effect.gen(function* () {...@@ -416,11 +438,94 @@ export const make = Effect.gen(function* () {
416 },438 },
417 );439 );
418440
441 const quota = yield* Cache.makeWith(
442 (key: string) => {
443 const host = key.split("\0")[0]!;
444 return executeRaw({
445 cwd: globalThis.process.cwd(),
446 args: [
447 "api",
448 "rate_limit",
449 "--hostname",
450 host,
451 "--jq",
452 ".resources.graphql | {data:{rateLimit:{cost:1,limit:.limit,remaining:.remaining,resetAt:(.reset|todateiso8601)}}}",
453 ],
454 }).pipe(
455 Effect.tap((result) => budget.observe(host, result.stdout)),
456 Effect.asVoid,
457 );
458 },
459 {
460 capacity: 32,
461 timeToLive: (exit) => (Exit.isSuccess(exit) ? Duration.seconds(30) : Duration.zero),
462 },
463 );
464 const execute: GitHubCli["Service"]["execute"] = Effect.fn("GitHubCli.execute")(
465 function* (input) {
466 const [command, action] = input.args;
467 if (
468 !(
469 (command === "pr" && (action === "list" || action === "view")) ||
470 (command === "repo" && action === "view")
471 )
472 )
473 return yield* executeRaw(input);
474 const credential = yield* PinnedGitHubCredential;
475 if (credential !== null && !targetsVerifiedHost(input.args, credential.host))
476 return yield* executeRaw(input);
477 const allowReserve = input.allowReserve ?? (yield* AllowGitHubReserve);
478 const host = (
479 credential?.host ??
480 commandHosts(input.args).find((host) => host !== null) ??
481 input.rateLimitHost ??
482 input.env?.GH_HOST ??
483 globalThis.process.env.GH_HOST ??
484 "github.com"
485 ).toLowerCase();
486 const key = { provider: "github" as const, host };
487 const guarded = Effect.gen(function* () {
488 const lease = yield* limits.check(key, allowReserve ? { allowPaused: true } : undefined);
489 return yield* Effect.gen(function* () {
490 yield* Cache.get(quota, `${host}\0${credential?.credentialFingerprint ?? ""}`);
491 yield* budget.query(host, "query {}", allowReserve ? { allowReserve: true } : undefined);
492 return yield* executeRaw(input);
493 }).pipe(
494 Effect.tap(() => limits.recordSuccess({ ...key, lease })),
495 Effect.tapError((error) =>
496 error._tag === "GitHubCliRateLimitError"
497 ? limits.recordRateLimit({ ...key, lease })
498 : Effect.void,
499 ),
500 );
501 });
502 return yield* guarded.pipe(
503 Effect.provideService(
504 SourceControlRateLimit.CredentialScope,
505 credential?.credentialFingerprint ?? (yield* SourceControlRateLimit.CredentialScope),
506 ),
507 Effect.catchTags({
508 SourceControlRateLimitPausedError: (cause) =>
509 Effect.fail(
510 new GitHubCliRateLimitError({
511 command: "gh",
512 cwd: input.cwd,
513 retryAt: cause.retryAt,
514 cause,
515 }),
516 ),
517 }),
518 );
519 },
520 );
521
419 return GitHubCli.of({522 return GitHubCli.of({
420 execute,523 execute,
421 listOpenPullRequests: (input) =>524 listOpenPullRequests: (input) =>
422 execute({525 execute({
423 cwd: input.cwd,526 cwd: input.cwd,
527 ...(input.rateLimitHost === undefined ? {} : { rateLimitHost: input.rateLimitHost }),
528 allowReserve: true,
424 args: [529 args: [
425 "pr",530 "pr",
426 "list",531 "list",
...@@ -458,6 +563,8 @@ export const make = Effect.gen(function* () {...@@ -458,6 +563,8 @@ export const make = Effect.gen(function* () {
458 getPullRequest: (input) =>563 getPullRequest: (input) =>
459 execute({564 execute({
460 cwd: input.cwd,565 cwd: input.cwd,
566 ...(input.rateLimitHost === undefined ? {} : { rateLimitHost: input.rateLimitHost }),
567 allowReserve: true,
461 args: [568 args: [
462 "pr",569 "pr",
463 "view",570 "view",
...@@ -533,6 +640,7 @@ export const make = Effect.gen(function* () {...@@ -533,6 +640,7 @@ export const make = Effect.gen(function* () {
533 getDefaultBranch: (input) =>640 getDefaultBranch: (input) =>
534 execute({641 execute({
535 cwd: input.cwd,642 cwd: input.cwd,
643 ...(input.rateLimitHost === undefined ? {} : { rateLimitHost: input.rateLimitHost }),
536 args: ["repo", "view", "--json", "defaultBranchRef", "--jq", ".defaultBranchRef.name"],644 args: ["repo", "view", "--json", "defaultBranchRef", "--jq", ".defaultBranchRef.name"],
537 }).pipe(645 }).pipe(
538 Effect.map((value) => {646 Effect.map((value) => {
...@@ -548,4 +656,7 @@ export const make = Effect.gen(function* () {...@@ -548,4 +656,7 @@ export const make = Effect.gen(function* () {
548 });656 });
549});657});
550658
551export const layer = Layer.effect(GitHubCli, make);659export const layer = Layer.effect(GitHubCli, make).pipe(
660 Layer.provideMerge(GitHubGraphQlBudget.layer),
661 Layer.provideMerge(SourceControlRateLimit.layer),
662);
apps/server/src/sourceControl/GitHubSourceControlProvider.test.ts+27
...@@ -30,6 +30,33 @@ function makeProvider(github: Partial<GitHubCli.GitHubCli["Service"]>) {...@@ -30,6 +30,33 @@ function makeProvider(github: Partial<GitHubCli.GitHubCli["Service"]>) {
30 );30 );
31}31}
3232
33it.effect("uses the enterprise quota for a current-repository default branch read", () =>
34 Effect.gen(function* () {
35 const provider = yield* GitHubSourceControlProvider.make.pipe(
36 Effect.provide(GitHubCli.layer),
37 Effect.provideService(VcsProcess.VcsProcess, {
38 run: (input) =>
39 Effect.sync(() => {
40 if (input.args[1] !== "rate_limit") return processResult("main");
41 assert.strictEqual(input.args[3], "enterprise.test");
42 return processResult(
43 '{"data":{"rateLimit":{"cost":1,"limit":5000,"remaining":5000,"resetAt":"2099-01-01T00:00:00Z"}}}',
44 );
45 }),
46 }),
47 );
48 const branch = yield* provider.getDefaultBranch({
49 cwd: "/enterprise-repo",
50 context: {
51 provider: { kind: "github", name: "GitHub Enterprise", baseUrl: "https://enterprise.test" },
52 remoteName: "origin",
53 remoteUrl: "https://enterprise.test/acme/web.git",
54 },
55 });
56 assert.strictEqual(branch, "main");
57 }),
58);
59
33it.effect("maps GitHub PR summaries into provider-neutral change requests", () =>60it.effect("maps GitHub PR summaries into provider-neutral change requests", () =>
34 Effect.gen(function* () {61 Effect.gen(function* () {
35 const provider = yield* makeProvider({62 const provider = yield* makeProvider({
apps/server/src/sourceControl/GitHubSourceControlProvider.ts+48-28
...@@ -127,6 +127,9 @@ export const make = Effect.gen(function* () {...@@ -127,6 +127,9 @@ export const make = Effect.gen(function* () {
127 .listOpenPullRequests({127 .listOpenPullRequests({
128 cwd: input.cwd,128 cwd: input.cwd,
129 headSelector: input.headSelector,129 headSelector: input.headSelector,
130 ...(input.context === undefined
131 ? {}
132 : { rateLimitHost: new URL(input.context.provider.baseUrl).host }),
130 ...(input.limit !== undefined ? { limit: input.limit } : {}),133 ...(input.limit !== undefined ? { limit: input.limit } : {}),
131 })134 })
132 .pipe(135 .pipe(
...@@ -152,6 +155,9 @@ export const make = Effect.gen(function* () {...@@ -152,6 +155,9 @@ export const make = Effect.gen(function* () {
152 return github155 return github
153 .execute({156 .execute({
154 cwd: input.cwd,157 cwd: input.cwd,
158 ...(input.context === undefined
159 ? {}
160 : { rateLimitHost: new URL(input.context.provider.baseUrl).host }),
155 args: [161 args: [
156 "pr",162 "pr",
157 "list",163 "list",
...@@ -267,23 +273,30 @@ export const make = Effect.gen(function* () {...@@ -267,23 +273,30 @@ export const make = Effect.gen(function* () {
267 },273 },
268 listChangeRequests,274 listChangeRequests,
269 getChangeRequest: (input) =>275 getChangeRequest: (input) =>
270 github.getPullRequest(input).pipe(276 github
271 Effect.map(toChangeRequest),277 .getPullRequest({
272 Effect.mapError(278 ...input,
273 (error) =>279 ...(input.context === undefined
274 new SourceControlProviderError({280 ? {}
275 provider: "github",281 : { rateLimitHost: new URL(input.context.provider.baseUrl).host }),
276 operation: "getChangeRequest",282 })
277 command: error.command,283 .pipe(
278 cwd: input.cwd,284 Effect.map(toChangeRequest),
279 reference: SourceControlProvider.transportSafeSourceControlErrorValue(285 Effect.mapError(
280 input.reference,286 (error) =>
281 ),287 new SourceControlProviderError({
282 detail: error.detail,288 provider: "github",
283 cause: error,289 operation: "getChangeRequest",
284 }),290 command: error.command,
291 cwd: input.cwd,
292 reference: SourceControlProvider.transportSafeSourceControlErrorValue(
293 input.reference,
294 ),
295 detail: error.detail,
296 cause: error,
297 }),
298 ),
285 ),299 ),
286 ),
287 createChangeRequest: (input) =>300 createChangeRequest: (input) =>
288 github301 github
289 .createPullRequest({302 .createPullRequest({
...@@ -344,19 +357,26 @@ export const make = Effect.gen(function* () {...@@ -344,19 +357,26 @@ export const make = Effect.gen(function* () {
344 ),357 ),
345 ),358 ),
346 getDefaultBranch: (input) =>359 getDefaultBranch: (input) =>
347 github.getDefaultBranch(input).pipe(360 github
348 Effect.mapError(361 .getDefaultBranch({
349 (error) =>362 ...input,
350 new SourceControlProviderError({363 ...(input.context === undefined
351 provider: "github",364 ? {}
352 operation: "getDefaultBranch",365 : { rateLimitHost: new URL(input.context.provider.baseUrl).host }),
353 command: error.command,366 })
354 cwd: input.cwd,367 .pipe(
355 detail: error.detail,368 Effect.mapError(
356 cause: error,369 (error) =>
357 }),370 new SourceControlProviderError({
371 provider: "github",
372 operation: "getDefaultBranch",
373 command: error.command,
374 cwd: input.cwd,
375 detail: error.detail,
376 cause: error,
377 }),
378 ),
358 ),379 ),
359 ),
360 checkoutChangeRequest: (input) =>380 checkoutChangeRequest: (input) =>
361 github.checkoutPullRequest(input).pipe(381 github.checkoutPullRequest(input).pipe(
362 Effect.mapError(382 Effect.mapError(
apps/server/src/sourceControl/SourceControlRateLimit.test.ts+16
...@@ -6,6 +6,22 @@ import * as SourceControlRateLimit from "./SourceControlRateLimit.ts";...@@ -6,6 +6,22 @@ import * as SourceControlRateLimit from "./SourceControlRateLimit.ts";
66
7const github = { provider: "github" as const, host: "github.com" };7const github = { provider: "github" as const, host: "github.com" };
88
9it.effect("isolates cooldowns for verified credentials on the same host", () =>
10 Effect.gen(function* () {
11 const limits = yield* SourceControlRateLimit.SourceControlRateLimit;
12 yield* limits
13 .recordRateLimit({ ...github, lease: 0 })
14 .pipe(Effect.provideService(SourceControlRateLimit.CredentialScope, "first"));
15 yield* limits
16 .check(github)
17 .pipe(Effect.provideService(SourceControlRateLimit.CredentialScope, "second"));
18 const error = yield* limits
19 .check(github)
20 .pipe(Effect.provideService(SourceControlRateLimit.CredentialScope, "first"), Effect.flip);
21 assert.strictEqual(error._tag, "SourceControlRateLimitPausedError");
22 }).pipe(Effect.provide(SourceControlRateLimit.layer)),
23);
24
9it("parses Retry-After seconds and HTTP dates", () => {25it("parses Retry-After seconds and HTTP dates", () => {
10 assert.equal(SourceControlRateLimit.retryAtFromHeader("120", 1_000), 121_000);26 assert.equal(SourceControlRateLimit.retryAtFromHeader("120", 1_000), 121_000);
11 assert.equal(27 assert.equal(
apps/server/src/sourceControl/SourceControlRateLimit.ts+10-5
...@@ -13,6 +13,10 @@ import {...@@ -13,6 +13,10 @@ import {
13const FALLBACK_COOLDOWN = Duration.seconds(30);13const FALLBACK_COOLDOWN = Duration.seconds(30);
14const MAX_FALLBACK_COOLDOWN = Duration.minutes(15);14const MAX_FALLBACK_COOLDOWN = Duration.minutes(15);
1515
16export const CredentialScope = Context.Reference<string>("t3/sourceControl/CredentialScope", {
17 defaultValue: () => "",
18});
19
16interface RateLimitKey {20interface RateLimitKey {
17 readonly provider: SourceControlProviderKind;21 readonly provider: SourceControlProviderKind;
18 readonly host: string;22 readonly host: string;
...@@ -59,8 +63,8 @@ export class SourceControlRateLimit extends Context.Service<...@@ -59,8 +63,8 @@ export class SourceControlRateLimit extends Context.Service<
59 }63 }
60>()("t3/sourceControl/SourceControlRateLimit") {}64>()("t3/sourceControl/SourceControlRateLimit") {}
6165
62function normalizedKey(key: RateLimitKey): string {66function normalizedKey(key: RateLimitKey, scope: string): string {
63 return `${key.provider}\0${key.host.trim().toLowerCase()}`;67 return `${key.provider}\0${key.host.trim().toLowerCase()}\0${scope}`;
64}68}
6569
66function fallbackCooldownMs(attempt: number): number {70function fallbackCooldownMs(attempt: number): number {
...@@ -90,7 +94,8 @@ export const make = Effect.gen(function* () {...@@ -90,7 +94,8 @@ export const make = Effect.gen(function* () {
90 "SourceControlRateLimit.check",94 "SourceControlRateLimit.check",
91 )(function* (input, options) {95 )(function* (input, options) {
92 const now = yield* Clock.currentTimeMillis;96 const now = yield* Clock.currentTimeMillis;
93 const entry = (yield* Ref.get(entries)).get(normalizedKey(input));97 const key = normalizedKey(input, yield* CredentialScope);
98 const entry = (yield* Ref.get(entries)).get(key);
94 if (entry !== undefined && entry.retryAt > now && options?.allowPaused !== true) {99 if (entry !== undefined && entry.retryAt > now && options?.allowPaused !== true) {
95 return yield* new SourceControlRateLimitPausedError({100 return yield* new SourceControlRateLimitPausedError({
96 provider: input.provider,101 provider: input.provider,
...@@ -105,8 +110,8 @@ export const make = Effect.gen(function* () {...@@ -105,8 +110,8 @@ export const make = Effect.gen(function* () {
105 "SourceControlRateLimit.recordRateLimit",110 "SourceControlRateLimit.recordRateLimit",
106 )(function* (input) {111 )(function* (input) {
107 const now = yield* Clock.currentTimeMillis;112 const now = yield* Clock.currentTimeMillis;
113 const key = normalizedKey(input, yield* CredentialScope);
108 yield* Ref.update(entries, (current) => {114 yield* Ref.update(entries, (current) => {
109 const key = normalizedKey(input);
110 const previous = current.get(key);115 const previous = current.get(key);
111 if (previous !== undefined && previous.generation > input.lease) {116 if (previous !== undefined && previous.generation > input.lease) {
112 if (previous.retryAt <= now && (input.retryAt === undefined || input.retryAt <= now)) {117 if (previous.retryAt <= now && (input.retryAt === undefined || input.retryAt <= now)) {
...@@ -145,8 +150,8 @@ export const make = Effect.gen(function* () {...@@ -145,8 +150,8 @@ export const make = Effect.gen(function* () {
145 "SourceControlRateLimit.recordSuccess",150 "SourceControlRateLimit.recordSuccess",
146 )(function* (input) {151 )(function* (input) {
147 const now = yield* Clock.currentTimeMillis;152 const now = yield* Clock.currentTimeMillis;
153 const key = normalizedKey(input, yield* CredentialScope);
148 yield* Ref.update(entries, (current) => {154 yield* Ref.update(entries, (current) => {
149 const key = normalizedKey(input);
150 const previous = current.get(key);155 const previous = current.get(key);
151 if (previous === undefined || previous.generation !== input.lease || previous.retryAt > now) {156 if (previous === undefined || previous.generation !== input.lease || previous.retryAt > now) {
152 return current;157 return current;
apps/server/src/sourceControl/githubGraphQlBudget.test.ts+31
...@@ -3,6 +3,7 @@ import * as Effect from "effect/Effect";...@@ -3,6 +3,7 @@ import * as Effect from "effect/Effect";
3import * as TestClock from "effect/testing/TestClock";3import * as TestClock from "effect/testing/TestClock";
44
5import * as GitHubGraphQlBudget from "./githubGraphQlBudget.ts";5import * as GitHubGraphQlBudget from "./githubGraphQlBudget.ts";
6import { CredentialScope } from "./SourceControlRateLimit.ts";
67
7const RESET_AT = "2026-08-13T14:00:00.000Z";8const RESET_AT = "2026-08-13T14:00:00.000Z";
8const NEXT_RESET_AT = "2026-08-13T15:00:00.000Z";9const NEXT_RESET_AT = "2026-08-13T15:00:00.000Z";
...@@ -71,6 +72,23 @@ describe("GitHub GraphQL budget", () => {...@@ -71,6 +72,23 @@ describe("GitHub GraphQL budget", () => {
71 }).pipe(Effect.provide(GitHubGraphQlBudget.layer)),72 }).pipe(Effect.provide(GitHubGraphQlBudget.layer)),
72 );73 );
7374
75 it.effect("isolates query reservations and observations by credential", () =>
76 Effect.gen(function* () {
77 yield* TestClock.setTime(BEFORE_RESET);
78 const budget = yield* GitHubGraphQlBudget.GitHubGraphQlBudget;
79 const query = budget.query("github.com", "query { viewer { login } }");
80 yield* budget
81 .observe("github.com", rateLimit(0))
82 .pipe(Effect.provideService(CredentialScope, "first"));
83 yield* budget
84 .observe("github.com", rateLimit(5000))
85 .pipe(Effect.provideService(CredentialScope, "second"));
86 yield* query.pipe(Effect.provideService(CredentialScope, "second"));
87 const error = yield* query.pipe(Effect.provideService(CredentialScope, "first"), Effect.flip);
88 expect(error.retryAt).toBe(Date.parse(RESET_AT));
89 }).pipe(Effect.provide(GitHubGraphQlBudget.layer)),
90 );
91
74 it.effect("keeps the lower remaining value from out-of-order responses", () =>92 it.effect("keeps the lower remaining value from out-of-order responses", () =>
75 Effect.gen(function* () {93 Effect.gen(function* () {
76 yield* TestClock.setTime(BEFORE_RESET);94 yield* TestClock.setTime(BEFORE_RESET);
...@@ -181,6 +199,19 @@ describe("GitHub GraphQL budget", () => {...@@ -181,6 +199,19 @@ describe("GitHub GraphQL budget", () => {
181 }).pipe(Effect.provide(GitHubGraphQlBudget.layer)),199 }).pipe(Effect.provide(GitHubGraphQlBudget.layer)),
182 );200 );
183201
202 it.effect("stops interactive reads when the reserve is exhausted", () =>
203 Effect.gen(function* () {
204 yield* TestClock.setTime(BEFORE_RESET);
205 const budget = yield* GitHubGraphQlBudget.GitHubGraphQlBudget;
206 yield* budget.observe("github.com", rateLimit(1, 5000, RESET_AT, 1));
207 yield* budget.query("github.com", "query { viewer { login } }", { allowReserve: true });
208 const error = yield* budget
209 .query("github.com", "query { viewer { login } }", { allowReserve: true })
210 .pipe(Effect.flip);
211 expect(error.retryAt).toBe(Date.parse(RESET_AT));
212 }).pipe(Effect.provide(GitHubGraphQlBudget.layer)),
213 );
214
184 it.effect("ignores malformed or partial rate metadata", () =>215 it.effect("ignores malformed or partial rate metadata", () =>
185 Effect.gen(function* () {216 Effect.gen(function* () {
186 yield* TestClock.setTime(BEFORE_RESET);217 yield* TestClock.setTime(BEFORE_RESET);
apps/server/src/sourceControl/githubGraphQlBudget.ts+7-4
...@@ -85,8 +85,8 @@ export const make = Effect.gen(function* () {...@@ -85,8 +85,8 @@ export const make = Effect.gen(function* () {
85 function* (host, document, options) {85 function* (host, document, options) {
86 if (!isReadOperation(document)) return document;86 if (!isReadOperation(document)) return document;
87 const now = yield* Clock.currentTimeMillis;87 const now = yield* Clock.currentTimeMillis;
88 const key = `${hostKey(host)}\0${yield* SourceControlRateLimit.CredentialScope}`;
88 const retryAt = yield* Ref.modify(snapshots, (current) => {89 const retryAt = yield* Ref.modify(snapshots, (current) => {
89 const key = hostKey(host);
90 const snapshot = current.get(key);90 const snapshot = current.get(key);
91 if (snapshot === undefined) return [null, current] as const;91 if (snapshot === undefined) return [null, current] as const;
92 if (snapshot.resetAtMs <= now) {92 if (snapshot.resetAtMs <= now) {
...@@ -94,8 +94,11 @@ export const make = Effect.gen(function* () {...@@ -94,8 +94,11 @@ export const make = Effect.gen(function* () {
94 next.delete(key);94 next.delete(key);
95 return [null, next] as const;95 return [null, next] as const;
96 }96 }
97 const remaining = Math.max(0, snapshot.remaining - Math.max(1, snapshot.cost));97 const remaining = snapshot.remaining - Math.max(1, snapshot.cost);
98 if (options?.allowReserve !== true && remaining < snapshot.limit * GRAPHQL_RESERVE_RATIO) {98 if (
99 remaining < 0 ||
100 (options?.allowReserve !== true && remaining < snapshot.limit * GRAPHQL_RESERVE_RATIO)
101 ) {
99 return [snapshot.resetAtMs, current] as const;102 return [snapshot.resetAtMs, current] as const;
100 }103 }
101 const next = new Map(current);104 const next = new Map(current);
...@@ -118,8 +121,8 @@ export const make = Effect.gen(function* () {...@@ -118,8 +121,8 @@ export const make = Effect.gen(function* () {
118 )(function* (host, raw) {121 )(function* (host, raw) {
119 const snapshot = snapshotFrom(raw);122 const snapshot = snapshotFrom(raw);
120 if (snapshot === null) return;123 if (snapshot === null) return;
124 const key = `${hostKey(host)}\0${yield* SourceControlRateLimit.CredentialScope}`;
121 yield* Ref.update(snapshots, (current) => {125 yield* Ref.update(snapshots, (current) => {
122 const key = hostKey(host);
123 const previous = current.get(key);126 const previous = current.get(key);
124 // Concurrent reads can finish out of order. Quota only falls within one reset window, and127 // Concurrent reads can finish out of order. Quota only falls within one reset window, and
125 // an answer from an older window must not replace the current one.128 // an answer from an older window must not replace the current one.
packages/client-runtime/src/state/pullRequestRouting.ts+22-7
...@@ -49,6 +49,17 @@ const writes = new Set<string>([...@@ -49,6 +49,17 @@ const writes = new Set<string>([
49]);49]);
50const isRef = Schema.is(PullRequestRef);50const isRef = Schema.is(PullRequestRef);
51const isInvalidation = Schema.is(PullRequestInvalidateInput);51const isInvalidation = Schema.is(PullRequestInvalidateInput);
52const readTimeout = (environmentId: EnvironmentId) =>
53 Effect.timeoutOrElse({
54 duration: "30 seconds",
55 orElse: () =>
56 Effect.fail(
57 new EnvironmentRpcUnavailableError({
58 environmentId,
59 message: "The environment did not respond to the PR request.",
60 }),
61 ),
62 });
52interface RoutedRead {63interface RoutedRead {
53 origin: EnvironmentId;64 origin: EnvironmentId;
54 reference: PullRequestRef;65 reference: PullRequestRef;
...@@ -318,7 +329,8 @@ export function createPullRequestRouter() {...@@ -318,7 +329,8 @@ export function createPullRequestRouter() {
318 if (!(yield* routingAllowed(registry, origin.target.environmentId, id, writes.has(tag))))329 if (!(yield* routingAllowed(registry, origin.target.environmentId, id, writes.has(tag))))
319 return yield* visit(index + 1);330 return yield* visit(index + 1);
320 }331 }
321 return yield* run(id).pipe(332 const operation = run(id);
333 return yield* (reads.has(tag) ? operation.pipe(readTimeout(id)) : operation).pipe(
322 Effect.catch((error) => {334 Effect.catch((error) => {
323 if (335 if (
324 (reads.has(tag) || rejectedBeforeDispatch(error)) &&336 (reads.has(tag) || rejectedBeforeDispatch(error)) &&
...@@ -366,12 +378,15 @@ export function createPullRequestRouter() {...@@ -366,12 +378,15 @@ export function createPullRequestRouter() {
366 }378 }
367 if (!allowed) return yield* request(tag, input);379 if (!allowed) return yield* request(tag, input);
368 const strictInput = { ...input, allowStale: false };380 const strictInput = { ...input, allowStale: false };
369 const source = yield* Effect.cached(request(tag, strictInput));381 const source = yield* Effect.cached(
370 // Cached source reads usually finish before another environment can verify its account.382 request(tag, strictInput).pipe(readTimeout(origin.target.environmentId)),
371 // Hedge slow reads only; never race mutations or retry an ambiguous write.383 );
372 return yield* Effect.race(384 const sourceEntry = entries.get(origin.target.environmentId);
373 source,385 const routed = routedRequest(tag, strictInput, source);
374 routedRequest(tag, strictInput, source).pipe(Effect.delay("75 millis")),386 return yield* (
387 sourceEntry !== undefined && isLocal(sourceEntry)
388 ? source.pipe(Effect.catch(() => routed))
389 : routed
375 ).pipe(390 ).pipe(
376 Effect.catch((error) =>391 Effect.catch((error) =>
377 input.allowStale !== false &&392 input.allowStale !== false &&
packages/client-runtime/src/state/pullRequests.test.ts+28-19
...@@ -7,6 +7,9 @@ import {...@@ -7,6 +7,9 @@ import {
7} from "@t3tools/contracts";7} from "@t3tools/contracts";
8import { expect, it } from "@effect/vitest";8import { expect, it } from "@effect/vitest";
9import * as Data from "effect/Data";9import * as Data from "effect/Data";
10import * as Deferred from "effect/Deferred";
11import * as Fiber from "effect/Fiber";
12import * as TestClock from "effect/testing/TestClock";
10import * as Effect from "effect/Effect";13import * as Effect from "effect/Effect";
11import * as Latch from "effect/Latch";14import * as Latch from "effect/Latch";
12import * as Layer from "effect/Layer";15import * as Layer from "effect/Layer";
...@@ -52,7 +55,7 @@ for (const scenario of [...@@ -52,7 +55,7 @@ for (const scenario of [
52 "prefers the local environment with the same github account",55 "prefers the local environment with the same github account",
53 "falls back before mutation when the local account differs",56 "falls back before mutation when the local account differs",
54 "never retries an ambiguous mutation failure",57 "never retries an ambiguous mutation failure",
55 "returns a fast source read without checking alternate identities",58 "returns a fast local source read without checking alternate identities",
56 "keeps single-environment requests free of identity lookups",59 "keeps single-environment requests free of identity lookups",
57 "keeps a local origin ahead of another local environment",60 "keeps a local origin ahead of another local environment",
58 "keeps mutations on an old origin server without retrying them",61 "keeps mutations on an old origin server without retrying them",
...@@ -79,10 +82,12 @@ for (const scenario of [...@@ -79,10 +82,12 @@ for (const scenario of [
79 switchedAccount;82 switchedAccount;
80 const ambiguous = scenario === "never retries an ambiguous mutation failure";83 const ambiguous = scenario === "never retries an ambiguous mutation failure";
81 const reading =84 const reading =
82 scenario === "returns a fast source read without checking alternate identities";85 scenario === "returns a fast local source read without checking alternate identities";
83 const single = scenario === "keeps single-environment requests free of identity lookups";86 const single = scenario === "keeps single-environment requests free of identity lookups";
84 const localOrigin =87 const localOrigin =
85 scenario === "keeps a local origin ahead of another local environment" || switchedAccount;88 scenario === "keeps a local origin ahead of another local environment" ||
89 switchedAccount ||
90 reading;
86 const oldOrigin =91 const oldOrigin =
87 scenario === "keeps mutations on an old origin server without retrying them";92 scenario === "keeps mutations on an old origin server without retrying them";
88 const oldAlternate =93 const oldAlternate =
...@@ -557,15 +562,15 @@ for (const probe of ["origin", "alternate"] as const) {...@@ -557,15 +562,15 @@ for (const probe of ["origin", "alternate"] as const) {
557 );562 );
558}563}
559564
560for (const source of ["pending", "pending-local", "failed", "offline"] as const) {565for (const source of ["pending", "pending-local", "failed-local", "failed", "offline"] as const) {
561 it.live(566 it.effect(
562 source === "offline"567 source === "offline"
563 ? "returns held source data only after both fresh paths fail"568 ? "returns held source data only after both fresh paths fail"
564 : `hedges a ${source} source read to local and interrupts the losing read`,569 : `uses one shared reader with a ${source} source`,
565 () =>570 () =>
566 Effect.scoped(571 Effect.scoped(
567 Effect.gen(function* () {572 Effect.gen(function* () {
568 let interrupted = false;573 const started = yield* Deferred.make<void>();
569 const calls: string[] = [];574 const calls: string[] = [];
570 const clientFor = (local: boolean) =>575 const clientFor = (local: boolean) =>
571 ({576 ({
...@@ -590,21 +595,16 @@ for (const source of ["pending", "pending-local", "failed", "offline"] as const)...@@ -590,21 +595,16 @@ for (const source of ["pending", "pending-local", "failed", "offline"] as const)
590 operation: "summary",595 operation: "summary",
591 detail: "github unreachable",596 detail: "github unreachable",
592 });597 });
593 return yield* Effect.never.pipe(598 yield* Deferred.succeed(started, undefined);
594 Effect.onInterrupt(() =>599 return yield* Effect.never;
595 Effect.sync(() => {
596 interrupted = true;
597 }),
598 ),
599 );
600 }),600 }),
601 }) as unknown as WsRpcProtocolClient;601 }) as unknown as WsRpcProtocolClient;
602 const { environmentRegistry, supervisor } = yield* makeTestRuntime(602 const { environmentRegistry, supervisor } = yield* makeTestRuntime(
603 clientFor(false),603 clientFor(false),
604 clientFor(true),604 clientFor(true),
605 source === "pending-local",605 source === "failed-local" || source === "pending-local",
606 );606 );
607 const result = yield* createPullRequestRouter()(WS_METHODS.pullRequestsSummary, {607 const request = createPullRequestRouter()(WS_METHODS.pullRequestsSummary, {
608 projectId: ProjectId.make("project-1"),608 projectId: ProjectId.make("project-1"),
609 repository: "acme/web",609 repository: "acme/web",
610 number: 7,610 number: 7,
...@@ -613,14 +613,23 @@ for (const source of ["pending", "pending-local", "failed", "offline"] as const)...@@ -613,14 +613,23 @@ for (const source of ["pending", "pending-local", "failed", "offline"] as const)
613 Effect.provideService(GitHubRoutingPermissions, trustedRouting),613 Effect.provideService(GitHubRoutingPermissions, trustedRouting),
614 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),614 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
615 );615 );
616 const fiber = yield* request.pipe(Effect.forkChild);
617 if (source === "pending-local") {
618 yield* Deferred.await(started);
619 yield* TestClock.adjust("30 seconds");
620 }
621 const result = yield* Fiber.join(fiber);
616 if (source === "offline") {622 if (source === "offline") {
617 expect(result).toEqual({ state: "open" });623 expect(result).toEqual({ state: "open" });
618 expect(calls).toEqual(["origin", "local", "held"]);624 expect(calls).toEqual(["local", "origin", "held"]);
619 } else {625 } else {
620 expect(result).toBeNull();626 expect(result).toBeNull();
621 expect(calls).toEqual(["origin", "local"]);627 expect(calls).toEqual(
628 source === "failed-local" || source === "pending-local"
629 ? ["origin", "local"]
630 : ["local"],
631 );
622 }632 }
623 expect(interrupted).toBe(source === "pending" || source === "pending-local");
624 }),633 }),
625 ),634 ),
626 );635 );