fix(worker): Run setInterval as blocking

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### 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)
 
 ## [4.9.1] - 2025-11-07
 

packages/backend/src/connectionManager.ts

@@ -7,7 +7,7 @@ import{Job, Queue, ReservedJob, Worker } from "groupmq"
 import{Redis } from 'ioredis'
 import{compileAzureDevOpsConfig, compileBitbucketConfig, compileGenericGitHostConfig, compileGerritConfig, compileGiteaConfig, compileGithubConfig, compileGitlabConfig } from "./repoCompileUtils.js"
 import{Settings } from "./types.js"
-import{groupmqLifecycleExceptionWrapper } from "./utils.js"
+import{groupmqLifecycleExceptionWrapper, setIntervalAsync } from "./utils.js"
 import{syncSearchContexts } from "./ee/syncSearchContexts.js"
 import{captureEvent } from "./posthog.js"
 import{PromClient } from "./promClient.js"
@@ -66,7 +66,7 @@ export class ConnectionManager{
 
  public startScheduler(){
  logger.debug('Starting scheduler');
- this.interval = setInterval(async () =>{
+ this.interval = setIntervalAsync(async () =>{
  const thresholdDate = new Date(Date.now() - this.settings.resyncConnectionIntervalMs);
  const timeoutDate = new Date(Date.now() - JOB_TIMEOUT_MS);
 

packages/backend/src/ee/accountPermissionSyncer.ts

@@ -7,6 +7,7 @@ import{PERMISSION_SYNC_SUPPORTED_CODE_HOST_TYPES } from "../constants.js"
 import{createOctokitFromToken, getReposForAuthenticatedUser } from "../github.js"
 import{createGitLabFromOAuthToken, getProjectsForAuthenticatedUser } from "../gitlab.js"
 import{Settings } from "../types.js"
+import{setIntervalAsync } from "../utils.js"
 
 const LOG_TAG = 'user-permission-syncer'
 const logger = createLogger(LOG_TAG);
@@ -46,7 +47,7 @@ export class AccountPermissionSyncer{
 
  logger.debug('Starting scheduler');
 
- this.interval = setInterval(async () =>{
+ this.interval = setIntervalAsync(async () =>{
  const thresholdDate = new Date(Date.now() - this.settings.experiment_userDrivenPermissionSyncIntervalMs);
 
  const accounts = await this.db.account.findMany({

packages/backend/src/ee/repoPermissionSyncer.ts

@@ -8,7 +8,7 @@ import{PERMISSION_SYNC_SUPPORTED_CODE_HOST_TYPES } from "../constants.js"
 import{createOctokitFromToken, getRepoCollaborators, GITHUB_CLOUD_HOSTNAME } from "../github.js"
 import{createGitLabFromPersonalAccessToken, getProjectMembers } from "../gitlab.js"
 import{Settings } from "../types.js"
-import{getAuthCredentialsForRepo } from "../utils.js"
+import{getAuthCredentialsForRepo, setIntervalAsync } from "../utils.js"
 
 type RepoPermissionSyncJob ={
  jobId: string;
@@ -48,7 +48,7 @@ export class RepoPermissionSyncer{
 
  logger.debug('Starting scheduler');
 
- this.interval = setInterval(async () =>{
+ this.interval = setIntervalAsync(async () =>{
  // @todo: make this configurable
  const thresholdDate = new Date(Date.now() - this.settings.experiment_repoDrivenPermissionSyncIntervalMs);
 

packages/backend/src/repoIndexManager.ts

@@ -12,7 +12,7 @@ import{cloneRepository, fetchRepository, getBranches, getCommitHashForRefName,
 import{captureEvent } from './posthog.js'
 import{PromClient } from './promClient.js'
 import{RepoWithConnections, Settings } from "./types.js"
-import{getAuthCredentialsForRepo, getRepoPath, getShardPrefix, groupmqLifecycleExceptionWrapper, measure } from './utils.js'
+import{getAuthCredentialsForRepo, getRepoPath, getShardPrefix, groupmqLifecycleExceptionWrapper, measure, setIntervalAsync } from './utils.js'
 import{indexGitRepository } from './zoekt.js'
 
 const LOG_TAG = 'repo-index-manager'
@@ -72,9 +72,9 @@ export class RepoIndexManager{
  this.worker.on('error', this.onWorkerError.bind(this));
  }
 
- public async startScheduler(){
+ public startScheduler(){
  logger.debug('Starting scheduler');
- this.interval = setInterval(async () =>{
+ this.interval = setIntervalAsync(async () =>{
  await this.scheduleIndexJobs();
  await this.scheduleCleanupJobs();
  }, this.settings.reindexRepoPollingIntervalMs);

packages/backend/src/utils.ts

@@ -268,3 +268,27 @@ export const groupmqLifecycleExceptionWrapper = async (name: string, logger: Log
  }
 }
 
+
+// setInterval wrapper that ensures async callbacks are not executed concurrently.
+// @see: https://mottaquikarim.github.io/dev/posts/setinterval-that-blocks-on-await/
+export const setIntervalAsync = (target: () => Promise<void>, pollingIntervalMs: number): NodeJS.Timeout =>{
+ const setIntervalWithPromise = <T extends (...args: any[]) => Promise<any>>(
+ target: T
+ ): (...args: Parameters<T>) => Promise<void> =>{
+ return async function (...args: Parameters<T>): Promise<void>{
+ if ((target as any).isRunning) return;
+
+ (target as any).isRunning = true;
+ try{
+ await target(...args);
+ } finally{
+ (target as any).isRunning = false;
+ }
+ };
+ }
+
+ return setInterval(
+ setIntervalWithPromise(target),
+ pollingIntervalMs
+ );
+}