diff --git a/CHANGELOG.md b/CHANGELOG.md index 46c363645..92ccb9b50 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### 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) + ## [5.1.10] - 2026-08-27 ### Fixed diff --git a/packages/backend/src/index.ts b/packages/backend/src/index.ts index fc2ec78fc..91e32240e 100644 --- a/packages/backend/src/index.ts +++ b/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, diff --git a/packages/backend/src/repoCleanupWorkload.test.ts b/packages/backend/src/repoCleanupWorkload.test.ts index 4a1e770e5..e01e53084 100644 --- a/packages/backend/src/repoCleanupWorkload.test.ts +++ b/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,216 @@ 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("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 `.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), + ); + }); +}); diff --git a/packages/backend/src/repoCleanupWorkload.ts b/packages/backend/src/repoCleanupWorkload.ts index d0fedbe81..6b6ef66d0 100644 --- a/packages/backend/src/repoCleanupWorkload.ts +++ b/packages/backend/src/repoCleanupWorkload.ts @@ -3,13 +3,14 @@ import { createLogger, getRepoIdFromPath, getRepoPath, + JOB_PRIORITIES, REPO_CLEANUP_QUEUE, } from "@sourcebot/shared"; import { existsSync } from "fs"; import { readdir, rm } from "fs/promises"; import { INDEX_CACHE_DIR, REPOS_CACHE_DIR } from "./constants.js"; import { REPOSITORY_EXECUTION_LOCK } from "./repoLock.js"; -import type { Settings, Workload } from "./types.js"; +import type { JobManager, Settings, Workload } from "./types.js"; import { getRepoIdFromShardFileName } from "./utils.js"; const logger = createLogger("repo-cleanup-workload"); @@ -231,3 +232,79 @@ export const cleanupOrphanedRepoResources = async (db: PrismaClient) => { } } }; + +// Handles the inverse of cleanupOrphanedRepoResources: repos the DB believes are +// indexed but whose shard files are missing from disk (e.g., INDEX_CACHE_DIR was +// wiped independently of the DB, as happens when it's placed on ephemeral storage). +// Without this, such repos would silently return empty search results until their +// next scheduled reindex, which can be a long time away. +export const reindexReposWithMissingShards = async ( + db: PrismaClient, + jobManager: JobManager, +) => { + // A missing directory means zero shards exist, not that there's nothing to + // recover: it's the same "everything is gone" scenario this function exists + // to handle, so it must still fall through to the DB lookup below. + let entries: string[]; + if (existsSync(INDEX_CACHE_DIR)) { + entries = await readdir(INDEX_CACHE_DIR); + } else { + entries = []; + } + + const repoIdsWithShards = new Set(); + for (const entry of entries) { + // Only a real, searchable shard file counts. This excludes in-progress + // or failed .tmp artifacts, the .meta sidecar zoekt writes alongside + // each shard, and any other numeric-prefixed file that isn't actually + // an index (e.g. a stray backup file). + if (!entry.endsWith(".zoekt")) { + continue; + } + const repoId = getRepoIdFromShardFileName(entry); + if (repoId !== undefined) { + repoIdsWithShards.add(repoId); + } + } + + // Considers the same set of repos reconcileJobSchedulers keeps on a recurring + // reindex schedule: attached to a connection, or explicitly pinned via + // isAutoCleanupDisabled. Anything outside that set is owned by the cleanup + // workload above, not re-indexed. + const indexedRepos = await db.repo.findMany({ + where: { + indexedAt: { not: null }, + OR: [ + { connections: { some: {} } }, + { isAutoCleanupDisabled: true }, + ], + }, + select: { id: true, name: true }, + }); + + const reposMissingShards = indexedRepos.filter( + (repo) => !repoIdsWithShards.has(repo.id), + ); + + // Triggered sequentially so that one repo failing to enqueue (e.g. a + // transient Redis error) doesn't stop the rest from being recovered, and + // can't take down startup: this runs before the worker installs its + // uncaught-exception handlers. + for (const repo of reposMissingShards) { + logger.warn( + `Repo ${repo.name} (id: ${repo.id}) is marked as indexed but has no shard files on disk. Re-queuing for indexing.`, + ); + try { + await jobManager.trigger( + "repo-index", + { repoId: repo.id }, + { priority: JOB_PRIORITIES.SCHEDULED }, + ); + } catch (error) { + logger.error( + `Failed to re-queue repo ${repo.name} (id: ${repo.id}) for indexing:`, + error, + ); + } + } +};