Merge branch 'main' into ghes-review-agent

Author: brianphillips

CHANGELOG.md

@@ -7,10 +7,26 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+- Fixed spurious infinite loads with explore panel, file tree, and file search command. [#617](https://github.com/sourcebot-dev/sourcebot/pull/617)
+- Wipe search context on init if entitlement no longer exists [#618](https://github.com/sourcebot-dev/sourcebot/pull/618)
+- Fixed review agent so that it works with GHES instances [#611](https://github.com/sourcebot-dev/sourcebot/pull/611)
+
+## [4.9.2] - 2025-11-13
+
+### Changed
+- Bumped the default requested search result count from 5k to 10k after optimization pass. [#615](https://github.com/sourcebot-dev/sourcebot/pull/615)
+
 ### Fixed
 - Fixed incorrect shutdown of PostHog SDK in the worker. [#609](https://github.com/sourcebot-dev/sourcebot/pull/609)
 - Fixed race condition in job schedulers. [#607](https://github.com/sourcebot-dev/sourcebot/pull/607)
-- Fixed review agent so that it works with GHES instances [#611](https://github.com/sourcebot-dev/sourcebot/pull/611)
+- Fixed connection sync jobs getting stuck in pending or in progress after restarting the worker. [#612](https://github.com/sourcebot-dev/sourcebot/pull/612)
+- Fixed issue where connections would always sync on startup, regardless if they changed or not. [#613](https://github.com/sourcebot-dev/sourcebot/pull/613)
+- Fixed performance bottleneck in search api. Result is a order of magnitutde improvement to average search time according to benchmarks. [#615](https://github.com/sourcebot-dev/sourcebot/pull/615) 
+
+### Added
+- Added force resync buttons for connections and repositories. [#610](https://github.com/sourcebot-dev/sourcebot/pull/610)
+- Added environment variable to configure default search result count. [#616](https://github.com/sourcebot-dev/sourcebot/pull/616)
 
 ## [4.9.1] - 2025-11-07
 

docs/docs/configuration/environment-variables.mdx

@@ -34,6 +34,7 @@ The following environment variables allow you to configure your Sourcebot deploy
 | `SOURCEBOT_STRUCTURED_LOGGING_ENABLED` | `false` | <p>Enables/disable structured JSON logging. See [this doc](/docs/configuration/structured-logging) for more info.</p> |
 | `SOURCEBOT_STRUCTURED_LOGGING_FILE` | - | <p>Optional file to log to if structured logging is enabled</p> | 
 | `SOURCEBOT_TELEMETRY_DISABLED` | `false` | <p>Enables/disables telemetry collection in Sourcebot. See [this doc](/docs/overview.mdx#telemetry) for more info.</p> |
+| `DEFAULT_MAX_MATCH_COUNT` | `10000` | <p>The default maximum number of search results to return when using search in the web app.</p> |
 
 ### Enterprise Environment Variables
 | Variable | Default | Description |

packages/backend/package.json

@@ -40,6 +40,8 @@
         "cross-fetch": "^4.0.0",
         "dotenv": "^16.4.5",
         "express": "^4.21.2",
+        "express-async-errors": "^3.1.1",
+        "fast-deep-equal": "^3.1.3",
         "git-url-parse": "^16.1.0",
         "gitea-js": "^1.22.0",
         "glob": "^11.0.0",

packages/backend/src/api.ts

@@ -0,0 +1,103 @@
+import { PrismaClient, RepoIndexingJobType } from '@sourcebot/db';
+import { createLogger } from '@sourcebot/shared';
+import express, { Request, Response } from 'express';
+import 'express-async-errors';
+import * as http from "http";
+import z from 'zod';
+import { ConnectionManager } from './connectionManager.js';
+import { PromClient } from './promClient.js';
+import { RepoIndexManager } from './repoIndexManager.js';
+
+const logger = createLogger('api');
+const PORT = 3060;
+
+export class Api {
+    private server: http.Server;
+
+    constructor(
+        promClient: PromClient,
+        private prisma: PrismaClient,
+        private connectionManager: ConnectionManager,
+        private repoIndexManager: RepoIndexManager,
+    ) {
+        const app = express();
+        app.use(express.json());
+        app.use(express.urlencoded({ extended: true }));
+
+        // Prometheus metrics endpoint
+        app.use('/metrics', async (_req: Request, res: Response) => {
+            res.set('Content-Type', promClient.registry.contentType);
+            const metrics = await promClient.registry.metrics();
+            res.end(metrics);
+        });
+
+        app.post('/api/sync-connection', this.syncConnection.bind(this));
+        app.post('/api/index-repo', this.indexRepo.bind(this));
+
+        this.server = app.listen(PORT, () => {
+            logger.info(`API server is running on port ${PORT}`);
+        });
+    }
+
+    private async syncConnection(req: Request, res: Response) {
+        const schema = z.object({
+            connectionId: z.number(),
+        }).strict();
+
+        const parsed = schema.safeParse(req.body);
+        if (!parsed.success) {
+            res.status(400).json({ error: parsed.error.message });
+            return;
+        }
+
+        const { connectionId } = parsed.data;
+        const connection = await this.prisma.connection.findUnique({
+            where: {
+                id: connectionId,
+            }
+        });
+
+        if (!connection) {
+            res.status(404).json({ error: 'Connection not found' });
+            return;
+        }
+
+        const [jobId] = await this.connectionManager.createJobs([connection]);
+
+        res.status(200).json({ jobId });
+    }
+
+    private async indexRepo(req: Request, res: Response) {
+        const schema = z.object({
+            repoId: z.number(),
+        }).strict();
+
+        const parsed = schema.safeParse(req.body);
+        if (!parsed.success) {
+            res.status(400).json({ error: parsed.error.message });
+            return;
+        }
+
+        const { repoId } = parsed.data;
+        const repo = await this.prisma.repo.findUnique({
+            where: { id: repoId },
+        });
+
+        if (!repo) {
+            res.status(404).json({ error: 'Repo not found' });
+            return;
+        }
+
+        const [jobId] = await this.repoIndexManager.createJobs([repo], RepoIndexingJobType.INDEX);
+        res.status(200).json({ jobId });
+    }
+
+    public async dispose() {
+        return new Promise<void>((resolve, reject) => {
+            this.server.close((err) => {
+                if (err) reject(err);
+                else resolve(undefined);
+            });
+        });
+    }
+}

packages/backend/src/configManager.ts

@@ -6,6 +6,7 @@ import chokidar, { FSWatcher } from 'chokidar';
 import { ConnectionManager } from "./connectionManager.js";
 import { SINGLE_TENANT_ORG_ID } from "./constants.js";
 import { syncSearchContexts } from "./ee/syncSearchContexts.js";
+import isEqual from 'fast-deep-equal';
 
 const logger = createLogger('config-manager');
 
@@ -64,8 +65,8 @@ export class ConfigManager {
 
                 const existingConnectionConfig = existingConnection ? existingConnection.config as unknown as ConnectionConfig : undefined;
                 const connectionNeedsSyncing =
-                    !existingConnection ||
-                    (JSON.stringify(existingConnectionConfig) !== JSON.stringify(newConnectionConfig));
+                    !existingConnectionConfig ||
+                    !isEqual(existingConnectionConfig, newConnectionConfig);
 
                 // Either update the existing connection or create a new one.
                 const connection = existingConnection ?
@@ -93,8 +94,8 @@ export class ConfigManager {
                     });
 
                 if (connectionNeedsSyncing) {
-                    const [jobId] = await this.connectionManager.createJobs([connection]);
-                    logger.info(`Change detected for connection '${key}' (id: ${connection.id}). Created sync job ${jobId}.`);
+                    logger.info(`Change detected for connection '${key}' (id: ${connection.id}). Creating sync job.`);
+                    await this.connectionManager.createJobs([connection]);
                 }
             }
         }

packages/backend/src/connectionManager.ts

@@ -11,10 +11,12 @@ import { groupmqLifecycleExceptionWrapper, setIntervalAsync } from "./utils.js";
 import { syncSearchContexts } from "./ee/syncSearchContexts.js";
 import { captureEvent } from "./posthog.js";
 import { PromClient } from "./promClient.js";
+import { GROUPMQ_WORKER_STOP_GRACEFUL_TIMEOUT_MS } from "./constants.js";
 
 const LOG_TAG = 'connection-manager';
 const logger = createLogger(LOG_TAG);
 const createJobLogger = (jobId: string) => createLogger(`${LOG_TAG}:job:${jobId}`);
+const QUEUE_NAME = 'connection-sync-queue';
 
 type JobPayload = {
     jobId: string,
@@ -30,19 +32,19 @@ type JobResult = {
 const JOB_TIMEOUT_MS = 1000 * 60 * 60 * 2; // 2 hour timeout
 
 export class ConnectionManager {
-    private worker: Worker;
+    private worker: Worker<JobPayload>;
     private queue: Queue<JobPayload>;
     private interval?: NodeJS.Timeout;
 
     constructor(
         private db: PrismaClient,
         private settings: Settings,
-        redis: Redis,
+        private redis: Redis,
         private promClient: PromClient,
     ) {
         this.queue = new Queue<JobPayload>({
             redis,
-            namespace: 'connection-sync-queue',
+            namespace: QUEUE_NAME,
             jobTimeoutMs: JOB_TIMEOUT_MS,
             maxAttempts: 3,
             logger: env.DEBUG_ENABLE_GROUPMQ_LOGGING === 'true',
@@ -62,6 +64,10 @@ export class ConnectionManager {
         this.worker.on('failed', this.onJobFailed.bind(this));
         this.worker.on('stalled', this.onJobStalled.bind(this));
         this.worker.on('error', this.onWorkerError.bind(this));
+        // graceful-timeout is triggered when a job is still processing after
+        // worker.close() is called and the timeout period has elapsed. In this case,
+        // we fail the job with no retry.
+        this.worker.on('graceful-timeout', this.onJobGracefulTimeout.bind(this));
     }
 
     public startScheduler() {
@@ -128,6 +134,7 @@ export class ConnectionManager {
         });
 
         for (const job of jobs) {
+            logger.info(`Scheduling job ${job.id} for connection ${job.connection.name} (id: ${job.connectionId})`);
             await this.queue.add({
                 groupId: `connection:${job.connectionId}`,
                 data: {
@@ -150,6 +157,22 @@ export class ConnectionManager {
         const logger = createJobLogger(jobId);
         logger.info(`Running connection sync job ${jobId} for connection ${connectionName} (id: ${job.data.connectionId}) (attempt ${job.attempts + 1} / ${job.maxAttempts})`);
 
+        const currentStatus = await this.db.connectionSyncJob.findUniqueOrThrow({
+            where: {
+                id: jobId,
+            },
+            select: {
+                status: true,
+            }
+        });
+
+        // Fail safe: if the job is not PENDING (first run) or IN_PROGRESS (retry), it indicates the job
+        // is in an invalid state and should be skipped.
+        if (currentStatus.status !== ConnectionSyncJobStatus.PENDING && currentStatus.status !== ConnectionSyncJobStatus.IN_PROGRESS) {
+            throw new Error(`Job ${jobId} is not in a valid state. Expected: ${ConnectionSyncJobStatus.PENDING} or ${ConnectionSyncJobStatus.IN_PROGRESS}. Actual: ${currentStatus.status}. Skipping.`);
+        }
+
+
         this.promClient.pendingConnectionSyncJobs.dec({ connection: connectionName });
         this.promClient.activeConnectionSyncJobs.inc({ connection: connectionName });
 
@@ -178,7 +201,7 @@ export class ConnectionManager {
         const result = await (async () => {
             switch (config.type) {
                 case 'github': {
-                    return await compileGithubConfig(config, job.data.connectionId, abortController);
+                    return await compileGithubConfig(config, job.data.connectionId, abortController.signal);
                 }
                 case 'gitlab': {
                     return await compileGitlabConfig(config, job.data.connectionId);
@@ -200,7 +223,7 @@ export class ConnectionManager {
                 }
             }
         })();
-       
+
         let { repoData, warnings } = result;
 
         await this.db.connectionSyncJob.update({
@@ -383,6 +406,33 @@ export class ConnectionManager {
             });
         });
 
+    private onJobGracefulTimeout = async (job: Job<JobPayload>) =>
+        groupmqLifecycleExceptionWrapper('onJobGracefulTimeout', logger, async () => {
+            const logger = createJobLogger(job.id);
+
+            const { connection } = await this.db.connectionSyncJob.update({
+                where: { id: job.id },
+                data: {
+                    status: ConnectionSyncJobStatus.FAILED,
+                    completedAt: new Date(),
+                    errorMessage: 'Job timed out',
+                },
+                select: {
+                    connection: true,
+                }
+            });
+
+            this.promClient.activeConnectionSyncJobs.dec({ connection: connection.name });
+            this.promClient.connectionSyncJobFailTotal.inc({ connection: connection.name });
+
+            logger.error(`Job ${job.id} timed out for connection ${connection.name} (id: ${connection.id})`);
+
+            captureEvent('backend_connection_sync_job_failed', {
+                connectionId: connection.id,
+                error: 'Job timed out',
+            });
+        });
+
     private async onWorkerError(error: Error) {
         Sentry.captureException(error);
         logger.error(`Connection syncer worker error.`, error);
@@ -392,8 +442,28 @@ export class ConnectionManager {
         if (this.interval) {
             clearInterval(this.interval);
         }
-        await this.worker.close();
-        await this.queue.close();
+
+        const inProgressJobs = this.worker.getCurrentJobs();
+        await this.worker.close(GROUPMQ_WORKER_STOP_GRACEFUL_TIMEOUT_MS);
+
+        // Manually release group locks for in progress jobs to prevent deadlocks.
+        // @see: https://github.com/Openpanel-dev/groupmq/issues/8
+        for (const { job } of inProgressJobs) {
+            const lockKey = `groupmq:${QUEUE_NAME}:lock:${job.groupId}`;
+            logger.debug(`Releasing group lock ${lockKey} for in progress job ${job.id}`);
+            try {
+                await this.redis.del(lockKey);
+            } catch (error) {
+                Sentry.captureException(error);
+                logger.error(`Failed to release group lock ${lockKey} for in progress job ${job.id}. Error: `, error);
+            }
+        }
+
+        // @note: As of groupmq v1.0.0, queue.close() will just close the underlying
+        // redis connection. Since we share the same redis client between, skip this
+        // step and close the redis client directly in index.ts.
+        // @see: https://github.com/Openpanel-dev/groupmq/blob/main/src/queue.ts#L1900
+        // await this.queue.close();
     }
 }
 

packages/backend/src/constants.ts

@@ -10,4 +10,24 @@ export const PERMISSION_SYNC_SUPPORTED_CODE_HOST_TYPES: CodeHostType[] = [
 ];
 
 export const REPOS_CACHE_DIR = path.join(env.DATA_CACHE_DIR, 'repos');
-export const INDEX_CACHE_DIR = path.join(env.DATA_CACHE_DIR, 'index');
+export const INDEX_CACHE_DIR = path.join(env.DATA_CACHE_DIR, 'index');
+
+// Maximum time to wait for current job to finish
+export const GROUPMQ_WORKER_STOP_GRACEFUL_TIMEOUT_MS = 5 * 1000; // 5 seconds
+
+// List of shutdown signals
+export const SHUTDOWN_SIGNALS: string[] = [
+    'SIGHUP',
+    'SIGINT',
+    'SIGQUIT',
+    'SIGILL',
+    'SIGTRAP',
+    'SIGABRT',
+    'SIGBUS',
+    'SIGFPE',
+    'SIGSEGV',
+    'SIGUSR2',
+    'SIGTERM',
+    // @note: SIGKILL and SIGSTOP cannot have listeners installed.
+    // @see: https://nodejs.org/api/process.html#signal-events
+];