fix(worker): reindex repos with missing zoekt shards

CHANGELOG.md

@@ -14,6 +14,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Updated the bundled Zoekt version. [#1628](https://github.com/sourcebot-dev/sourcebot/pull/1628)
 
 ### Fixed
+- Reindexed repositories on startup when their persisted indexed state no longer had corresponding Zoekt shard files on disk. [#1621](https://github.com/sourcebot-dev/sourcebot/pull/1621)
 - Upgraded `browserslist` to `^4.28.8`. [#1624](https://github.com/sourcebot-dev/sourcebot/pull/1624)
 - Upgraded `postcss-selector-parser` to `^6.1.4`. [#1625](https://github.com/sourcebot-dev/sourcebot/pull/1625)
 - Upgraded `fast-uri` to `3.1.7`. [#1626](https://github.com/sourcebot-dev/sourcebot/pull/1626)

packages/backend/src/index.ts

@@ -13,7 +13,7 @@ import { prisma } from "./prisma.js";
 import { PromClient } from './promClient.js';
 import { redis } from "./redis.js";
 import { createConnectionSyncWorkload } from "./connectionSyncWorkload.js";
-import { cleanupOrphanedRepoResources, createRepoCleanupWorkload } from "./repoCleanupWorkload.js";
+import { cleanupOrphanedRepoResources, createRepoCleanupWorkload, reindexReposWithMissingShards } from "./repoCleanupWorkload.js";
 import { createRepoIndexWorkload } from "./repoIndexWorkload.js";
 import { Api } from "./api.js";
 import { createAccountPermissionSyncWorkload } from "./ee/accountPermissionSyncWorkload.js";
@@ -97,6 +97,14 @@ await cleanupOrphanedRepoResources(prisma);
 const configManager = new ConfigManager(jobManager, env.CONFIG_PATH);
 await configManager.syncConfig();
 
+// Runs after config sync so a repo whose connection this sync just removed
+// (handled synchronously in syncConfig) isn't wrongly re-queued right before
+// it's orphaned. Connections added or changed by this sync are applied
+// asynchronously by their own connection-sync job, which can't run until
+// jobManager.start() below, so that side of eligibility is unaffected by
+// this ordering either way.
+await reindexReposWithMissingShards(prisma, jobManager);
+
 await reconcileJobSchedulers({
     db: prisma,
     jobManager,

packages/backend/src/repoCleanupWorkload.test.ts

@@ -1,6 +1,8 @@
 import type { PrismaClient } from "@sourcebot/db";
+import { JOB_PRIORITIES } from "@sourcebot/shared";
 import { beforeEach, describe, expect, test, vi } from "vitest";
-import { createRepoCleanupWorkload } from "./repoCleanupWorkload.js";
+import { createRepoCleanupWorkload, reindexReposWithMissingShards } from "./repoCleanupWorkload.js";
+import type { JobManager } from "./types.js";
 
 const fsMocks = vi.hoisted(() => ({
     existsSync: vi.fn(),
@@ -30,12 +32,14 @@ vi.mock("fs/promises", () => ({
 }));
 
 const repoFindUnique = vi.fn();
+const repoFindMany = vi.fn();
 const repoDeleteMany = vi.fn();
 const repoUpdate = vi.fn();
 
 const db = {
     repo: {
         findUnique: repoFindUnique,
+        findMany: repoFindMany,
         deleteMany: repoDeleteMany,
         update: repoUpdate,
     },
@@ -77,6 +81,7 @@ describe("repoCleanupWorkload", () => {
         fsMocks.readdir.mockResolvedValue([]);
         fsMocks.rm.mockResolvedValue(undefined);
         repoFindUnique.mockResolvedValue(eligibleRepo);
+        repoFindMany.mockResolvedValue([]);
         repoDeleteMany.mockResolvedValue({ count: 1 });
         repoUpdate.mockResolvedValue(undefined);
     });
@@ -183,3 +188,257 @@ describe("repoCleanupWorkload", () => {
         );
     });
 });
+
+describe("reindexReposWithMissingShards", () => {
+    const trigger = vi.fn();
+    const jobManager = { trigger } as unknown as JobManager;
+
+    beforeEach(() => {
+        vi.clearAllMocks();
+        fsMocks.existsSync.mockReturnValue(true);
+        fsMocks.readdir.mockResolvedValue([]);
+        repoFindMany.mockResolvedValue([]);
+        trigger.mockResolvedValue("job-id");
+    });
+
+    test("no-op when there are no indexed repos, even if stray shard-like files exist on disk", async () => {
+        fsMocks.readdir.mockResolvedValue(["1_42_v16.00000.zoekt"]);
+        repoFindMany.mockResolvedValue([]);
+
+        await reindexReposWithMissingShards(db, jobManager);
+
+        expect(trigger).not.toHaveBeenCalled();
+    });
+
+    test("logs and returns instead of throwing when the DB lookup fails", async () => {
+        // Regression: this detection phase runs before the worker installs its
+        // uncaught-exception handlers (see repoCleanupWorkload.ts), so a transient
+        // failure here must degrade to "recovery skipped this run", not an unhandled
+        // rejection that crashes startup.
+        repoFindMany.mockRejectedValue(new Error("connection reset"));
+
+        await expect(
+            reindexReposWithMissingShards(db, jobManager),
+        ).resolves.not.toThrow();
+
+        expect(trigger).not.toHaveBeenCalled();
+        expect(lifecycleLogger.error).toHaveBeenCalledWith(
+            expect.stringContaining("Failed to detect repos with missing shard files"),
+            expect.any(Error),
+        );
+    });
+
+    test("logs and returns instead of throwing when reading the index directory fails", async () => {
+        fsMocks.readdir.mockRejectedValue(new Error("EACCES: permission denied"));
+
+        await expect(
+            reindexReposWithMissingShards(db, jobManager),
+        ).resolves.not.toThrow();
+
+        expect(trigger).not.toHaveBeenCalled();
+        expect(lifecycleLogger.error).toHaveBeenCalledWith(
+            expect.stringContaining("Failed to detect repos with missing shard files"),
+            expect.any(Error),
+        );
+    });
+
+    test("still recovers eligible repos when the index directory doesn't exist", async () => {
+        fsMocks.existsSync.mockReturnValue(false);
+        repoFindMany.mockResolvedValue([
+            { id: 42, name: "github.com/acme/repo" },
+        ]);
+
+        await reindexReposWithMissingShards(db, jobManager);
+
+        expect(fsMocks.readdir).not.toHaveBeenCalled();
+        expect(repoFindMany).toHaveBeenCalled();
+        expect(trigger).toHaveBeenCalledWith(
+            "repo-index",
+            { repoId: 42 },
+            { priority: JOB_PRIORITIES.SCHEDULED },
+        );
+    });
+
+    test("re-queues an indexed repo with no shard on disk", async () => {
+        fsMocks.readdir.mockResolvedValue([]);
+        repoFindMany.mockResolvedValue([
+            { id: 42, name: "github.com/acme/repo" },
+        ]);
+
+        await reindexReposWithMissingShards(db, jobManager);
+
+        // Pins down the exact where-clause: repos still eligible for reindex
+        // scheduling (has a connection, or explicitly pinned via
+        // isAutoCleanupDisabled) that the DB believes are indexed. This mirrors
+        // the set reconcileJobSchedulers.ts keeps on a recurring reindex
+        // schedule, since orphaned repos with no such pin are the cleanup
+        // workload's responsibility, not this one's.
+        expect(repoFindMany).toHaveBeenCalledWith({
+            where: {
+                indexedAt: { not: null },
+                OR: [
+                    { connections: { some: {} } },
+                    { isAutoCleanupDisabled: true },
+                ],
+            },
+            select: { id: true, name: true },
+        });
+        expect(trigger).toHaveBeenCalledWith(
+            "repo-index",
+            { repoId: 42 },
+            { priority: JOB_PRIORITIES.SCHEDULED },
+        );
+    });
+
+    test("does not re-queue a repo that already has a shard on disk", async () => {
+        fsMocks.readdir.mockResolvedValue(["1_42_v16.00000.zoekt"]);
+        repoFindMany.mockResolvedValue([
+            { id: 42, name: "github.com/acme/repo" },
+        ]);
+
+        await reindexReposWithMissingShards(db, jobManager);
+
+        expect(trigger).not.toHaveBeenCalled();
+    });
+
+    test("does not re-queue a repo whose shard and .meta sidecar are both present", async () => {
+        // The normal healthy state: zoekt always writes the .meta sidecar
+        // alongside the real shard, so both show up in the same readdir().
+        fsMocks.readdir.mockResolvedValue([
+            "1_42_v16.00000.zoekt",
+            "1_42_v16.00000.zoekt.meta",
+        ]);
+        repoFindMany.mockResolvedValue([
+            { id: 42, name: "github.com/acme/repo" },
+        ]);
+
+        await reindexReposWithMissingShards(db, jobManager);
+
+        expect(trigger).not.toHaveBeenCalled();
+    });
+
+    test("treats a lingering .tmp shard as missing", async () => {
+        fsMocks.readdir.mockResolvedValue(["1_42_v16.00000.zoekt.tmp"]);
+        repoFindMany.mockResolvedValue([
+            { id: 42, name: "github.com/acme/repo" },
+        ]);
+
+        await reindexReposWithMissingShards(db, jobManager);
+
+        expect(trigger).toHaveBeenCalledWith(
+            "repo-index",
+            { repoId: 42 },
+            { priority: JOB_PRIORITIES.SCHEDULED },
+        );
+    });
+
+    test("treats the .meta sidecar file alone as missing", async () => {
+        // zoekt writes a `<shard>.meta` file alongside every real shard. If
+        // only the sidecar survives a partial wipe, the repo has no searchable
+        // index and must still be re-queued.
+        fsMocks.readdir.mockResolvedValue(["1_42_v16.00000.zoekt.meta"]);
+        repoFindMany.mockResolvedValue([
+            { id: 42, name: "github.com/acme/repo" },
+        ]);
+
+        await reindexReposWithMissingShards(db, jobManager);
+
+        expect(trigger).toHaveBeenCalledWith(
+            "repo-index",
+            { repoId: 42 },
+            { priority: JOB_PRIORITIES.SCHEDULED },
+        );
+    });
+
+    test("does not treat a numeric-prefixed non-shard file as a valid shard", async () => {
+        fsMocks.readdir.mockResolvedValue(["1_42_backup"]);
+        repoFindMany.mockResolvedValue([
+            { id: 42, name: "github.com/acme/repo" },
+        ]);
+
+        await reindexReposWithMissingShards(db, jobManager);
+
+        expect(trigger).toHaveBeenCalledWith(
+            "repo-index",
+            { repoId: 42 },
+            { priority: JOB_PRIORITIES.SCHEDULED },
+        );
+    });
+
+    test("ignores unrelated files in the index directory", async () => {
+        fsMocks.readdir.mockResolvedValue([".DS_Store", "README.md"]);
+        repoFindMany.mockResolvedValue([
+            { id: 42, name: "github.com/acme/repo" },
+        ]);
+
+        await reindexReposWithMissingShards(db, jobManager);
+
+        expect(trigger).toHaveBeenCalledWith(
+            "repo-index",
+            { repoId: 42 },
+            { priority: JOB_PRIORITIES.SCHEDULED },
+        );
+    });
+
+    test("recognizes a repo whose content is split across multiple shard files", async () => {
+        fsMocks.readdir.mockResolvedValue([
+            "1_42_v16.00000.zoekt",
+            "1_42_v16.00001.zoekt",
+        ]);
+        repoFindMany.mockResolvedValue([
+            { id: 42, name: "github.com/acme/repo" },
+        ]);
+
+        await reindexReposWithMissingShards(db, jobManager);
+
+        expect(trigger).not.toHaveBeenCalled();
+    });
+
+    test("only re-queues the repo actually missing a shard among many", async () => {
+        fsMocks.readdir.mockResolvedValue(["1_42_v16.00000.zoekt"]);
+        repoFindMany.mockResolvedValue([
+            { id: 42, name: "github.com/acme/healthy-repo" },
+            { id: 43, name: "github.com/acme/broken-repo" },
+        ]);
+
+        await reindexReposWithMissingShards(db, jobManager);
+
+        expect(trigger).toHaveBeenCalledTimes(1);
+        expect(trigger).toHaveBeenCalledWith(
+            "repo-index",
+            { repoId: 43 },
+            { priority: JOB_PRIORITIES.SCHEDULED },
+        );
+    });
+
+    test("re-queues remaining repos even if one fails to enqueue", async () => {
+        fsMocks.readdir.mockResolvedValue([]);
+        repoFindMany.mockResolvedValue([
+            { id: 42, name: "github.com/acme/flaky-repo" },
+            { id: 43, name: "github.com/acme/broken-repo" },
+        ]);
+        trigger.mockImplementation(async (_name, data: { repoId: number }) => {
+            if (data.repoId === 42) {
+                throw new Error("redis connection reset");
+            }
+            return "job-id";
+        });
+
+        await expect(
+            reindexReposWithMissingShards(db, jobManager),
+        ).resolves.not.toThrow();
+
+        expect(trigger).toHaveBeenCalledTimes(2);
+        expect(trigger).toHaveBeenCalledWith(
+            "repo-index",
+            { repoId: 43 },
+            { priority: JOB_PRIORITIES.SCHEDULED },
+        );
+        expect(lifecycleLogger.error).toHaveBeenCalledWith(
+            expect.stringContaining(
+                "Failed to re-queue repo github.com/acme/flaky-repo (id: 42)",
+            ),
+            expect.any(Error),
+        );
+    });
+});