fix: address background sync edge cases

Author: brendan-kellam

docs/snippets/schemas/v3/index.schema.mdx

@@ -38,7 +38,8 @@
         "reindexRepoPollingIntervalMs": {
           "type": "number",
           "description": "The polling rate (in milliseconds) at which the db should be checked for repos that should be re-indexed. Defaults to 1 second.",
-          "minimum": 1
+          "minimum": 1,
+          "deprecated": true
         },
         "maxConnectionSyncJobConcurrency": {
           "type": "number",
@@ -224,7 +225,8 @@
         "reindexRepoPollingIntervalMs": {
           "type": "number",
           "description": "The polling rate (in milliseconds) at which the db should be checked for repos that should be re-indexed. Defaults to 1 second.",
-          "minimum": 1
+          "minimum": 1,
+          "deprecated": true
         },
         "maxConnectionSyncJobConcurrency": {
           "type": "number",

packages/backend/src/ee/repoPermissionSyncWorkload.test.ts

@@ -77,6 +77,7 @@ const repo = {
     name: "github.com/sourcebot-dev/sourcebot",
     displayName: "sourcebot-dev/sourcebot",
     external_codeHostType: "github",
+    external_codeHostUrl: "https://github.com",
     external_id: "123",
     metadata: {},
     connections: [],
@@ -239,7 +240,7 @@ describe("repoPermissionSyncWorkload", () => {
         expect(repoFindUniqueOrThrow).not.toHaveBeenCalled();
     });
 
-    test("replaces all permissions for a complete GitHub sync", async () => {
+    test("uses the canonical repo issuer for a hostless GitHub Cloud sync", async () => {
         const githubRepo = {
             ...repo,
             external_codeHostType: "github",
@@ -248,7 +249,6 @@ describe("repoPermissionSyncWorkload", () => {
         };
         repoFindUniqueOrThrow.mockResolvedValue(githubRepo);
         mocks.getAuthCredentialsForRepo.mockResolvedValue({
-            hostUrl: "https://github.com",
             token: "token",
         });
         const octokit = {};
@@ -292,6 +292,7 @@ describe("repoPermissionSyncWorkload", () => {
         const bitbucketRepo = {
             ...repo,
             external_codeHostType: "bitbucketCloud",
+            external_codeHostUrl: "https://bitbucket.org",
             external_id: "repo-uuid",
             metadata: {
                 codeHostMetadata: {
@@ -304,7 +305,7 @@ describe("repoPermissionSyncWorkload", () => {
         };
         repoFindUniqueOrThrow.mockResolvedValue(bitbucketRepo);
         mocks.getAuthCredentialsForRepo.mockResolvedValue({
-            hostUrl: "https://bitbucket.org",
+            hostUrl: "https://bitbucket.org/",
             token: "token",
             connectionConfig: {
                 user: "service-account",
@@ -318,6 +319,15 @@ describe("repoPermissionSyncWorkload", () => {
 
         await createWorkload().process(processContext);
 
+        expect(accountFindMany).toHaveBeenCalledWith({
+            where: {
+                providerType: "bitbucket-cloud",
+                providerAccountId: {
+                    in: ["upstream-account"],
+                },
+                issuerUrl: "https://bitbucket.org",
+            },
+        });
         expect(repoUpdate).toHaveBeenCalledWith({
             where: { id: 42 },
             data: {

packages/backend/src/ee/repoPermissionSyncWorkload.ts

@@ -278,7 +278,7 @@ const getGitHubPermissionSyncResult = async ({
             providerAccountId: {
                 in: githubUserIds,
             },
-            issuerUrl: credentials.hostUrl,
+            issuerUrl: repo.external_codeHostUrl,
         },
     });
 
@@ -311,7 +311,7 @@ const getGitLabPermissionSyncResult = async ({
             providerAccountId: {
                 in: gitlabUserIds,
             },
-            issuerUrl: credentials.hostUrl,
+            issuerUrl: repo.external_codeHostUrl,
         },
     });
 
@@ -370,7 +370,7 @@ const getBitbucketCloudPermissionSyncResult = async ({
             providerAccountId: {
                 in: userAccountIds,
             },
-            issuerUrl: credentials.hostUrl,
+            issuerUrl: repo.external_codeHostUrl,
         },
     });
 
@@ -429,7 +429,7 @@ const getBitbucketServerPermissionSyncResult = async ({
         where: {
             providerType: "bitbucket-server",
             providerAccountId: { in: userIds },
-            issuerUrl: credentials.hostUrl,
+            issuerUrl: repo.external_codeHostUrl,
         },
     });
 

packages/backend/src/jobManager.test.ts

@@ -389,4 +389,46 @@ describe("BullMQJobManager lifecycle", () => {
             expect.objectContaining({ attempt: 2 }),
         );
     });
+
+    test("treats BullMQ's stalled-limit failure as terminal", async () => {
+        const onTerminalFailure = vi.fn(async () => undefined);
+        const manager = new BullMQJobManager({} as Redis);
+        manager.register(createWorkload({ onTerminalFailure }));
+        await manager.start();
+
+        const stalledJob = { ...job, attemptsMade: 1 };
+        const error = new Error("job stalled more than allowable limit");
+        mocks.workers[0].handlers.get("failed")?.(stalledJob, error);
+
+        await vi.waitFor(() => {
+            expect(onTerminalFailure).toHaveBeenCalledWith(
+                expect.objectContaining({
+                    jobId: "job-1",
+                    attemptsMade: 1,
+                    maxAttempts: 2,
+                }),
+                error,
+            );
+        });
+    });
+
+    test("keeps ordinary failures retryable before attempts are exhausted", async () => {
+        const onTerminalFailure = vi.fn(async () => undefined);
+        const manager = new BullMQJobManager({} as Redis);
+        manager.register(createWorkload({ onTerminalFailure }));
+        await manager.start();
+
+        const retryableJob = { ...job, attemptsMade: 1 };
+        mocks.workers[0].handlers.get("failed")?.(
+            retryableJob,
+            new Error("temporary failure"),
+        );
+
+        await vi.waitFor(() => {
+            expect(mocks.logger.warn).toHaveBeenCalledWith(
+                expect.stringContaining("will retry"),
+            );
+        });
+        expect(onTerminalFailure).not.toHaveBeenCalled();
+    });
 });

packages/backend/src/jobManager.ts

@@ -20,6 +20,7 @@ import type { JobManager } from "./types.js";
 import { prisma } from "./prisma.js";
 
 const LOG_TAG = "job-manager";
+const STALLED_JOB_TERMINAL_ERROR = "job stalled more than allowable limit";
 const logger = createLogger(LOG_TAG);
 
 export class BullMQJobManager implements JobManager {
@@ -278,7 +279,9 @@ export class BullMQJobManager implements JobManager {
             return;
         }
         const maxAttempts = job.opts.attempts ?? 1;
-        const isTerminal = job.attemptsMade >= maxAttempts;
+        const isTerminal =
+            job.attemptsMade >= maxAttempts ||
+            error.message === STALLED_JOB_TERMINAL_ERROR;
         if (!isTerminal) {
             logger.warn(
                 `Workload "${workload.queueSpec.name}" job ${job.id} failed attempt ${job.attemptsMade}/${maxAttempts}; will retry: ${error.message}`,

packages/schemas/src/v3/index.schema.ts

@@ -37,7 +37,8 @@ const schema = {
         "reindexRepoPollingIntervalMs": {
           "type": "number",
           "description": "The polling rate (in milliseconds) at which the db should be checked for repos that should be re-indexed. Defaults to 1 second.",
-          "minimum": 1
+          "minimum": 1,
+          "deprecated": true
         },
         "maxConnectionSyncJobConcurrency": {
           "type": "number",
@@ -223,7 +224,8 @@ const schema = {
         "reindexRepoPollingIntervalMs": {
           "type": "number",
           "description": "The polling rate (in milliseconds) at which the db should be checked for repos that should be re-indexed. Defaults to 1 second.",
-          "minimum": 1
+          "minimum": 1,
+          "deprecated": true
         },
         "maxConnectionSyncJobConcurrency": {
           "type": "number",

packages/schemas/src/v3/index.type.ts

@@ -106,6 +106,7 @@ export interface Settings {
    */
   resyncConnectionPollingIntervalMs?: number;
   /**
+   * @deprecated
    * The polling rate (in milliseconds) at which the db should be checked for repos that should be re-indexed. Defaults to 1 second.
    */
   reindexRepoPollingIntervalMs?: number;

packages/shared/src/jobLogger.test.ts

@@ -102,6 +102,30 @@ describe("createBullMQJobLogSink", () => {
 
         expect(JSON.parse(log.mock.calls[0][0])).toMatchObject({ attempt: 2 });
     });
+
+    test("closes before draining and ignores later writes", async () => {
+        let resolveWrite: (value: number) => void = () => undefined;
+        const pendingWrite = new Promise<number>((resolve) => {
+            resolveWrite = resolve;
+        });
+        const log = vi.fn().mockReturnValue(pendingWrite);
+        const sink = createBullMQJobLogSink({
+            id: "job-1",
+            name: "connection",
+            queueName: "connection",
+            attemptsMade: 0,
+            log,
+        });
+
+        sink.info("Before flush");
+        const flush = sink.flush();
+        sink.info("During flush");
+        resolveWrite(1);
+        await flush;
+        sink.info("After flush");
+
+        expect(log).toHaveBeenCalledOnce();
+    });
 });
 
 describe("readBullMQJobLogs", () => {

packages/shared/src/jobLogger.ts

@@ -177,12 +177,17 @@ export const createBullMQJobLogSink = (
         options.label ?? `${job.queueName}:job:${job.id ?? "unknown"}`;
     const attempt = options.attempt ?? job.attemptsMade + 1;
     const pendingWrites = new Set<Promise<void>>();
+    let closed = false;
 
     const write = (
         level: JobLogLevel,
         message: string,
         rawFields?: unknown,
     ): void => {
+        if (closed) {
+            return;
+        }
+
         const fields = sanitizeFields(rawFields);
 
         const entry: JobLogEntry = {
@@ -221,7 +226,10 @@ export const createBullMQJobLogSink = (
         error: (message: string, fields?: unknown) =>
             write("error", message, fields),
         flush: async () => {
-            await Promise.all([...pendingWrites]);
+            closed = true;
+            while (pendingWrites.size > 0) {
+                await Promise.all([...pendingWrites]);
+            }
         },
     } satisfies JobLogSink & { flush(): Promise<void> };
 };

schemas/v3/index.json

@@ -36,7 +36,8 @@
                 "reindexRepoPollingIntervalMs": {
                     "type": "number",
                     "description": "The polling rate (in milliseconds) at which the db should be checked for repos that should be re-indexed. Defaults to 1 second.",
-                    "minimum": 1
+                    "minimum": 1,
+                    "deprecated": true
                 },
                 "maxConnectionSyncJobConcurrency": {
                     "type": "number",
@@ -176,4 +177,4 @@
         }
     },
     "additionalProperties": false
-}
+}