# Files changed · PR \#23889 · pingdotgg/t3code

[View on GitCafe](https://git.cafe/pingdotgg/t3code/pull/23889/diffs)

Repository: [pingdotgg/t3code](https://git.cafe/pingdotgg/t3code)

Visibility: public

[fix\(bitbucket\): stop the startup sweep from flooding Bitbucket with 429s](https://git.cafe/pingdotgg/t3code/pull/23889.md)

Version: 23

Head commit: 11753719dd93192fe8015c535feb19430f8ef5ce

Base commit: 4812dcad02a86874533540e6a0c4b7f5eaea297a

Comparison base: 4812dcad02a86874533540e6a0c4b7f5eaea297a

Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing.

## apps/server/src/git/GitManager.ts

Status: modified; +7; −2

```diff
@@ -95,6 +95,7 @@
 import type { GitManagerServiceError } from "@t3tools/contracts";
 import * as GitVcsDriver from "../vcs/GitVcsDriver.ts";
 import * as SourceControlProvider from "@t3tools/source-control-core/server/SourceControlProvider";
+import * as SourceControlRateLimit from "@t3tools/source-control-core/server/SourceControlRateLimit";
 import * as SourceControlProviderRegistry from "../sourceControl/SourceControlProviderRegistry.ts";
 import type { ChangeRequest } from "@t3tools/contracts";
 
@@ -2715,6 +2716,9 @@
     },
   );
 
+  // A user is waiting on these, so a host's background rate-limit pause must not refuse them.
+  const asUserAction = Effect.provideService(SourceControlRateLimit.Interactive, true);
+
   const resolvePullRequest: GitManager["Service"]["resolvePullRequest"] = Effect.fn(
     "resolvePullRequest",
   )(function* (input) {
@@ -2726,7 +2730,7 @@
       .pipe(Effect.map((resolved) => toResolvedPullRequest(resolved)));
 
     return { pullRequest };
+  }, asUserAction);
-  });
 
   const ensureNoTrackedChanges = Effect.fn("ensureNoTrackedChanges")(function* (
     cwd: string,
@@ -3036,7 +3040,7 @@
         isOnPullRequestHead: true,
       };
     }).pipe(Effect.ensuring(invalidateStatus(input.cwd)));
-  });
+  }, asUserAction);
 
   const runFeatureBranchStep = Effect.fn("runFeatureBranchStep")(function* (
     settings: SourceControlTextGenerationSettings,
@@ -3281,6 +3285,7 @@
         ),
       );
     },
+    asUserAction,
   );
 
   return GitManager.of({
```

## apps/server/src/sourceControl/BitbucketApi.test.ts

Status: modified; +92; −1

```diff
@@ -3,14 +3,19 @@
 import * as NodeServices from "@effect/platform-node/NodeServices";
 import * as ConfigProvider from "effect/ConfigProvider";
 import * as DateTime from "effect/DateTime";
+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 Option from "effect/Option";
+import * as Queue from "effect/Queue";
 import * as Schema from "effect/Schema";
+import * as TestClock from "effect/testing/TestClock";
 import { HttpClient, HttpClientError, HttpClientRequest, HttpClientResponse } from "effect/http";
 
 import * as BitbucketApi from "@t3tools/source-control-bitbucket/server/BitbucketApi";
+import * as SourceControlRateLimit from "@t3tools/source-control-core/server/SourceControlRateLimit";
 import * as ServerSettings from "../serverSettings.ts";
 import * as GitVcsDriver from "../vcs/GitVcsDriver.ts";
 import * as VcsDriverRegistry from "../vcs/VcsDriverRegistry.ts";
@@ -60,6 +65,8 @@
 
 function makeLayer(input: {
   readonly response: (request: HttpClientRequest.HttpClientRequest) => Response;
+  /** Holds the answer until it resolves, in place of answering at once. */
+  readonly answer?: (response: Response) => Effect.Effect<Response>;
   readonly requestFailure?: (
     request: HttpClientRequest.HttpClientRequest,
   ) => HttpClientError.HttpClientError;
@@ -72,7 +79,9 @@
       ? Effect.never
       : input.requestFailure
         ? Effect.fail(input.requestFailure(request))
+        : (input.answer ?? Effect.succeed)(input.response(request)).pipe(
-        : Effect.succeed(HttpClientResponse.fromWeb(request, input.response(request))),
+            Effect.map((response) => HttpClientResponse.fromWeb(request, response)),
+          ),
   );
   const gitMock = {
     readConfigValue: vi.fn<GitVcsDriver.GitVcsDriver["Service"]["readConfigValue"]>(() =>
@@ -144,6 +153,7 @@
 
   const layer = BitbucketApi.layer.pipe(
     Layer.provide(ServerSourceControlHost.layer),
+    Layer.provide(SourceControlRateLimit.layer),
     Layer.provide(Layer.mock(VcsProcess.VcsProcess)({})),
     Layer.provide(
       Layer.succeed(
@@ -736,3 +746,84 @@
     ),
   ),
 );
+
+it.effect("pauses background reads after a 429 until Bitbucket's retry-after passes", () => {
+  let answered = 0;
+  const { execute, layer } = makeLayer({
+    response: () => {
+      answered += 1;
+      return answered === 1
+        ? new Response("Rate limit exceeded", { status: 429, headers: { "retry-after": "60" } })
+        : Response.json({ values: [] });
+    },
+  });
+
+  return Effect.gen(function* () {
+    const bitbucket = yield* BitbucketApi.BitbucketApi;
+    const lookup = bitbucket.listPullRequests({
+      cwd: "/repo",
+      headSelector: "feature/a",
+      state: "all",
+    });
+
+    yield* Effect.flip(lookup);
+    const paused = yield* Effect.flip(lookup);
+    // Keyed by the repository host, where PullRequestService records its own backoff.
+    assert.deepInclude(paused, {
+      _tag: "SourceControlRateLimitPausedError",
+      host: "bitbucket.org",
+      retryAt: 60_000,
+    });
+    assert.strictEqual(execute.mock.calls.length, 1);
+
+    // A user waiting on an answer is not held back by the background pause, and their success
+    // does not end it early.
+    yield* lookup.pipe(Effect.provideService(SourceControlRateLimit.Interactive, true));
+    yield* bitbucket.request({
+      method: "POST",
+      url: "/repositories/acme/web/pullrequests/1/approve",
+    });
+
+    yield* TestClock.adjust("59 seconds");
+    assert.strictEqual((yield* Effect.flip(lookup))._tag, "SourceControlRateLimitPausedError");
+
+    yield* TestClock.adjust("1 second");
+    yield* lookup;
+  }).pipe(Effect.provide(layer));
+});
+
+it.effect("keeps a sweep over many branches to a few Bitbucket requests at a time", () =>
+  Effect.gen(function* () {
+    const started = yield* Queue.unbounded<void>();
+    const release = yield* Deferred.make<void>();
+    let inFlight = 0;
+    let peak = 0;
+    const { layer } = makeLayer({
+      response: () => Response.json({ values: [] }),
+      answer: (response) =>
+        Effect.gen(function* () {
+          inFlight += 1;
+          peak = Math.max(peak, inFlight);
+          yield* Queue.offer(started, undefined);
+          yield* Deferred.await(release);
+          inFlight -= 1;
+          return response;
+        }),
+    });
+
+    yield* Effect.gen(function* () {
+      const bitbucket = yield* BitbucketApi.BitbucketApi;
+      const sweep = yield* Effect.forEach(
+        Array.from({ length: 12 }, (_, index) => `feature/${index}`),
+        (branch) =>
+          bitbucket.listPullRequests({ cwd: "/repo", headSelector: branch, state: "all" }),
+        { concurrency: "unbounded", discard: true },
+      ).pipe(Effect.forkChild);
+      for (let index = 0; index < 4; index += 1) yield* Queue.take(started);
+      yield* Deferred.succeed(release, undefined);
+      yield* Fiber.join(sweep);
+    }).pipe(Effect.provide(layer));
+
+    assert.strictEqual(peak, 4);
+  }),
+);
```

## apps/server/src/sourceControl/SourceControlRepositoryService.ts

Status: modified; +18; −14

```diff
@@ -24,6 +24,7 @@
 
 import * as ServerConfig from "../config.ts";
 import { expandHomePath } from "@t3tools/provider-core/server/pathExpansion";
+import * as SourceControlRateLimit from "@t3tools/source-control-core/server/SourceControlRateLimit";
 import {
   parseGitCloneProgressLine,
   type GitCloneProgressLine,
@@ -231,20 +232,23 @@
     );
   };
 
-  const lookupRepository = Effect.fn("SourceControlRepositoryService.lookupRepository")(function* (
-    input: SourceControlRepositoryLookupInput,
-  ) {
-    const providerKind = yield* ensureConcreteProvider({
-      operation: "lookupRepository",
-      provider: input.provider,
-    });
-    const provider = yield* providers.get(providerKind);
-    const urls = yield* provider.getRepositoryCloneUrls({
-      cwd: input.cwd ?? config.cwd,
-      repository: input.repository.trim(),
-    });
-    return toRepositoryInfo(providerKind, urls);
-  });
+  const lookupRepository = Effect.fn("SourceControlRepositoryService.lookupRepository")(
+    function* (input: SourceControlRepositoryLookupInput) {
+      const providerKind = yield* ensureConcreteProvider({
+        operation: "lookupRepository",
+        provider: input.provider,
+      });
+      const provider = yield* providers.get(providerKind);
+      const urls = yield* provider.getRepositoryCloneUrls({
+        cwd: input.cwd ?? config.cwd,
+        repository: input.repository.trim(),
+      });
+      return toRepositoryInfo(providerKind, urls);
+    },
+    // Someone adding or cloning a repository is waiting, so a background rate-limit pause must not
+    // refuse the lookup.
+    Effect.provideService(SourceControlRateLimit.Interactive, true),
+  );
 
   const normalizeDestinationPath = Effect.fn("SourceControlRepositoryService.normalizeDestination")(
     function* (destinationPath: string) {
```

## apps/server/src/sourceControl/builtInDrivers.ts

Status: modified; +4; −1

```diff
@@ -16,6 +16,7 @@
 import * as GitLabPullRequestCli from "@t3tools/source-control-gitlab/server/GitLabPullRequestCli";
 import * as GitLabDriver from "@t3tools/source-control-gitlab/server/driver";
 import type { SourceControlDriver } from "@t3tools/source-control-core/server/driver";
+import * as SourceControlRateLimit from "@t3tools/source-control-core/server/SourceControlRateLimit";
 import * as Layer from "effect/Layer";
 
 import * as ServerSourceControlHost from "./ServerSourceControlHost.ts";
@@ -41,7 +42,9 @@
   // reader shares, so the server builds it once, here.
   GitHubPullRequestApi.layer.pipe(Layer.provideMerge(GitHubRest.layerWithDependencies)),
   AzureDevOpsPullRequestCli.layer.pipe(Layer.provideMerge(AzureDevOpsCli.layer)),
+  BitbucketPullRequestApi.layer.pipe(
+    Layer.provideMerge(BitbucketApi.layer.pipe(Layer.provide(SourceControlRateLimit.layer))),
-  BitbucketPullRequestApi.layer.pipe(Layer.provideMerge(BitbucketApi.layer)),
+  ),
   ForgejoCli.layer,
   GitCafeApi.layer.pipe(Layer.provideMerge(GitCafeCredentials.layer)),
   GitLabPullRequestCli.layer.pipe(Layer.provideMerge(GitLabCli.layer)),
```

## packages/source-control-bitbucket/src/server/BitbucketApi.ts

Status: modified; +90; −45

```diff
@@ -6,8 +6,10 @@
 import * as Layer from "effect/Layer";
 import * as Option from "effect/Option";
 import * as Schema from "effect/Schema";
+import * as Semaphore from "effect/Semaphore";
 import {
   NonNegativeInt,
+  SourceControlProviderKind,
   TrimmedNonEmptyString,
   type SourceControlProviderAuth,
   type SourceControlRepositoryCloneUrls,
@@ -32,12 +34,14 @@
 import { readSourceControlHostSettings } from "@t3tools/source-control-core/client/definition";
 
 import * as BitbucketClient from "../client/definition.ts";
+import * as SourceControlRateLimit from "@t3tools/source-control-core/server/SourceControlRateLimit";
-import { retryAtFromHeader } from "@t3tools/source-control-core/server/SourceControlRateLimit";
 
 const DEFAULT_API_BASE_URL = "https://api.bitbucket.org/2.0";
 const DEFAULT_MAX_RESPONSE_BYTES = 8 * 1024 * 1024;
 const MAX_REDIRECTS = 3;
 const BITBUCKET_API_REQUEST_TIMEOUT = "30 seconds";
+/** A sweep over many branches would otherwise reach Bitbucket as one burst of lookups. */
+const CONCURRENCY = 4;
 
 const BitbucketApiEnvConfig = Config.all({
   baseUrl: Config.String("T3CODE_BITBUCKET_API_BASE_URL").pipe(
@@ -262,6 +266,7 @@
   BitbucketRepositoryRemoteNotFoundError,
   BitbucketPullRequestBodyReadError,
   BitbucketCheckoutError,
+  SourceControlRateLimit.SourceControlRateLimitPausedError,
 ]);
 export type BitbucketApiError = typeof BitbucketApiError.Type;
 const isBitbucketApiError = Schema.is(BitbucketApiError);
@@ -607,7 +612,7 @@
   response: HttpClientResponse.HttpClientResponse,
 ): Effect.fn.Return<never, BitbucketApiError> {
   const now = yield* Clock.currentTimeMillis;
-  const retryAt = retryAtFromHeader(response.headers["retry-after"], now);
+  const retryAt = SourceControlRateLimit.retryAtFromHeader(response.headers["retry-after"], now);
   const collected = yield* collectUint8StreamText({
     stream: response.stream,
     maxBytes: DEFAULT_MAX_RESPONSE_BYTES,
@@ -637,6 +642,13 @@
   const git = host.git;
   const httpClient = yield* HttpClient.HttpClient;
   const fileSystem = yield* FileSystem.FileSystem;
+  const limits = yield* SourceControlRateLimit.SourceControlRateLimit;
+  const slots = yield* Semaphore.make(CONCURRENCY);
+  // The repository host PullRequestService keys its own backoff under, so both share one pause.
+  const rateLimitKey = {
+    provider: SourceControlProviderKind.make("bitbucket"),
+    host: "bitbucket.org",
+  };
 
   const apiUrl = (path: string) => `${config.baseUrl.replace(/\/+$/u, "")}${path}`;
 
@@ -673,6 +685,33 @@
       ),
     );
 
+  /**
+   * Every request waits for one of the shared slots, then honors the pause a 429 starts for the
+   * whole host. Each queued lookup would otherwise spend its own 429 and keep the quota exhausted.
+   * Background reads fail at once while paused; a write or an interactive read has a user waiting
+   * on it, so it still goes out.
+   */
+  const throttled = Effect.fn("BitbucketApi.throttled")(
+    function* <A, R>(method: string, effect: Effect.Effect<A, BitbucketApiError, R>) {
+      const interactive = yield* SourceControlRateLimit.Interactive;
+      const lease = yield* limits.check(
+        rateLimitKey,
+        interactive || method !== "GET" ? { allowPaused: true } : undefined,
+      );
+      return yield* effect.pipe(
+        Effect.tap(() => limits.recordSuccess({ ...rateLimitKey, lease })),
+        Effect.tapError((error) =>
+          (error._tag === "BitbucketResponseError" ||
+            error._tag === "BitbucketResponseBodyReadError") &&
+          error.status === 429
+            ? limits.recordRateLimit({ ...rateLimitKey, lease, retryAt: error.retryAt })
+            : Effect.void,
+        ),
+      );
+    },
+    (effect) => slots.withPermit(effect),
+  );
+
   const decodeResponse = <S extends Schema.Top>(
     operation: BitbucketApiOperation,
     schema: S,
@@ -698,20 +737,23 @@
     request: HttpClientRequest.HttpClientRequest,
     schema: S,
   ): Effect.Effect<S["Type"], BitbucketApiError, S["DecodingServices"]> =>
-    withAuth(request.pipe(HttpClientRequest.acceptJson)).pipe(
-      Effect.flatMap(httpClient.execute),
-      Effect.mapError(
-        (cause) =>
-          new BitbucketRequestError({
-            operation,
-            cause,
-          }),
+    throttled(
+      request.method,
+      withAuth(request.pipe(HttpClientRequest.acceptJson)).pipe(
+        Effect.flatMap(httpClient.execute),
+        Effect.mapError(
+          (cause) =>
+            new BitbucketRequestError({
+              operation,
+              cause,
+            }),
+        ),
+        Effect.flatMap((response) => decodeResponse(operation, schema, response)),
+        Effect.timeout(BITBUCKET_API_REQUEST_TIMEOUT),
+        Effect.catchTags({
+          TimeoutError: (cause) => Effect.fail(new BitbucketRequestError({ operation, cause })),
+        }),
       ),
-      Effect.flatMap((response) => decodeResponse(operation, schema, response)),
-      Effect.timeout(BITBUCKET_API_REQUEST_TIMEOUT),
-      Effect.catchTags({
-        TimeoutError: (cause) => Effect.fail(new BitbucketRequestError({ operation, cause })),
-      }),
     );
 
   const resolveRepository = Effect.fn("BitbucketApi.resolveRepository")(function* (input: {
@@ -913,38 +955,41 @@
   };
 
   const request: BitbucketApi["Service"]["request"] = (input) =>
+    throttled(
+      input.method,
+      send({ ...input, redirects: 0 }).pipe(
+        Effect.flatMap((response) =>
+          HttpClientResponse.matchStatus({
-    send({ ...input, redirects: 0 }).pipe(
-      Effect.flatMap((response) =>
-        HttpClientResponse.matchStatus({
-          // Read through the body stream rather than `text`, so an oversized diff is stopped as
-          // it arrives instead of being materialized whole and then cut.
-          "2xx": (success) =>
+            // Read through the body stream rather than `text`, so an oversized diff is stopped as
+            // it arrives instead of being materialized whole and then cut.
+            "2xx": (success) =>
+              collectUint8StreamText({
+                stream: success.stream,
+                maxBytes: input.maxBytes ?? DEFAULT_MAX_RESPONSE_BYTES,
+              }).pipe(
+                Effect.mapError(
+                  (cause) =>
+                    new BitbucketResponseBodyReadError({
+                      operation: "request",
+                      status: success.status,
+                      cause,
+                    }),
+                ),
+                Effect.map((collected) => ({
+                  body: collected.text,
+                  truncated: collected.truncated,
+                })),
-            collectUint8StreamText({
-              stream: success.stream,
-              maxBytes: input.maxBytes ?? DEFAULT_MAX_RESPONSE_BYTES,
-            }).pipe(
-              Effect.mapError(
-                (cause) =>
-                  new BitbucketResponseBodyReadError({
-                    operation: "request",
-                    status: success.status,
-                    cause,
-                  }),
               ),
+            orElse: (failed) => responseError("request", failed),
+          })(response),
+        ),
+        // This also caps redirect hops: no request can finish headers then stall its body forever.
+        Effect.timeout(BITBUCKET_API_REQUEST_TIMEOUT),
+        Effect.catchTags({
-              Effect.map((collected) => ({
-                body: collected.text,
-                truncated: collected.truncated,
-              })),
-            ),
-          orElse: (failed) => responseError("request", failed),
-        })(response),
+          TimeoutError: (cause) =>
+            Effect.fail(new BitbucketRequestError({ operation: "request", cause })),
+        }),
       ),
-      // This also caps redirect hops: no request can finish headers then stall its body forever.
-      Effect.timeout(BITBUCKET_API_REQUEST_TIMEOUT),
-      Effect.catchTags({
-        TimeoutError: (cause) =>
-          Effect.fail(new BitbucketRequestError({ operation: "request", cause })),
-      }),
     );
 
   return BitbucketApi.of({
```

## packages/source-control-bitbucket/src/server/BitbucketPullRequestProvider.test.ts

Status: modified; +12; −0

```diff
@@ -1,7 +1,9 @@
+import { SourceControlProviderKind } from "@t3tools/contracts";
 import { describe, expect, it } from "@effect/vitest";
 import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
 import * as Result from "effect/Result";
+import * as SourceControlRateLimit from "@t3tools/source-control-core/server/SourceControlRateLimit";
 
 import * as BitbucketApi from "./BitbucketApi.ts";
 import * as BitbucketPullRequestApi from "./BitbucketPullRequestApi.ts";
@@ -188,6 +190,16 @@
     expect(bitbucketProviderFailure(responseError(401)).reason).toBe("unauthenticated");
     expect(bitbucketProviderFailure(responseError(403)).reason).toBe("failed");
   });
+
+  it("reports a request held back by the host-wide pause as rate limited until it ends", () => {
+    const paused = new SourceControlRateLimit.SourceControlRateLimitPausedError({
+      provider: SourceControlProviderKind.make("bitbucket"),
+      host: "bitbucket.org",
+      retryAt: 120_000,
+    });
+
+    expect(bitbucketProviderFailure(paused)).toEqual({ reason: "rate-limited", retryAt: 120_000 });
+  });
 });
 
 describe("bitbucketViewerPermissions", () => {
```

## packages/source-control-bitbucket/src/server/BitbucketPullRequestProvider.ts

Status: modified; +3; −0

```diff
@@ -61,6 +61,9 @@
   if (error._tag === "BitbucketResponseError" && error.status === 401) {
     return { reason: "unauthenticated" };
   }
+  if (error._tag === "SourceControlRateLimitPausedError") {
+    return { reason: "rate-limited", retryAt: error.retryAt };
+  }
   if (
     (error._tag === "BitbucketResponseError" || error._tag === "BitbucketResponseBodyReadError") &&
     error.status === 429
```