diff --git a/apps/server/integration/OrchestrationEngineHarness.integration.ts b/apps/server/integration/OrchestrationEngineHarness.integration.ts index 4d8a384ee997..510478a57a02 100644 --- a/apps/server/integration/OrchestrationEngineHarness.integration.ts +++ b/apps/server/integration/OrchestrationEngineHarness.integration.ts @@ -352,7 +352,7 @@ export const makeOrchestrationIntegrationHarness = ( Layer.provideMerge(runtimeServicesLayer), Layer.provideMerge( Layer.mock(PullRequestService.PullRequestService)({ - refreshAfterTurn: Effect.void, + refreshAfterTurn: () => Effect.void, }), ), Layer.provideMerge( diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts index 07fb9cdcc0cd..2cc7a4399915 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts @@ -336,7 +336,7 @@ describe("CheckpointReactor", () => { prefix: "t3-checkpoint-reactor-test-", }); const pullRequestRefreshes: number[] = []; - const refreshAfterTurn = Effect.sync(() => void pullRequestRefreshes.push(1)); + const refreshAfterTurn = () => Effect.sync(() => void pullRequestRefreshes.push(1)); const vcsStatusBroadcasterLayer = Layer.succeed(VcsStatusBroadcaster, { getStatus: () => Effect.die("getStatus should not be called in this test"), refreshLocalStatus: (cwd: string) => diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.ts index 38868430ca01..d0d867fdd25e 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.ts @@ -886,7 +886,7 @@ const make = Effect.gen(function* () { (startedTurnId === undefined && !thread.session?.activeTurnId)) ) { pending.delete(event.threadId); - yield* pullRequests.refreshAfterTurn; + yield* pullRequests.refreshAfterTurn(thread.projectId); } if ( event.type === "turn.aborted" && diff --git a/apps/server/src/pullRequest/GitHubPullRequestCli.test.ts b/apps/server/src/pullRequest/GitHubPullRequestCli.test.ts index a62d1ed21970..2f90c5b79553 100644 --- a/apps/server/src/pullRequest/GitHubPullRequestCli.test.ts +++ b/apps/server/src/pullRequest/GitHubPullRequestCli.test.ts @@ -9,6 +9,7 @@ import { ChildProcessSpawner } from "effect/unstable/process"; import * as GitHubCli from "../sourceControl/GitHubCli.ts"; import * as VcsProcess from "../vcs/VcsProcess.ts"; +import * as SourceControlRateLimit from "../sourceControl/SourceControlRateLimit.ts"; import * as GitHubGraphQlBudget from "../sourceControl/githubGraphQlBudget.ts"; import * as GitHubPullRequestCli from "./GitHubPullRequestCli.ts"; import { BASE_COMPARISON_GRAPHQL_QUERY } from "./gitHubPullRequestJson.ts"; @@ -202,6 +203,7 @@ it.effect( let activeToken = "broad-credential"; const commands: VcsProcess.VcsProcessInput[] = []; const github = yield* GitHubCli.make.pipe( + Effect.provide(Layer.merge(GitHubGraphQlBudget.layer, SourceControlRateLimit.layer)), Effect.provideService(VcsProcess.VcsProcess, { run: (input) => Effect.sync(() => { diff --git a/apps/server/src/pullRequest/GitHubPullRequestCli.ts b/apps/server/src/pullRequest/GitHubPullRequestCli.ts index 2558f6695f1a..1b2efa7ec70e 100644 --- a/apps/server/src/pullRequest/GitHubPullRequestCli.ts +++ b/apps/server/src/pullRequest/GitHubPullRequestCli.ts @@ -1106,6 +1106,7 @@ export const make = Effect.gen(function* () { captureVerifiedCredential(input).pipe( Effect.flatMap(({ host, token, accountId, viewer, credentialFingerprint }) => use({ accountId, viewer, credentialFingerprint }).pipe( + Effect.provideService(SourceControlRateLimit.CredentialScope, credentialFingerprint), Effect.provideService(GitHubCli.PinnedGitHubCredential, { host, token, diff --git a/apps/server/src/pullRequest/GitHubPullRequestProvider.ts b/apps/server/src/pullRequest/GitHubPullRequestProvider.ts index f8019ca36b4f..1d003a970ee8 100644 --- a/apps/server/src/pullRequest/GitHubPullRequestProvider.ts +++ b/apps/server/src/pullRequest/GitHubPullRequestProvider.ts @@ -109,7 +109,11 @@ export function gitHubProviderFailure( ): PullRequestProviderFailure { if (error._tag === "GitHubCliUnavailableError") return { reason: "missing-tool" }; if (error._tag === "GitHubCliAuthenticationError") return { reason: "unauthenticated" }; - if (error._tag === "GitHubCliRateLimitError") return { reason: "rate-limited" }; + if (error._tag === "GitHubCliRateLimitError") + return { + reason: "rate-limited", + ...(error.retryAt === undefined ? {} : { retryAt: error.retryAt }), + }; if (error._tag === "SourceControlRateLimitPausedError") { return { reason: "rate-limited", retryAt: error.retryAt }; } diff --git a/apps/server/src/pullRequest/PullRequestReadCache.test.ts b/apps/server/src/pullRequest/PullRequestReadCache.test.ts index f94ff3cf586d..03bfe88abfd0 100644 --- a/apps/server/src/pullRequest/PullRequestReadCache.test.ts +++ b/apps/server/src/pullRequest/PullRequestReadCache.test.ts @@ -5,18 +5,12 @@ import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; -import * as Layer from "effect/Layer"; import * as TestClock from "effect/testing/TestClock"; import * as KeyValueStore from "effect/unstable/persistence/KeyValueStore"; -import * as Persistence from "effect/unstable/persistence/Persistence"; import * as PullRequestReadCache from "./PullRequestReadCache.ts"; const cacheLayer = (directory: string) => - PullRequestReadCache.make.pipe( - Effect.provide( - Persistence.layerKvs.pipe(Layer.provideMerge(KeyValueStore.layerFileSystem(directory))), - ), - ); + PullRequestReadCache.make.pipe(Effect.provide(KeyValueStore.layerFileSystem(directory))); it.layer(NodeServices.layer)("PR filesystem cache", (it) => { it.effect("reuses files after restart and respects the original expiry", () => @@ -52,15 +46,106 @@ it.layer(NodeServices.layer)("PR filesystem cache", (it) => { Effect.andThen(Deferred.await(release)), Effect.as("old"), ), + ["pr"], ) .pipe(Effect.forkChild); yield* Deferred.await(started); - const invalidate = yield* cache.invalidate.pipe(Effect.forkChild({ startImmediately: true })); + const invalidate = yield* cache + .invalidate("pr") + .pipe(Effect.forkChild({ startImmediately: true })); yield* Deferred.succeed(release, undefined); yield* Fiber.join(read); yield* Fiber.join(invalidate); const restarted = yield* cacheLayer(directory); - assert.strictEqual(yield* restarted.get("summary", Effect.succeed("new")), "new"); + assert.strictEqual(yield* restarted.get("summary", Effect.succeed("new"), ["pr"]), "new"); + }), + ); + + it.effect("invalidates only the changed scope across restarts and coalesces its next reads", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pr-cache-" }); + let reads = 0; + const lookup = Effect.sync(() => String(++reads)); + const cache = yield* cacheLayer(directory); + yield* cache.get("first", lookup, ["project", "pr-1"]); + yield* cache.get("second", lookup, ["project", "pr-2"]); + yield* cache.get("third", lookup, ["other-project", "pr-3"]); + yield* cache.invalidate("pr-1"); + const restarted = yield* cacheLayer(directory); + const answers = yield* Effect.all( + Array.from({ length: 10 }, () => restarted.get("first", lookup, ["project", "pr-1"])), + { concurrency: 10 }, + ); + assert.deepStrictEqual(answers, Array(10).fill("4")); + assert.strictEqual(yield* restarted.get("second", lookup, ["project", "pr-2"]), "2"); + yield* restarted.invalidate("project"); + const again = yield* cacheLayer(directory); + assert.strictEqual(yield* again.get("second", lookup, ["project", "pr-2"]), "5"); + assert.strictEqual(yield* again.get("third", lookup, ["other-project", "pr-3"]), "3"); + assert.strictEqual(reads, 5); + const files = (yield* fs.readDirectory(directory)).length; + for (let index = 0; index < 3; index++) { + yield* again.invalidate("pr-1"); + yield* again.get("first", lookup, ["project", "pr-1"]); + } + assert.strictEqual((yield* fs.readDirectory(directory)).length, files); + }), + ); + + it.effect("shares a pending refresh without blocking an unrelated cached PR", () => + Effect.gen(function* () { + const cache = yield* PullRequestReadCache.make; + yield* cache.get("first", Effect.succeed("old"), ["pr-1"]); + yield* cache.get("second", Effect.succeed("warm"), ["pr-2"]); + yield* cache.invalidate("pr-1"); + const started = yield* Deferred.make(); + const release = yield* Deferred.make(); + let reads = 0; + const refresh = cache.get( + "first", + Effect.gen(function* () { + reads++; + yield* Deferred.succeed(started, undefined); + yield* Deferred.await(release); + return "fresh"; + }), + ["pr-1"], + ); + const pending = yield* Effect.all( + Array.from({ length: 10 }, () => refresh), + { + concurrency: 10, + }, + ).pipe(Effect.forkChild); + yield* Deferred.await(started); + assert.strictEqual(yield* cache.get("second", Effect.die("cache miss"), ["pr-2"]), "warm"); + yield* Deferred.succeed(release, undefined); + assert.deepStrictEqual(yield* Fiber.join(pending), Array(10).fill("fresh")); + assert.strictEqual(reads, 1); + }).pipe(Effect.provide(KeyValueStore.layerMemory)), + ); + + it.effect("compacts expired scope records without discarding fresh PR data", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pr-cache-" }); + const cache = yield* cacheLayer(directory); + yield* cache.get("summary", Effect.succeed("old"), ["pr"]); + yield* cache.invalidate("pr"); + for (let index = 0; index < 100; index++) yield* cache.invalidate(`pr-${index}`); + assert.strictEqual((yield* fs.readDirectory(directory)).length, 2); + const before = (yield* fs.stat(`${directory}/revisions`)).size; + yield* TestClock.adjust("59 seconds"); + assert.strictEqual(yield* cache.get("summary", Effect.succeed("fresh"), ["pr"]), "fresh"); + yield* TestClock.adjust("1 second"); + yield* cache.invalidate("other-pr"); + assert.isTrue((yield* fs.stat(`${directory}/revisions`)).size < before); + const restarted = yield* cacheLayer(directory); + assert.strictEqual( + yield* restarted.get("summary", Effect.die("cache miss"), ["pr"]), + "fresh", + ); }), ); @@ -75,4 +160,85 @@ it.layer(NodeServices.layer)("PR filesystem cache", (it) => { assert.strictEqual(yield* restarted.get("summary", Effect.succeed("recovered")), "recovered"); }), ); + + it.effect("resumes caching after a failed scope read", () => + Effect.gen(function* () { + const backing = yield* KeyValueStore.KeyValueStore; + let fail = true; + let reads = 0; + const cache = yield* PullRequestReadCache.make.pipe( + Effect.provideService(KeyValueStore.KeyValueStore, { + ...backing, + get: (key) => + Effect.suspend(() => { + if (!fail) return backing.get(key); + fail = false; + return Effect.fail( + new KeyValueStore.KeyValueStoreError({ method: "get", message: "unavailable" }), + ); + }), + }), + ); + const read = cache.get( + "summary", + Effect.sync(() => String(++reads)), + ["pr"], + ); + assert.strictEqual(yield* read, "1"); + assert.strictEqual(yield* read, "2"); + assert.strictEqual(yield* read, "2"); + }).pipe(Effect.provide(KeyValueStore.layerMemory)), + ); + + it.effect("cancels abandoned reads without blocking invalidation", () => + Effect.gen(function* () { + const cache = yield* PullRequestReadCache.make; + const started = yield* Deferred.make(); + const read = yield* cache + .get("summary", Deferred.succeed(started, undefined).pipe(Effect.andThen(Effect.never)), [ + "pr", + ]) + .pipe(Effect.forkChild); + yield* Deferred.await(started); + yield* Fiber.interrupt(read); + yield* cache.invalidate("pr"); + assert.strictEqual(yield* cache.get("summary", Effect.succeed("fresh"), ["pr"]), "fresh"); + }).pipe(Effect.provide(KeyValueStore.layerMemory)), + ); + + it.effect( + "finishes the in-memory revision update when invalidation is canceled after writing", + () => + Effect.gen(function* () { + const backing = yield* KeyValueStore.KeyValueStore; + const written = yield* Deferred.make(); + const release = yield* Deferred.make(); + const cache = yield* PullRequestReadCache.make.pipe( + Effect.provideService(KeyValueStore.KeyValueStore, { + ...backing, + set: (key, value) => + backing + .set(key, value) + .pipe( + Effect.andThen( + key === "revisions" + ? Deferred.succeed(written, undefined).pipe( + Effect.andThen(Deferred.await(release)), + ) + : Effect.void, + ), + ), + }), + ); + yield* cache.get("summary", Effect.succeed("old"), ["pr"]); + const invalidation = yield* cache.invalidate("pr").pipe(Effect.forkChild); + yield* Deferred.await(written); + const interrupt = yield* Fiber.interrupt(invalidation).pipe( + Effect.forkChild({ startImmediately: true }), + ); + yield* Deferred.succeed(release, undefined); + yield* Fiber.join(interrupt); + assert.strictEqual(yield* cache.get("summary", Effect.succeed("fresh"), ["pr"]), "fresh"); + }).pipe(Effect.provide(KeyValueStore.layerMemory)), + ); }); diff --git a/apps/server/src/pullRequest/PullRequestReadCache.ts b/apps/server/src/pullRequest/PullRequestReadCache.ts index 62d1cffc3c83..1b4ded39b953 100644 --- a/apps/server/src/pullRequest/PullRequestReadCache.ts +++ b/apps/server/src/pullRequest/PullRequestReadCache.ts @@ -5,7 +5,6 @@ import * as Hash from "effect/Hash"; import { PullRequestOperationError, PullRequestUnavailableError } from "@t3tools/contracts"; import * as Crypto from "effect/Crypto"; import * as Encoding from "effect/Encoding"; -import * as Option from "effect/Option"; import * as Context from "effect/Context"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; @@ -22,19 +21,35 @@ import { ServerConfig } from "../config.ts"; const CONCURRENT_READS = 512; type ReadError = PullRequestOperationError | PullRequestUnavailableError; - +const revisionCodec = Schema.fromJsonString( + Schema.Record( + Schema.String, + Schema.Struct({ revision: Schema.String, expiresAt: Schema.Finite }), + ), +); class Read extends Persistable.Class<{ - payload: { key: string; lookup: Effect.Effect }; + payload: { key: string; revision: string; lookup: Effect.Effect }; }>()("PullRequestRead", { primaryKey: ({ key }) => key, - success: Schema.Struct({ payload: Schema.String, expiresAt: Schema.Finite }), + success: Schema.Struct({ + payload: Schema.String, + expiresAt: Schema.Finite, + revision: Schema.optionalKey(Schema.String), + }), error: Schema.Union([PullRequestOperationError, PullRequestUnavailableError]), }) { + matchesRevision(revision: string | undefined): boolean { + const stored = revision?.split(":") ?? []; + return this.revision + .split(":") + .every((value, index) => value === "" || value === stored[index]); + } + [Equal.symbol](that: unknown): boolean { - return that instanceof Read && that.key === this.key; + return that instanceof Read && that.key === this.key && that.revision === this.revision; } [Hash.symbol](): number { - return Hash.string(this.key); + return Hash.string(`${this.key}:${this.revision}`); } } @@ -44,8 +59,9 @@ export class PullRequestReadCache extends Context.Service< readonly get: ( key: string, lookup: Effect.Effect, + scopes?: ReadonlyArray, ) => Effect.Effect; - readonly invalidate: Effect.Effect; + readonly invalidate: (scope: string) => Effect.Effect; } >()("t3/pullRequest/PullRequestReadCache") {} @@ -55,6 +71,18 @@ export const make = Effect.gen(function* () { const clock = yield* Clock.Clock; let enabled = true; const lock = yield* Semaphore.make(CONCURRENT_READS); + const digest = (key: string) => + crypto.digest("SHA-256", new TextEncoder().encode(key)).pipe(Effect.map(Encoding.encodeHex)); + const revisions = yield* Cache.makeWith( + () => + backing + .get("revisions") + .pipe(Effect.flatMap((raw) => Schema.decodeUnknownEffect(revisionCodec)(raw ?? "{}"))), + { + capacity: 1, + timeToLive: (exit) => (Exit.isSuccess(exit) ? Duration.infinity : Duration.zero), + }, + ); const timeToLive: Persistable.TimeToLiveFn = (exit) => Exit.isSuccess(exit) ? Duration.millis(Math.max(0, exit.value.expiresAt - clock.currentTimeMillisUnsafe())) @@ -62,7 +90,11 @@ export const make = Effect.gen(function* () { const cache = yield* PersistedCache.make( (request: Read) => request.lookup.pipe( - Effect.map((payload) => ({ payload, expiresAt: clock.currentTimeMillisUnsafe() + 60_000 })), + Effect.map((payload) => ({ + payload, + expiresAt: clock.currentTimeMillisUnsafe() + 60_000, + revision: request.revision, + })), ), { storeId: "pr-v2", @@ -70,36 +102,63 @@ export const make = Effect.gen(function* () { inMemoryTTL: timeToLive, inMemoryCapacity: CONCURRENT_READS, }, + ).pipe(Effect.provide(Persistence.layerKvs)); + const refreshes = yield* Cache.makeWith( + Effect.fn("PullRequestReadCache.refresh")(function* (request: Read) { + const stored = yield* cache.get(request); + if (request.matchesRevision(stored.revision)) return stored; + yield* cache.invalidate(request); + return yield* cache.get(request); + }), + { capacity: CONCURRENT_READS, timeToLive: () => Duration.zero }, ); return PullRequestReadCache.of({ - get: Effect.fn("PullRequestReadCache.get")(function* (key, lookup) { + get: Effect.fn("PullRequestReadCache.get")(function* (key, lookup, scopes = []) { if (!enabled) return yield* lookup; - const digest = yield* crypto - .digest("SHA-256", new TextEncoder().encode(key)) - .pipe(Effect.option); - if (Option.isNone(digest)) return yield* lookup; const read = yield* Effect.cached(lookup); - return yield* cache - .get(new Read({ key: Encoding.encodeHex(digest.value), lookup: read })) - .pipe( - Effect.map((result) => result.payload), - Effect.catchTags({ - PersistenceError: () => read, - SchemaError: () => read, - }), - Effect.uninterruptible, - lock.withPermits(1), - ); + return yield* Effect.gen(function* () { + const current = yield* Cache.get(revisions, undefined); + const now = clock.currentTimeMillisUnsafe(); + const revision = scopes + .map((scope) => { + const value = current[scope]; + return value !== undefined && value.expiresAt > now ? value.revision : ""; + }) + .join(":"); + const request = new Read({ key: yield* digest(key), revision, lookup: read }); + const stored = yield* cache.get(request); + return ( + request.matchesRevision(stored.revision) ? stored : yield* Cache.get(refreshes, request) + ).payload; + }).pipe( + Effect.catchTags({ + PlatformError: () => read, + KeyValueStoreError: () => read, + PersistenceError: () => read, + SchemaError: () => read, + }), + lock.withPermits(1), + ); }), - // Let existing reads finish before clearing, so they cannot repopulate stale entries. - invalidate: Cache.invalidateAll(cache.inMemory).pipe( - Effect.andThen(backing.clear), - Effect.catch(() => { - enabled = false; - return Effect.logWarning("PR cache disabled after clearing failed"); - }), - lock.withPermits(CONCURRENT_READS), - ), + invalidate: (scope) => + Effect.gen(function* () { + const now = clock.currentTimeMillisUnsafe(); + const current = yield* Cache.get(revisions, undefined); + const next = Object.fromEntries( + Object.entries(current).filter(([, value]) => value.expiresAt > now), + ); + next[scope] = { revision: yield* crypto.randomUUIDv4, expiresAt: now + 60_000 }; + const encoded = yield* Schema.encodeEffect(revisionCodec)(next); + yield* backing.set("revisions", encoded); + yield* Cache.set(revisions, undefined, next); + }).pipe( + Effect.catch(() => { + enabled = false; + return Effect.logWarning("PR cache disabled after clearing failed"); + }), + Effect.uninterruptible, + lock.withPermits(CONCURRENT_READS), + ), }); }); @@ -108,7 +167,6 @@ export const layer = Layer.unwrap( const config = yield* ServerConfig; const path = yield* Path.Path; return Layer.effect(PullRequestReadCache, make).pipe( - Layer.provide(Persistence.layerKvs), Layer.provide( KeyValueStore.layerFileSystem( path.join(config.providerStatusCacheDir, "pull-requests"), diff --git a/apps/server/src/pullRequest/PullRequestService.test.ts b/apps/server/src/pullRequest/PullRequestService.test.ts index 9a1009c243de..c576101fa5a9 100644 --- a/apps/server/src/pullRequest/PullRequestService.test.ts +++ b/apps/server/src/pullRequest/PullRequestService.test.ts @@ -1,6 +1,5 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import * as KeyValueStore from "effect/unstable/persistence/KeyValueStore"; -import * as Persistence from "effect/unstable/persistence/Persistence"; import { assert, it } from "@effect/vitest"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; @@ -204,7 +203,6 @@ function makeService(input: { }), SourceControlRateLimit.layer, Layer.effect(PullRequestReadCache.PullRequestReadCache, PullRequestReadCache.make).pipe( - Layer.provide(Persistence.layerKvs), Layer.provide(KeyValueStore.layerMemory), Layer.provide(NodeServices.layer), ), @@ -3117,6 +3115,107 @@ it.effect("a listing narrowed to some projects is its own cache entry", () => }), ); +it.effect("keeps unrelated PRs warm after a mutation, explicit refresh, and project turn", () => + Effect.gen(function* () { + const calls: string[] = []; + const service = yield* makeService({ + projects: [ + project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" }), + project({ id: "p2", title: "docs", workspaceRoot: "/b", repository: "acme/docs" }), + ], + providers: [ + fakeProvider("github", { + getChangeRequest: (input) => + Effect.sync(() => { + calls.push(`${input.repository}/${input.number}`); + return { ...hostedChangeRequest("body"), number: input.number }; + }), + }), + ], + }); + const refs = [ + { projectId: "p1" as ProjectId, repository: "acme/web", number: 1 }, + { projectId: "p1" as ProjectId, repository: "acme/web", number: 2 }, + { projectId: "p2" as ProjectId, repository: "acme/docs", number: 3 }, + ]; + const readAll = Effect.forEach(refs, (ref) => service.summary({ ...ref, allowStale: false })); + yield* readAll; + yield* service.invalidate({ reference: { ...refs[0]!, host: "github.com" } }); + yield* readAll; + assert.deepStrictEqual(calls, ["acme/web/1", "acme/web/2", "acme/docs/3", "acme/web/1"]); + yield* service.comment({ ...refs[0]!, body: "hello" }); + yield* readAll; + assert.deepStrictEqual(calls.slice(4), ["acme/web/1"]); + yield* service.refreshAfterTurn("p1" as ProjectId); + yield* readAll; + assert.deepStrictEqual(calls.slice(5), ["acme/web/1", "acme/web/2"]); + }), +); + +it.effect( + "keeps matching PR numbers on different hosts separate and refreshes the serving project", + () => + Effect.gen(function* () { + const hosts: string[] = []; + const service = yield* makeService({ + projects: [ + project({ id: "p1", title: "public", workspaceRoot: "/a", repository: "acme/web" }), + project({ + id: "p2", + title: "enterprise", + workspaceRoot: "/b", + repository: "acme/web", + host: "enterprise.test", + }), + ], + providers: [ + fakeProvider("github", { + getChangeRequest: (input) => + Effect.sync(() => { + hosts.push(input.host); + return hostedChangeRequest("body"); + }), + }), + ], + }); + const own = { projectId: "p1" as ProjectId, repository: "acme/web", number: 1 }; + const other = { ...own, host: "enterprise.test" }; + const readBoth = Effect.all([service.summary(own), service.summary(other)]); + yield* readBoth; + yield* service.invalidate({ reference: { ...own, host: "github.com" } }); + yield* readBoth; + assert.deepStrictEqual(hosts, ["github.com", "enterprise.test", "github.com"]); + yield* service.invalidate({ reference: own }); + yield* readBoth; + assert.deepStrictEqual(hosts.slice(3), ["github.com"]); + yield* service.refreshAfterTurn("p2" as ProjectId); + yield* readBoth; + assert.deepStrictEqual(hosts.slice(4), ["enterprise.test"]); + }), +); + +it.effect("does not revive old summaries when project epochs are evicted", () => + Effect.gen(function* () { + let title = "old"; + const service = yield* makeService({ + projects: [project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" })], + providers: [ + fakeProvider("github", { + getChangeRequest: () => Effect.succeed({ ...hostedChangeRequest("body"), title }), + }), + ], + }); + const ref = { projectId: "p1" as ProjectId, repository: "acme/web", number: 1 }; + assert.strictEqual((yield* service.summary(ref))?.title, "old"); + title = "new"; + yield* service.refreshAfterTurn(ref.projectId); + assert.strictEqual((yield* service.summary(ref))?.title, "new"); + for (let index = 0; index < 2048; index++) + yield* service.refreshAfterTurn(`project-${index}` as ProjectId); + assert.strictEqual((yield* service.summary(ref))?.title, "new"); + }), +); + it.effect("explicit and turn invalidations make the next listing ask the host again", () => Effect.gen(function* () { let hostCalls = 0; @@ -3148,7 +3247,7 @@ it.effect("explicit and turn invalidations make the next listing ask the host ag yield* service.invalidate({ reference }); yield* service.list({ state: "open" }); assert.strictEqual(hostCalls, 2); - yield* service.refreshAfterTurn; + yield* service.refreshAfterTurn("p1" as ProjectId); const refresh = Option.getOrThrow(yield* Stream.runHead(service.subscribeRefreshes)); yield* service.list({ state: "open" }); assert.isAbove(refresh, 0); @@ -3551,7 +3650,7 @@ it.effect( yield* service.listStats({ refs: [ref(1), ref(2), ref(3)] }); assert.deepStrictEqual(asked, [[1, 2], [3], [2], [1, 2, 3]]); - yield* service.refreshAfterTurn; + yield* service.refreshAfterTurn("p1" as ProjectId); yield* service.listStats({ refs: [ref(1)] }); assert.deepStrictEqual(asked, [[1, 2], [3], [2], [1, 2, 3], [1]]); @@ -3768,9 +3867,9 @@ it.effect("shares linked summaries and reuses them for display without asking th }), ); -it.effect("keeps routed summaries and details separate when the GitHub account changes", () => +it.effect("keeps routed reads separate when the GitHub account changes", () => Effect.gen(function* () { - for (const operation of ["summary", "detail"] as const) { + for (const operation of ["summary", "detail", "diff"] as const) { let failing = false; let calls = 0; const read = () => @@ -3785,16 +3884,27 @@ it.effect("keeps routed summaries and details separate when the GitHub account c project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" }), ], providers: [ - fakeProvider("github", { getChangeRequestSummary: read, getChangeRequest: read }), + fakeProvider("github", { + getChangeRequestSummary: read, + getChangeRequest: read, + getDiff: () => + read().pipe( + Effect.as({ patch: "private patch", truncated: false, nextCursor: null }), + ), + }), ], }); + const readOperation = (input: Parameters[0]) => + Effect.gen(function* () { + yield* service[operation](input); + }); const reference = { projectId: "p1" as ProjectId, repository: "acme/web", number: 1 }; - yield* service[operation]({ ...reference, expectedAccountId: "101" }); + yield* readOperation({ ...reference, expectedAccountId: "101" }); failing = true; for (const allowStale of [false, true]) { const error = yield* Effect.flip( - service[operation]({ ...reference, expectedAccountId: "202", allowStale }), + readOperation({ ...reference, expectedAccountId: "202", allowStale }), ); assert.strictEqual(error._tag, "PullRequestOperationError"); } @@ -3805,7 +3915,7 @@ it.effect("keeps routed summaries and details separate when the GitHub account c it.effect("isolates routed caches for two credentials belonging to the same account", () => Effect.gen(function* () { - for (const operation of ["summary", "detail"] as const) { + for (const operation of ["summary", "detail", "diff"] as const) { let credential = "broad"; let calls = 0; const read = () => @@ -3831,9 +3941,17 @@ it.effect("isolates routed caches for two credentials belonging to the same acco ), getChangeRequest: read, getChangeRequestSummary: read, + getDiff: () => + read().pipe( + Effect.as({ patch: "private patch", truncated: false, nextCursor: null }), + ), }), ], }); + const readOperation = (input: Parameters[0]) => + Effect.gen(function* () { + yield* service[operation](input); + }); const reference = { projectId: "p1" as ProjectId, repository: "acme/web", @@ -3841,14 +3959,11 @@ it.effect("isolates routed caches for two credentials belonging to the same acco host: "github.com", expectedAccountId: "101", }; - yield* service.withRoutingCredential(reference, service[operation](reference)); + yield* service.withRoutingCredential(reference, readOperation(reference)); credential = "restricted"; for (const allowStale of [false, true]) { const error = yield* Effect.flip( - service.withRoutingCredential( - reference, - service[operation]({ ...reference, allowStale }), - ), + service.withRoutingCredential(reference, readOperation({ ...reference, allowStale })), ); assert.strictEqual(error._tag, "PullRequestOperationError"); } @@ -4354,90 +4469,104 @@ it.effect('resolves an author filter of "me" to the viewer before narrowing a ho }), ); -it.effect("authorizes stack rebases independently of whether the selected layer is behind", () => - Effect.gen(function* () { - let taken = 0; - let summaryReads = 0; - let mutationFails = false; - let stackRebase = true; - let stackActions = true; - const capabilities = { - diff: true, - comment: true, - actions: ["update-branch"] as const, - mergeMethods: ["merge"] as const, - updateMethods: ["rebase"] as const, - get stackActions() { - return stackActions; - }, - search: true, - reactions: true, - review: FULL_REVIEW, - reviewers: FULL_REVIEWERS, - }; - const service = yield* makeService({ - projects: [project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" })], - providers: [ - fakeProvider("github", { - capabilities, - getViewerPermissions: () => - Effect.succeed({ - actions: [], - stackRebase, - comment: true, - resolve: false, - verdicts: [], - requestReviewers: false, - }), - getChangeRequestSummary: () => - Effect.sync(() => { - summaryReads++; - return changeRequest(8, "2026-07-01T00:00:00Z"); +for (const crossHost of [false, true]) { + it.effect( + `authorizes stack rebases and refreshes sibling layers (cross-host: ${crossHost})`, + () => + Effect.gen(function* () { + let taken = 0; + let summaryReads = 0; + let mutationFails = false; + let stackRebase = true; + let stackActions = true; + const capabilities = { + diff: true, + comment: true, + actions: ["update-branch"] as const, + mergeMethods: ["merge"] as const, + updateMethods: ["rebase"] as const, + get stackActions() { + return stackActions; + }, + search: true, + reactions: true, + review: FULL_REVIEW, + reviewers: FULL_REVIEWERS, + }; + const service = yield* makeService({ + projects: [ + project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" }), + project({ + id: "p2", + title: "enterprise", + workspaceRoot: "/b", + repository: "acme/web", + host: "enterprise.test", }), - runAction: () => - Effect.gen(function* () { - taken++; - if (mutationFails) return yield* requestFailed; + ], + providers: [ + fakeProvider("github", { + capabilities, + getViewerPermissions: () => + Effect.succeed({ + actions: [], + stackRebase, + comment: true, + resolve: false, + verdicts: [], + requestReviewers: false, + }), + getChangeRequestSummary: () => + Effect.sync(() => { + summaryReads++; + return changeRequest(8, "2026-07-01T00:00:00Z"); + }), + runAction: () => + Effect.gen(function* () { + taken++; + if (mutationFails) return yield* requestFailed; + }), }), - }), - ], - }); - const input = { - projectId: "p1" as ProjectId, - repository: "acme/web", - number: 3, - action: "update-branch" as const, - updateMethod: "rebase" as const, - stackNumber: 50, - expectedStackHeads: [{ number: 3, headSha: "ccc" }], - }; - yield* service.runAction(input); - assert.strictEqual(taken, 1); - const unrelated = { ...input, number: 8 }; - yield* service.summary(unrelated); - assert.strictEqual(summaryReads, 1); - stackRebase = false; - assert.strictEqual( - (yield* Effect.flip(service.runAction(input)))._tag, - "PullRequestOperationError", - ); - stackRebase = true; - stackActions = false; - assert.strictEqual( - (yield* Effect.flip(service.runAction(input)))._tag, - "PullRequestOperationError", - ); - assert.strictEqual(taken, 1); - yield* service.summary(unrelated); - assert.strictEqual(summaryReads, 1); - stackActions = true; - mutationFails = true; - yield* Effect.flip(service.runAction(input)); - assert.strictEqual(taken, 2); - yield* service.summary(unrelated); - assert.strictEqual(summaryReads, 2); - }), -); + ], + }); + const input = { + ...(crossHost ? { host: "enterprise.test" } : {}), + projectId: "p1" as ProjectId, + repository: "acme/web", + number: 3, + action: "update-branch" as const, + updateMethod: "rebase" as const, + stackNumber: 50, + expectedStackHeads: [{ number: 3, headSha: "ccc" }], + }; + yield* service.runAction(input); + assert.strictEqual(taken, 1); + const unrelated = { ...input, number: 8 }; + yield* service.summary(unrelated); + assert.strictEqual(summaryReads, 1); + stackRebase = false; + assert.strictEqual( + (yield* Effect.flip(service.runAction(input)))._tag, + "PullRequestOperationError", + ); + stackRebase = true; + stackActions = false; + assert.strictEqual( + (yield* Effect.flip(service.runAction(input)))._tag, + "PullRequestOperationError", + ); + assert.strictEqual(taken, 1); + yield* service.summary(unrelated); + assert.strictEqual(summaryReads, 1); + stackActions = true; + mutationFails = true; + yield* Effect.flip(service.runAction(input)); + assert.strictEqual(taken, 2); + yield* service.summary(unrelated); + assert.strictEqual(summaryReads, 2); + }), + ); +} it.effect("refuses a way of updating a branch that the host or the viewer does not allow", () => Effect.gen(function* () { @@ -4773,7 +4902,7 @@ it.effect("forgets the cached detail after a rewrite or terminal turn", () => yield* service.detail(reference); assert.strictEqual(coreCalls, 2); - yield* service.refreshAfterTurn; + yield* service.refreshAfterTurn("p1" as ProjectId); yield* service.detail(reference); assert.strictEqual(coreCalls, 3); }), diff --git a/apps/server/src/pullRequest/PullRequestService.ts b/apps/server/src/pullRequest/PullRequestService.ts index e4bd807cbefd..d0d57f4ae8a6 100644 --- a/apps/server/src/pullRequest/PullRequestService.ts +++ b/apps/server/src/pullRequest/PullRequestService.ts @@ -24,6 +24,7 @@ import { pullRequestProviderRequirement, resolvePullRequestAuthorFilter, type OrchestrationProjectShell, + type ProjectId, type PullRequestAction, type PullRequestActionInput, type PullRequestActivity, @@ -67,6 +68,7 @@ import { } from "@t3tools/contracts"; import { detectSourceControlProviderFromRemoteUrl } from "@t3tools/shared/sourceControl"; +import { AllowGitHubReserve } from "../sourceControl/GitHubCli.ts"; import * as ProjectionSnapshotQuery from "../orchestration/Services/ProjectionSnapshotQuery.ts"; import * as SourceControlProviderRegistry from "../sourceControl/SourceControlProviderRegistry.ts"; import * as SourceControlRateLimit from "../sourceControl/SourceControlRateLimit.ts"; @@ -186,7 +188,7 @@ export class PullRequestService extends Context.Service< Scope.Scope >; readonly subscribeRefreshes: Stream.Stream; - readonly refreshAfterTurn: Effect.Effect; + readonly refreshAfterTurn: (projectId: ProjectId) => Effect.Effect; readonly detail: (input: PullRequestRef) => Effect.Effect; readonly activity: ( input: PullRequestRef, @@ -464,6 +466,7 @@ function withRateLimitBackoff( ), Effect.flatMap((lease) => effect.pipe( + Effect.provideService(AllowGitHubReserve, allowPaused), Effect.tap(() => limits.recordSuccess({ ...key, lease })), Effect.tapError((error) => error.reason === "rate-limited" @@ -794,6 +797,18 @@ export const make = Effect.gen(function* () { }), ); + const canonicalRef = Effect.fn("PullRequestService.canonicalRef")(function* < + I extends PullRequestRef, + >(input: I) { + const project = yield* requireProject(input); + return { + ...input, + projectId: project.project.id, + host: project.host, + repository: project.repository, + }; + }); + /** * What the signed-in account may do with this change request, asked of the host itself. Every * 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* () { .pipe( // Once the authorized provider action starts, a failure may leave partial // remote updates. Validation and permission failures above changed nothing. - Effect.ensuring(input.stackNumber === undefined ? Effect.void : refreshAfterTurn), + Effect.ensuring( + input.stackNumber === undefined + ? Effect.void + : refreshAfterTurn(project.project.id), + ), Effect.mapError(toPullRequestError("runAction")), Effect.as( project.api.kind === "azure-devops" @@ -2416,13 +2435,22 @@ export const make = Effect.gen(function* () { // scope re-entering `refEpochs` after eviction can never mint a key an old entry still has. let epochCounter = 0; let listingsEpoch = 0; - let turnRefreshEpoch = 0; const refEpochs = new Map(); + const projectEpochs = new Map(); + let projectEpochFloor = 0; const REF_EPOCH_CAPACITY = 2_048; const refScope = (ref: PullRequestRef) => - `${ref.projectId} ${ref.host?.toLowerCase() ?? ""} ${ref.repository.toLowerCase()} ${ref.number}`; + JSON.stringify([ + ref.projectId, + ref.host?.toLowerCase() ?? "", + ref.repository.toLowerCase(), + ref.number, + ]); const refEpoch = (ref: PullRequestRef) => - Math.max(turnRefreshEpoch, refEpochs.get(refScope(ref)) ?? 0); + Math.max( + projectEpochs.get(ref.projectId) ?? projectEpochFloor, + refEpochs.get(refScope(ref)) ?? 0, + ); // Keys carry the reference back out of the cache loader, so the slot layout is shared with // `refOfCacheKey` rather than read positionally at every loader. const refCacheKey = (ref: CredentialRef) => @@ -2524,7 +2552,10 @@ export const make = Effect.gen(function* () { ), ), ); - const payload = yield* readCache.get(key, encodedRead); + const payload = yield* readCache.get(key, encodedRead, [ + `project:${input.projectId}`, + refScope(input), + ]); const decoded = yield* Schema.decodeUnknownEffect(codec)(payload).pipe(Effect.option); return Option.isSome(decoded) ? decoded.value : yield* lookup; }); @@ -2723,20 +2754,9 @@ export const make = Effect.gen(function* () { const diffCache = yield* Cache.makeWith( (key: string) => { - const [, projectId, host, repository, number, cursor, commit] = JSON.parse(key) as [ - number, - string, - string | null, - string, - number, - string | null, - string | null, - ]; + const [reference, cursor, commit] = JSON.parse(key) as [string, string | null, string | null]; return diffUncached({ - projectId, - ...(host === null ? {} : { host }), - repository, - number, + ...refOfCacheKey(reference), ...(cursor === null ? {} : { cursor }), ...(commit === null ? {} : { commit }), } as PullRequestDiffInput); @@ -2745,18 +2765,14 @@ export const make = Effect.gen(function* () { capacity: DIFF_CACHE_CAPACITY, timeToLive: (exit, key) => { if (!Exit.isSuccess(exit)) return Duration.zero; - const commit = (JSON.parse(key) as ReadonlyArray)[6]; + const commit = (JSON.parse(key) as ReadonlyArray)[2]; return commit === null ? DIFF_CACHE_TTL : COMMIT_DIFF_CACHE_TTL; }, }, ); const diff: PullRequestService["Service"]["diff"] = (input) => { const key = JSON.stringify([ - refEpoch(input), - input.projectId, - input.host?.toLowerCase() ?? null, - input.repository.toLowerCase(), - input.number, + refCacheKey(input), input.cursor ?? null, input.commit ?? null, input.commit === undefined @@ -2826,7 +2842,14 @@ export const make = Effect.gen(function* () { const invalidate: PullRequestService["Service"]["invalidate"] = (input) => { const reference = input.reference; if (reference !== undefined) { - return readCache.invalidate.pipe(Effect.andThen(Effect.sync(() => bumpRefEpoch(reference)))); + return canonicalRef(reference).pipe( + Effect.flatMap((ref) => + readCache + .invalidate(refScope(ref)) + .pipe(Effect.andThen(Effect.sync(() => bumpRefEpoch(ref)))), + ), + Effect.ignore, + ); } return Effect.sync(() => { listingsEpoch = ++epochCounter; @@ -2834,12 +2857,22 @@ export const make = Effect.gen(function* () { }).pipe(Effect.andThen(Cache.invalidateAll(viewerFlights))); }; - const refreshAfterTurn: PullRequestService["Service"]["refreshAfterTurn"] = Effect.suspend(() => { - turnRefreshEpoch = listingsEpoch = ++epochCounter; - return readCache.invalidate.pipe( - Effect.andThen(SubscriptionRef.set(pullRequestRefreshes, turnRefreshEpoch)), - ); - }); + const refreshAfterTurn: PullRequestService["Service"]["refreshAfterTurn"] = (projectId) => + Effect.suspend(() => { + listingsEpoch = ++epochCounter; + projectEpochs.delete(projectId); + if (projectEpochs.size >= REF_EPOCH_CAPACITY) { + const oldest = projectEpochs.keys().next().value; + if (oldest !== undefined) { + projectEpochFloor = projectEpochs.get(oldest)!; + projectEpochs.delete(oldest); + } + } + projectEpochs.set(projectId, listingsEpoch); + return readCache + .invalidate(`project:${projectId}`) + .pipe(Effect.andThen(SubscriptionRef.set(pullRequestRefreshes, listingsEpoch))); + }); // A mutation's own client re-reads right after it, and every other client's next read must // 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* () { method: (input: I) => Effect.Effect, ): ((input: I) => Effect.Effect) => (input) => - readCache.invalidate.pipe( - Effect.andThen(method(input)), - Effect.ensuring(readCache.invalidate), - Effect.tap(() => - Effect.sync(() => { - bumpRefEpoch(input); - listingsEpoch = ++epochCounter; - }), - ), - ); + Effect.gen(function* () { + const ref = yield* canonicalRef(input); + yield* readCache.invalidate(refScope(ref)).pipe( + Effect.andThen(method(input)), + Effect.ensuring(readCache.invalidate(refScope(ref))), + Effect.tap(() => + Effect.sync(() => { + bumpRefEpoch(ref); + listingsEpoch = ++epochCounter; + }), + ), + ); + }); const runActionAndInvalidate: PullRequestService["Service"]["runAction"] = Effect.fn( "PullRequestService.runActionAndInvalidate", )(function* (input) { - yield* readCache.invalidate; - const repository = yield* runAction(input).pipe(Effect.ensuring(readCache.invalidate)); - bumpRefEpoch({ ...input, repository }); + const ref = yield* canonicalRef(input); + yield* readCache.invalidate(refScope(ref)); + const repository = yield* runAction(input).pipe( + Effect.ensuring(readCache.invalidate(refScope(ref))), + ); + bumpRefEpoch({ ...ref, repository }); listingsEpoch = ++epochCounter; if (input.action === "merge") { // A successful merge action can merely enqueue the PR or enable auto-merge. @@ -2890,26 +2929,26 @@ export const make = Effect.gen(function* () { read: (input: I, ...args: Args) => Effect.Effect, ) => (input: I, ...args: Args) => - routingCredential.pipe( - Effect.flatMap((credential) => - read( - credential === null - ? input - : { - ...input, - [credentialNamespace]: credential.credentialFingerprint, - }, - ...args, - ), - ), - ); + Effect.gen(function* () { + const ref = yield* canonicalRef(input); + const credential = yield* routingCredential; + return yield* read( + credential === null + ? ref + : { ...ref, [credentialNamespace]: credential.credentialFingerprint }, + ...args, + ); + }); return PullRequestService.of({ routing, routingIdentity, withRoutingCredential, list, - listStats, + listStats: (input) => + Effect.forEach(input.refs, (ref) => canonicalRef(ref).pipe(Effect.option)).pipe( + Effect.flatMap((refs) => listStats({ ...input, refs: refs.flatMap(Option.toArray) })), + ), summary: credentialCached(summary), stack: credentialCached(stack), subscribeMerges: PubSub.subscribe(mergedPullRequests).pipe( @@ -2922,7 +2961,7 @@ export const make = Effect.gen(function* () { detail: credentialCached(detail), activity: credentialCached(activity), threadComments, - diff, + diff: credentialCached(diff), diffFileContents, runAction: runActionAndInvalidate, update: invalidatedByMutation(update), diff --git a/apps/server/src/sourceControl/GitHubCli.test.ts b/apps/server/src/sourceControl/GitHubCli.test.ts index d3fe840fa64c..5893c21ff772 100644 --- a/apps/server/src/sourceControl/GitHubCli.test.ts +++ b/apps/server/src/sourceControl/GitHubCli.test.ts @@ -1,5 +1,8 @@ import { assert, it, afterEach, describe, expect, vi } from "@effect/vitest"; import * as Cache from "effect/Cache"; +import * as TestClock from "effect/testing/TestClock"; +import * as Clock from "effect/Clock"; +import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as PlatformError from "effect/PlatformError"; @@ -10,6 +13,8 @@ import { VcsProcessExitError, VcsProcessSpawnError } from "@t3tools/contracts"; import * as VcsProcess from "../vcs/VcsProcess.ts"; import * as GitHubCli from "./GitHubCli.ts"; +import * as GitHubGraphQlBudget from "./githubGraphQlBudget.ts"; +import * as SourceControlRateLimit from "./SourceControlRateLimit.ts"; const encodeGitHubCliError = Schema.encodeEffect(Schema.fromJsonString(GitHubCli.GitHubCliError)); @@ -21,12 +26,18 @@ const processOutput = (stdout: string): VcsProcess.VcsProcessOutput => ({ stderrTruncated: false, }); +const quotaOutput = (remaining = 5000, resetAt = "2099-01-01T00:00:00Z") => + processOutput( + JSON.stringify({ data: { rateLimit: { cost: 1, limit: 5000, remaining, resetAt } } }), + ); + const mockRun = vi.fn(); const layer = GitHubCli.layer.pipe( Layer.provide( Layer.mock(VcsProcess.VcsProcess)({ - run: mockRun, + run: (input) => + input.args[1] === "rate_limit" ? Effect.succeed(quotaOutput()) : mockRun(input), }), ), ); @@ -35,7 +46,98 @@ afterEach(() => { mockRun.mockReset(); }); +it.effect("shares quota checks, preserves the reserve, and resumes after reset", () => + Effect.gen(function* () { + let probes = 0; + const commands: string[] = []; + let remaining = 501; + let resetAt = DateTime.formatIso( + DateTime.makeUnsafe((yield* Clock.currentTimeMillis) + 60_000), + ); + const gh = yield* GitHubCli.make.pipe( + Effect.provideService(VcsProcess.VcsProcess, { + run: (input) => + Effect.sync(() => { + if (input.args[1] === "rate_limit") { + probes++; + assert.strictEqual(input.args[3], "enterprise.test"); + return quotaOutput(remaining, resetAt); + } + commands.push(input.args.slice(0, 2).join(" ")); + return processOutput("[]"); + }), + }), + ); + const read = (command: string) => + gh.execute({ + cwd: "/repo", + args: + command === "repo" + ? ["repo", "view", "enterprise.test/acme/web", "--json", "name"] + : ["pr", command, "--repo=enterprise.test/acme/web", "--json", "number"], + }); + yield* read("list"); + const failure = yield* read("view").pipe(Effect.flip); + assert.strictEqual(failure._tag, "GitHubCliRateLimitError"); + assert.strictEqual(probes, 1); + assert.deepStrictEqual(commands, ["pr list"]); + yield* read("view").pipe(Effect.provideService(GitHubCli.AllowGitHubReserve, true)); + yield* gh.execute({ cwd: "/repo", args: ["pr", "merge", "1"] }); + assert.deepStrictEqual(commands, ["pr list", "pr view", "pr merge"]); + remaining = 0; + yield* TestClock.adjust("30 seconds"); + yield* read("repo").pipe(Effect.flip); + assert.strictEqual(probes, 2); + yield* TestClock.adjust("30 seconds"); + remaining = 5000; + resetAt = DateTime.formatIso(DateTime.makeUnsafe((yield* Clock.currentTimeMillis) + 60_000)); + yield* Effect.all([read("list"), read("repo")], { concurrency: 2 }); + assert.strictEqual(probes, 3); + assert.deepStrictEqual(commands.slice(3).toSorted(), ["pr list", "repo view"]); + }).pipe(Effect.provide(Layer.merge(GitHubGraphQlBudget.layer, SourceControlRateLimit.layer))), +); + describe("GitHubCli.layer", () => { + it.effect("shares the registry budget with CLI reads through nested layer providers", () => + Effect.gen(function* () { + const budget = yield* GitHubGraphQlBudget.GitHubGraphQlBudget; + const gh = yield* GitHubCli.GitHubCli; + yield* budget.observe("github.com", quotaOutput(0).stdout); + const error = yield* gh.execute({ cwd: "/repo", args: ["pr", "list"] }).pipe(Effect.flip); + assert.strictEqual(error._tag, "GitHubCliRateLimitError"); + expect(mockRun).not.toHaveBeenCalled(); + }).pipe(Effect.provide(layer.pipe(Layer.provide(GitHubGraphQlBudget.layer)))), + ); + + it.effect("keeps quota snapshots separate for verified credentials on the same host", () => + Effect.gen(function* () { + let reads = 0; + const gh = yield* GitHubCli.make.pipe( + Effect.provideService(VcsProcess.VcsProcess, { + run: (input) => + Effect.sync(() => { + if (input.args[1] === "rate_limit") + return quotaOutput(input.env?.GH_TOKEN === "empty" ? 0 : 5000); + reads++; + return processOutput("[]"); + }), + }), + ); + const read = (token: string) => + gh.execute({ cwd: "/repo", args: ["pr", "list", "--repo", "github.com/acme/web"] }).pipe( + Effect.provideService(GitHubCli.PinnedGitHubCredential, { + host: "github.com", + token: Redacted.make(token), + credentialFingerprint: token, + }), + ); + yield* read("empty").pipe(Effect.flip); + yield* read("healthy"); + yield* read("empty").pipe(Effect.flip); + assert.strictEqual(reads, 1); + }).pipe(Effect.provide(Layer.merge(GitHubGraphQlBudget.layer, SourceControlRateLimit.layer))), + ); + it.effect("pins concurrent cached commands to their own verified credentials", () => Effect.gen(function* () { mockRun.mockImplementation((input) => @@ -523,6 +625,15 @@ describe("GitHubCli.layer", () => { assert.include(error.detail, "gh api rate_limit"); assert.strictEqual(error.cause, cause); assert.notInclude(error.message, "user ID"); + const paused = yield* gh + .execute({ cwd: "/other-repo", args: ["pr", "list"] }) + .pipe(Effect.flip); + assert.strictEqual(paused._tag, "GitHubCliRateLimitError"); + expect(mockRun).toHaveBeenCalledTimes(1); + yield* TestClock.adjust("30 seconds"); + mockRun.mockReturnValueOnce(Effect.succeed(processOutput("[]"))); + yield* gh.execute({ cwd: "/other-repo", args: ["pr", "list"] }); + expect(mockRun).toHaveBeenCalledTimes(2); }).pipe(Effect.provide(layer)), ); }); diff --git a/apps/server/src/sourceControl/GitHubCli.ts b/apps/server/src/sourceControl/GitHubCli.ts index 30b0e4a09231..c525740efeae 100644 --- a/apps/server/src/sourceControl/GitHubCli.ts +++ b/apps/server/src/sourceControl/GitHubCli.ts @@ -1,3 +1,6 @@ +import * as Cache from "effect/Cache"; +import * as Duration from "effect/Duration"; +import * as Exit from "effect/Exit"; import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; @@ -15,6 +18,8 @@ import { } from "@t3tools/contracts"; import * as VcsProcess from "../vcs/VcsProcess.ts"; +import * as GitHubGraphQlBudget from "./githubGraphQlBudget.ts"; +import * as SourceControlRateLimit from "./SourceControlRateLimit.ts"; import { decodeGitHubPullRequestJson, decodeGitHubPullRequestListJson, @@ -30,7 +35,12 @@ export const PinnedGitHubCredential = Context.Reference<{ readonly credentialFingerprint: string; } | null>("t3/sourceControl/PinnedGitHubCredential", { defaultValue: () => null }); -function targetsVerifiedHost(args: ReadonlyArray, host: string): boolean { +export const AllowGitHubReserve = Context.Reference( + "t3/sourceControl/AllowGitHubReserve", + { defaultValue: () => false }, +); + +function commandHosts(args: ReadonlyArray): Array { const hosts: Array = []; const repositoryHost = (repository: string | undefined) => { if (repository === undefined) return null; @@ -54,6 +64,11 @@ function targetsVerifiedHost(args: ReadonlyArray, host: string): boolean else if (arg.startsWith("-R")) hosts.push(repositoryHost(arg.slice(2))); else if (/^https?:\/\//i.test(arg)) hosts.push(repositoryHost(arg)); } + return hosts; +} + +function targetsVerifiedHost(args: ReadonlyArray, host: string): boolean { + const hosts = commandHosts(args); return hosts.length > 0 && hosts.every((target) => target === host); } @@ -91,7 +106,7 @@ export class GitHubCliAuthenticationError extends Schema.TaggedError()( "GitHubCliRateLimitError", - gitHubCliFailureFields, + { ...gitHubCliFailureFields, retryAt: Schema.optionalKey(Schema.Finite) }, ) { get detail(): string { 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< readonly stdin?: string; readonly env?: NodeJS.ProcessEnv; readonly maxOutputBytes?: number; + readonly rateLimitHost?: string; + readonly allowReserve?: boolean; }) => Effect.Effect; readonly listOpenPullRequests: (input: { readonly cwd: string; readonly headSelector: string; readonly limit?: number; + readonly rateLimitHost?: string; }) => Effect.Effect, GitHubCliError>; readonly getPullRequest: (input: { readonly cwd: string; readonly reference: string; + readonly rateLimitHost?: string; }) => Effect.Effect; readonly getRepositoryCloneUrls: (input: { @@ -308,6 +327,7 @@ export class GitHubCli extends Context.Service< readonly getDefaultBranch: (input: { readonly cwd: string; + readonly rateLimitHost?: string; }) => Effect.Effect; readonly checkoutPullRequest: (input: { @@ -377,8 +397,10 @@ function deriveRepositoryCloneUrlsFromCreateOutput( /** @public Service construction is part of the canonical Effect module API. */ export const make = Effect.gen(function* () { const process = yield* VcsProcess.VcsProcess; + const budget = yield* GitHubGraphQlBudget.GitHubGraphQlBudget; + const limits = yield* SourceControlRateLimit.SourceControlRateLimit; - const execute: GitHubCli["Service"]["execute"] = Effect.fn("GitHubCli.execute")( + const executeRaw: GitHubCli["Service"]["execute"] = Effect.fn("GitHubCli.executeRaw")( function* (input) { const credential = yield* PinnedGitHubCredential; if (credential !== null && !targetsVerifiedHost(input.args, credential.host)) { @@ -416,11 +438,94 @@ export const make = Effect.gen(function* () { }, ); + const quota = yield* Cache.makeWith( + (key: string) => { + const host = key.split("\0")[0]!; + return executeRaw({ + cwd: globalThis.process.cwd(), + args: [ + "api", + "rate_limit", + "--hostname", + host, + "--jq", + ".resources.graphql | {data:{rateLimit:{cost:1,limit:.limit,remaining:.remaining,resetAt:(.reset|todateiso8601)}}}", + ], + }).pipe( + Effect.tap((result) => budget.observe(host, result.stdout)), + Effect.asVoid, + ); + }, + { + capacity: 32, + timeToLive: (exit) => (Exit.isSuccess(exit) ? Duration.seconds(30) : Duration.zero), + }, + ); + const execute: GitHubCli["Service"]["execute"] = Effect.fn("GitHubCli.execute")( + function* (input) { + const [command, action] = input.args; + if ( + !( + (command === "pr" && (action === "list" || action === "view")) || + (command === "repo" && action === "view") + ) + ) + return yield* executeRaw(input); + const credential = yield* PinnedGitHubCredential; + if (credential !== null && !targetsVerifiedHost(input.args, credential.host)) + return yield* executeRaw(input); + const allowReserve = input.allowReserve ?? (yield* AllowGitHubReserve); + const host = ( + credential?.host ?? + commandHosts(input.args).find((host) => host !== null) ?? + input.rateLimitHost ?? + input.env?.GH_HOST ?? + globalThis.process.env.GH_HOST ?? + "github.com" + ).toLowerCase(); + const key = { provider: "github" as const, host }; + const guarded = Effect.gen(function* () { + const lease = yield* limits.check(key, allowReserve ? { allowPaused: true } : undefined); + return yield* Effect.gen(function* () { + yield* Cache.get(quota, `${host}\0${credential?.credentialFingerprint ?? ""}`); + yield* budget.query(host, "query {}", allowReserve ? { allowReserve: true } : undefined); + return yield* executeRaw(input); + }).pipe( + Effect.tap(() => limits.recordSuccess({ ...key, lease })), + Effect.tapError((error) => + error._tag === "GitHubCliRateLimitError" + ? limits.recordRateLimit({ ...key, lease }) + : Effect.void, + ), + ); + }); + return yield* guarded.pipe( + Effect.provideService( + SourceControlRateLimit.CredentialScope, + credential?.credentialFingerprint ?? (yield* SourceControlRateLimit.CredentialScope), + ), + Effect.catchTags({ + SourceControlRateLimitPausedError: (cause) => + Effect.fail( + new GitHubCliRateLimitError({ + command: "gh", + cwd: input.cwd, + retryAt: cause.retryAt, + cause, + }), + ), + }), + ); + }, + ); + return GitHubCli.of({ execute, listOpenPullRequests: (input) => execute({ cwd: input.cwd, + ...(input.rateLimitHost === undefined ? {} : { rateLimitHost: input.rateLimitHost }), + allowReserve: true, args: [ "pr", "list", @@ -458,6 +563,8 @@ export const make = Effect.gen(function* () { getPullRequest: (input) => execute({ cwd: input.cwd, + ...(input.rateLimitHost === undefined ? {} : { rateLimitHost: input.rateLimitHost }), + allowReserve: true, args: [ "pr", "view", @@ -533,6 +640,7 @@ export const make = Effect.gen(function* () { getDefaultBranch: (input) => execute({ cwd: input.cwd, + ...(input.rateLimitHost === undefined ? {} : { rateLimitHost: input.rateLimitHost }), args: ["repo", "view", "--json", "defaultBranchRef", "--jq", ".defaultBranchRef.name"], }).pipe( Effect.map((value) => { @@ -548,4 +656,7 @@ export const make = Effect.gen(function* () { }); }); -export const layer = Layer.effect(GitHubCli, make); +export const layer = Layer.effect(GitHubCli, make).pipe( + Layer.provideMerge(GitHubGraphQlBudget.layer), + Layer.provideMerge(SourceControlRateLimit.layer), +); diff --git a/apps/server/src/sourceControl/GitHubSourceControlProvider.test.ts b/apps/server/src/sourceControl/GitHubSourceControlProvider.test.ts index 4c41b323f17b..0d46b6eab9ea 100644 --- a/apps/server/src/sourceControl/GitHubSourceControlProvider.test.ts +++ b/apps/server/src/sourceControl/GitHubSourceControlProvider.test.ts @@ -30,6 +30,33 @@ function makeProvider(github: Partial) { ); } +it.effect("uses the enterprise quota for a current-repository default branch read", () => + Effect.gen(function* () { + const provider = yield* GitHubSourceControlProvider.make.pipe( + Effect.provide(GitHubCli.layer), + Effect.provideService(VcsProcess.VcsProcess, { + run: (input) => + Effect.sync(() => { + if (input.args[1] !== "rate_limit") return processResult("main"); + assert.strictEqual(input.args[3], "enterprise.test"); + return processResult( + '{"data":{"rateLimit":{"cost":1,"limit":5000,"remaining":5000,"resetAt":"2099-01-01T00:00:00Z"}}}', + ); + }), + }), + ); + const branch = yield* provider.getDefaultBranch({ + cwd: "/enterprise-repo", + context: { + provider: { kind: "github", name: "GitHub Enterprise", baseUrl: "https://enterprise.test" }, + remoteName: "origin", + remoteUrl: "https://enterprise.test/acme/web.git", + }, + }); + assert.strictEqual(branch, "main"); + }), +); + it.effect("maps GitHub PR summaries into provider-neutral change requests", () => Effect.gen(function* () { const provider = yield* makeProvider({ diff --git a/apps/server/src/sourceControl/GitHubSourceControlProvider.ts b/apps/server/src/sourceControl/GitHubSourceControlProvider.ts index bb8662928688..372d2a032d79 100644 --- a/apps/server/src/sourceControl/GitHubSourceControlProvider.ts +++ b/apps/server/src/sourceControl/GitHubSourceControlProvider.ts @@ -127,6 +127,9 @@ export const make = Effect.gen(function* () { .listOpenPullRequests({ cwd: input.cwd, headSelector: input.headSelector, + ...(input.context === undefined + ? {} + : { rateLimitHost: new URL(input.context.provider.baseUrl).host }), ...(input.limit !== undefined ? { limit: input.limit } : {}), }) .pipe( @@ -152,6 +155,9 @@ export const make = Effect.gen(function* () { return github .execute({ cwd: input.cwd, + ...(input.context === undefined + ? {} + : { rateLimitHost: new URL(input.context.provider.baseUrl).host }), args: [ "pr", "list", @@ -267,23 +273,30 @@ export const make = Effect.gen(function* () { }, listChangeRequests, getChangeRequest: (input) => - github.getPullRequest(input).pipe( - Effect.map(toChangeRequest), - Effect.mapError( - (error) => - new SourceControlProviderError({ - provider: "github", - operation: "getChangeRequest", - command: error.command, - cwd: input.cwd, - reference: SourceControlProvider.transportSafeSourceControlErrorValue( - input.reference, - ), - detail: error.detail, - cause: error, - }), + github + .getPullRequest({ + ...input, + ...(input.context === undefined + ? {} + : { rateLimitHost: new URL(input.context.provider.baseUrl).host }), + }) + .pipe( + Effect.map(toChangeRequest), + Effect.mapError( + (error) => + new SourceControlProviderError({ + provider: "github", + operation: "getChangeRequest", + command: error.command, + cwd: input.cwd, + reference: SourceControlProvider.transportSafeSourceControlErrorValue( + input.reference, + ), + detail: error.detail, + cause: error, + }), + ), ), - ), createChangeRequest: (input) => github .createPullRequest({ @@ -344,19 +357,26 @@ export const make = Effect.gen(function* () { ), ), getDefaultBranch: (input) => - github.getDefaultBranch(input).pipe( - Effect.mapError( - (error) => - new SourceControlProviderError({ - provider: "github", - operation: "getDefaultBranch", - command: error.command, - cwd: input.cwd, - detail: error.detail, - cause: error, - }), + github + .getDefaultBranch({ + ...input, + ...(input.context === undefined + ? {} + : { rateLimitHost: new URL(input.context.provider.baseUrl).host }), + }) + .pipe( + Effect.mapError( + (error) => + new SourceControlProviderError({ + provider: "github", + operation: "getDefaultBranch", + command: error.command, + cwd: input.cwd, + detail: error.detail, + cause: error, + }), + ), ), - ), checkoutChangeRequest: (input) => github.checkoutPullRequest(input).pipe( Effect.mapError( diff --git a/apps/server/src/sourceControl/SourceControlRateLimit.test.ts b/apps/server/src/sourceControl/SourceControlRateLimit.test.ts index 5ee233a367dd..e46005c73f0e 100644 --- a/apps/server/src/sourceControl/SourceControlRateLimit.test.ts +++ b/apps/server/src/sourceControl/SourceControlRateLimit.test.ts @@ -6,6 +6,22 @@ import * as SourceControlRateLimit from "./SourceControlRateLimit.ts"; const github = { provider: "github" as const, host: "github.com" }; +it.effect("isolates cooldowns for verified credentials on the same host", () => + Effect.gen(function* () { + const limits = yield* SourceControlRateLimit.SourceControlRateLimit; + yield* limits + .recordRateLimit({ ...github, lease: 0 }) + .pipe(Effect.provideService(SourceControlRateLimit.CredentialScope, "first")); + yield* limits + .check(github) + .pipe(Effect.provideService(SourceControlRateLimit.CredentialScope, "second")); + const error = yield* limits + .check(github) + .pipe(Effect.provideService(SourceControlRateLimit.CredentialScope, "first"), Effect.flip); + assert.strictEqual(error._tag, "SourceControlRateLimitPausedError"); + }).pipe(Effect.provide(SourceControlRateLimit.layer)), +); + it("parses Retry-After seconds and HTTP dates", () => { assert.equal(SourceControlRateLimit.retryAtFromHeader("120", 1_000), 121_000); assert.equal( diff --git a/apps/server/src/sourceControl/SourceControlRateLimit.ts b/apps/server/src/sourceControl/SourceControlRateLimit.ts index b936c456079b..dc6242a60eb6 100644 --- a/apps/server/src/sourceControl/SourceControlRateLimit.ts +++ b/apps/server/src/sourceControl/SourceControlRateLimit.ts @@ -13,6 +13,10 @@ import { const FALLBACK_COOLDOWN = Duration.seconds(30); const MAX_FALLBACK_COOLDOWN = Duration.minutes(15); +export const CredentialScope = Context.Reference("t3/sourceControl/CredentialScope", { + defaultValue: () => "", +}); + interface RateLimitKey { readonly provider: SourceControlProviderKind; readonly host: string; @@ -59,8 +63,8 @@ export class SourceControlRateLimit extends Context.Service< } >()("t3/sourceControl/SourceControlRateLimit") {} -function normalizedKey(key: RateLimitKey): string { - return `${key.provider}\0${key.host.trim().toLowerCase()}`; +function normalizedKey(key: RateLimitKey, scope: string): string { + return `${key.provider}\0${key.host.trim().toLowerCase()}\0${scope}`; } function fallbackCooldownMs(attempt: number): number { @@ -90,7 +94,8 @@ export const make = Effect.gen(function* () { "SourceControlRateLimit.check", )(function* (input, options) { const now = yield* Clock.currentTimeMillis; - const entry = (yield* Ref.get(entries)).get(normalizedKey(input)); + const key = normalizedKey(input, yield* CredentialScope); + const entry = (yield* Ref.get(entries)).get(key); if (entry !== undefined && entry.retryAt > now && options?.allowPaused !== true) { return yield* new SourceControlRateLimitPausedError({ provider: input.provider, @@ -105,8 +110,8 @@ export const make = Effect.gen(function* () { "SourceControlRateLimit.recordRateLimit", )(function* (input) { const now = yield* Clock.currentTimeMillis; + const key = normalizedKey(input, yield* CredentialScope); yield* Ref.update(entries, (current) => { - const key = normalizedKey(input); const previous = current.get(key); if (previous !== undefined && previous.generation > input.lease) { if (previous.retryAt <= now && (input.retryAt === undefined || input.retryAt <= now)) { @@ -145,8 +150,8 @@ export const make = Effect.gen(function* () { "SourceControlRateLimit.recordSuccess", )(function* (input) { const now = yield* Clock.currentTimeMillis; + const key = normalizedKey(input, yield* CredentialScope); yield* Ref.update(entries, (current) => { - const key = normalizedKey(input); const previous = current.get(key); if (previous === undefined || previous.generation !== input.lease || previous.retryAt > now) { return current; diff --git a/apps/server/src/sourceControl/githubGraphQlBudget.test.ts b/apps/server/src/sourceControl/githubGraphQlBudget.test.ts index a5b0680fafc5..26f9747f3202 100644 --- a/apps/server/src/sourceControl/githubGraphQlBudget.test.ts +++ b/apps/server/src/sourceControl/githubGraphQlBudget.test.ts @@ -3,6 +3,7 @@ import * as Effect from "effect/Effect"; import * as TestClock from "effect/testing/TestClock"; import * as GitHubGraphQlBudget from "./githubGraphQlBudget.ts"; +import { CredentialScope } from "./SourceControlRateLimit.ts"; const RESET_AT = "2026-08-13T14:00:00.000Z"; const NEXT_RESET_AT = "2026-08-13T15:00:00.000Z"; @@ -71,6 +72,23 @@ describe("GitHub GraphQL budget", () => { }).pipe(Effect.provide(GitHubGraphQlBudget.layer)), ); + it.effect("isolates query reservations and observations by credential", () => + Effect.gen(function* () { + yield* TestClock.setTime(BEFORE_RESET); + const budget = yield* GitHubGraphQlBudget.GitHubGraphQlBudget; + const query = budget.query("github.com", "query { viewer { login } }"); + yield* budget + .observe("github.com", rateLimit(0)) + .pipe(Effect.provideService(CredentialScope, "first")); + yield* budget + .observe("github.com", rateLimit(5000)) + .pipe(Effect.provideService(CredentialScope, "second")); + yield* query.pipe(Effect.provideService(CredentialScope, "second")); + const error = yield* query.pipe(Effect.provideService(CredentialScope, "first"), Effect.flip); + expect(error.retryAt).toBe(Date.parse(RESET_AT)); + }).pipe(Effect.provide(GitHubGraphQlBudget.layer)), + ); + it.effect("keeps the lower remaining value from out-of-order responses", () => Effect.gen(function* () { yield* TestClock.setTime(BEFORE_RESET); @@ -181,6 +199,19 @@ describe("GitHub GraphQL budget", () => { }).pipe(Effect.provide(GitHubGraphQlBudget.layer)), ); + it.effect("stops interactive reads when the reserve is exhausted", () => + Effect.gen(function* () { + yield* TestClock.setTime(BEFORE_RESET); + const budget = yield* GitHubGraphQlBudget.GitHubGraphQlBudget; + yield* budget.observe("github.com", rateLimit(1, 5000, RESET_AT, 1)); + yield* budget.query("github.com", "query { viewer { login } }", { allowReserve: true }); + const error = yield* budget + .query("github.com", "query { viewer { login } }", { allowReserve: true }) + .pipe(Effect.flip); + expect(error.retryAt).toBe(Date.parse(RESET_AT)); + }).pipe(Effect.provide(GitHubGraphQlBudget.layer)), + ); + it.effect("ignores malformed or partial rate metadata", () => Effect.gen(function* () { yield* TestClock.setTime(BEFORE_RESET); diff --git a/apps/server/src/sourceControl/githubGraphQlBudget.ts b/apps/server/src/sourceControl/githubGraphQlBudget.ts index 05e4b3a5ce24..1021f6cde9fa 100644 --- a/apps/server/src/sourceControl/githubGraphQlBudget.ts +++ b/apps/server/src/sourceControl/githubGraphQlBudget.ts @@ -85,8 +85,8 @@ export const make = Effect.gen(function* () { function* (host, document, options) { if (!isReadOperation(document)) return document; const now = yield* Clock.currentTimeMillis; + const key = `${hostKey(host)}\0${yield* SourceControlRateLimit.CredentialScope}`; const retryAt = yield* Ref.modify(snapshots, (current) => { - const key = hostKey(host); const snapshot = current.get(key); if (snapshot === undefined) return [null, current] as const; if (snapshot.resetAtMs <= now) { @@ -94,8 +94,11 @@ export const make = Effect.gen(function* () { next.delete(key); return [null, next] as const; } - const remaining = Math.max(0, snapshot.remaining - Math.max(1, snapshot.cost)); - if (options?.allowReserve !== true && remaining < snapshot.limit * GRAPHQL_RESERVE_RATIO) { + const remaining = snapshot.remaining - Math.max(1, snapshot.cost); + if ( + remaining < 0 || + (options?.allowReserve !== true && remaining < snapshot.limit * GRAPHQL_RESERVE_RATIO) + ) { return [snapshot.resetAtMs, current] as const; } const next = new Map(current); @@ -118,8 +121,8 @@ export const make = Effect.gen(function* () { )(function* (host, raw) { const snapshot = snapshotFrom(raw); if (snapshot === null) return; + const key = `${hostKey(host)}\0${yield* SourceControlRateLimit.CredentialScope}`; yield* Ref.update(snapshots, (current) => { - const key = hostKey(host); const previous = current.get(key); // Concurrent reads can finish out of order. Quota only falls within one reset window, and // an answer from an older window must not replace the current one. diff --git a/packages/client-runtime/src/state/pullRequestRouting.ts b/packages/client-runtime/src/state/pullRequestRouting.ts index d058f6a863a6..555c870d04ef 100644 --- a/packages/client-runtime/src/state/pullRequestRouting.ts +++ b/packages/client-runtime/src/state/pullRequestRouting.ts @@ -49,6 +49,17 @@ const writes = new Set([ ]); const isRef = Schema.is(PullRequestRef); const isInvalidation = Schema.is(PullRequestInvalidateInput); +const readTimeout = (environmentId: EnvironmentId) => + Effect.timeoutOrElse({ + duration: "30 seconds", + orElse: () => + Effect.fail( + new EnvironmentRpcUnavailableError({ + environmentId, + message: "The environment did not respond to the PR request.", + }), + ), + }); interface RoutedRead { origin: EnvironmentId; reference: PullRequestRef; @@ -318,7 +329,8 @@ export function createPullRequestRouter() { if (!(yield* routingAllowed(registry, origin.target.environmentId, id, writes.has(tag)))) return yield* visit(index + 1); } - return yield* run(id).pipe( + const operation = run(id); + return yield* (reads.has(tag) ? operation.pipe(readTimeout(id)) : operation).pipe( Effect.catch((error) => { if ( (reads.has(tag) || rejectedBeforeDispatch(error)) && @@ -366,12 +378,15 @@ export function createPullRequestRouter() { } if (!allowed) return yield* request(tag, input); const strictInput = { ...input, allowStale: false }; - const source = yield* Effect.cached(request(tag, strictInput)); - // Cached source reads usually finish before another environment can verify its account. - // Hedge slow reads only; never race mutations or retry an ambiguous write. - return yield* Effect.race( - source, - routedRequest(tag, strictInput, source).pipe(Effect.delay("75 millis")), + const source = yield* Effect.cached( + request(tag, strictInput).pipe(readTimeout(origin.target.environmentId)), + ); + const sourceEntry = entries.get(origin.target.environmentId); + const routed = routedRequest(tag, strictInput, source); + return yield* ( + sourceEntry !== undefined && isLocal(sourceEntry) + ? source.pipe(Effect.catch(() => routed)) + : routed ).pipe( Effect.catch((error) => input.allowStale !== false && diff --git a/packages/client-runtime/src/state/pullRequests.test.ts b/packages/client-runtime/src/state/pullRequests.test.ts index 670575f6bbe1..5fcf5fdf681e 100644 --- a/packages/client-runtime/src/state/pullRequests.test.ts +++ b/packages/client-runtime/src/state/pullRequests.test.ts @@ -7,6 +7,9 @@ import { } from "@t3tools/contracts"; import { expect, it } from "@effect/vitest"; import * as Data from "effect/Data"; +import * as Deferred from "effect/Deferred"; +import * as Fiber from "effect/Fiber"; +import * as TestClock from "effect/testing/TestClock"; import * as Effect from "effect/Effect"; import * as Latch from "effect/Latch"; import * as Layer from "effect/Layer"; @@ -52,7 +55,7 @@ for (const scenario of [ "prefers the local environment with the same github account", "falls back before mutation when the local account differs", "never retries an ambiguous mutation failure", - "returns a fast source read without checking alternate identities", + "returns a fast local source read without checking alternate identities", "keeps single-environment requests free of identity lookups", "keeps a local origin ahead of another local environment", "keeps mutations on an old origin server without retrying them", @@ -79,10 +82,12 @@ for (const scenario of [ switchedAccount; const ambiguous = scenario === "never retries an ambiguous mutation failure"; const reading = - scenario === "returns a fast source read without checking alternate identities"; + scenario === "returns a fast local source read without checking alternate identities"; const single = scenario === "keeps single-environment requests free of identity lookups"; const localOrigin = - scenario === "keeps a local origin ahead of another local environment" || switchedAccount; + scenario === "keeps a local origin ahead of another local environment" || + switchedAccount || + reading; const oldOrigin = scenario === "keeps mutations on an old origin server without retrying them"; const oldAlternate = @@ -557,15 +562,15 @@ for (const probe of ["origin", "alternate"] as const) { ); } -for (const source of ["pending", "pending-local", "failed", "offline"] as const) { - it.live( +for (const source of ["pending", "pending-local", "failed-local", "failed", "offline"] as const) { + it.effect( source === "offline" ? "returns held source data only after both fresh paths fail" - : `hedges a ${source} source read to local and interrupts the losing read`, + : `uses one shared reader with a ${source} source`, () => Effect.scoped( Effect.gen(function* () { - let interrupted = false; + const started = yield* Deferred.make(); const calls: string[] = []; const clientFor = (local: boolean) => ({ @@ -590,21 +595,16 @@ for (const source of ["pending", "pending-local", "failed", "offline"] as const) operation: "summary", detail: "github unreachable", }); - return yield* Effect.never.pipe( - Effect.onInterrupt(() => - Effect.sync(() => { - interrupted = true; - }), - ), - ); + yield* Deferred.succeed(started, undefined); + return yield* Effect.never; }), }) as unknown as WsRpcProtocolClient; const { environmentRegistry, supervisor } = yield* makeTestRuntime( clientFor(false), clientFor(true), - source === "pending-local", + source === "failed-local" || source === "pending-local", ); - const result = yield* createPullRequestRouter()(WS_METHODS.pullRequestsSummary, { + const request = createPullRequestRouter()(WS_METHODS.pullRequestsSummary, { projectId: ProjectId.make("project-1"), repository: "acme/web", number: 7, @@ -613,14 +613,23 @@ for (const source of ["pending", "pending-local", "failed", "offline"] as const) Effect.provideService(GitHubRoutingPermissions, trustedRouting), Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), ); + const fiber = yield* request.pipe(Effect.forkChild); + if (source === "pending-local") { + yield* Deferred.await(started); + yield* TestClock.adjust("30 seconds"); + } + const result = yield* Fiber.join(fiber); if (source === "offline") { expect(result).toEqual({ state: "open" }); - expect(calls).toEqual(["origin", "local", "held"]); + expect(calls).toEqual(["local", "origin", "held"]); } else { expect(result).toBeNull(); - expect(calls).toEqual(["origin", "local"]); + expect(calls).toEqual( + source === "failed-local" || source === "pending-local" + ? ["origin", "local"] + : ["local"], + ); } - expect(interrupted).toBe(source === "pending" || source === "pending-local"); }), ), );