Ensure only a single instance performs concurrency sweeping by using redis-worker cron jobs

Author: ericallam

apps/webapp/app/env.server.ts

@@ -428,11 +428,10 @@ const EnvironmentSchema = z.object({
   RUN_ENGINE_PROCESS_WORKER_QUEUE_DEBOUNCE_MS: z.coerce.number().int().default(200),
   RUN_ENGINE_DEQUEUE_BLOCKING_TIMEOUT_SECONDS: z.coerce.number().int().default(10),
   RUN_ENGINE_MASTER_QUEUE_CONSUMERS_INTERVAL_MS: z.coerce.number().int().default(500),
-  RUN_ENGINE_CONCURRENCY_SWEEPER_SCAN_INTERVAL_MS: z.coerce.number().int().default(60_000),
-  RUN_ENGINE_CONCURRENCY_SWEEPER_PROCESS_MARKED_INTERVAL_MS: z.coerce.number().int().default(5_000),
-  RUN_ENGINE_CONCURRENCY_SWEEPER_LOG_LEVEL: z
-    .enum(["log", "error", "warn", "info", "debug"])
-    .default("info"),
+  RUN_ENGINE_CONCURRENCY_SWEEPER_SCAN_SCHEDULE: z.string().optional(),
+  RUN_ENGINE_CONCURRENCY_SWEEPER_PROCESS_MARKED_SCHEDULE: z.string().optional(),
+  RUN_ENGINE_CONCURRENCY_SWEEPER_SCAN_JITTER: z.coerce.number().int().optional(),
+  RUN_ENGINE_CONCURRENCY_SWEEPER_PROCESS_MARKED_JITTER: z.coerce.number().int().optional(),
 
   RUN_ENGINE_RUN_LOCK_DURATION: z.coerce.number().int().default(5000),
   RUN_ENGINE_RUN_LOCK_AUTOMATIC_EXTENSION_THRESHOLD: z.coerce.number().int().default(1000),

apps/webapp/app/v3/runEngine.server.ts

@@ -67,9 +67,10 @@ function createRunEngine() {
       masterQueueConsumersIntervalMs: env.RUN_ENGINE_MASTER_QUEUE_CONSUMERS_INTERVAL_MS,
       masterQueueConsumersDisabled: env.RUN_ENGINE_WORKER_ENABLED === "0",
       concurrencySweeper: {
-        scanIntervalMs: env.RUN_ENGINE_CONCURRENCY_SWEEPER_SCAN_INTERVAL_MS,
-        processMarkedIntervalMs: env.RUN_ENGINE_CONCURRENCY_SWEEPER_PROCESS_MARKED_INTERVAL_MS,
-        logLevel: env.RUN_ENGINE_CONCURRENCY_SWEEPER_LOG_LEVEL,
+        scanSchedule: env.RUN_ENGINE_CONCURRENCY_SWEEPER_SCAN_SCHEDULE,
+        processMarkedSchedule: env.RUN_ENGINE_CONCURRENCY_SWEEPER_PROCESS_MARKED_SCHEDULE,
+        scanJitter: env.RUN_ENGINE_CONCURRENCY_SWEEPER_SCAN_JITTER,
+        processMarkedJitter: env.RUN_ENGINE_CONCURRENCY_SWEEPER_PROCESS_MARKED_JITTER,
       },
     },
     runLock: {

internal-packages/run-engine/src/engine/index.ts

@@ -138,11 +138,10 @@ export class RunEngine {
         shutdownTimeoutMs: options.worker.shutdownTimeoutMs,
       },
       concurrencySweeper: {
-        enabled: !options.worker.disabled,
-        scanIntervalMs: options.queue?.concurrencySweeper?.scanIntervalMs ?? 60_000,
-        processMarkedIntervalMs:
-          options.queue?.concurrencySweeper?.processMarkedIntervalMs ?? 5_000,
-        logLevel: options.queue?.concurrencySweeper?.logLevel ?? options.queue?.logLevel,
+        scanSchedule: options.queue?.concurrencySweeper?.scanSchedule,
+        processMarkedSchedule: options.queue?.concurrencySweeper?.processMarkedSchedule,
+        scanJitter: options.queue?.concurrencySweeper?.scanJitter,
+        processMarkedJitter: options.queue?.concurrencySweeper?.processMarkedJitter,
         callback: this.#concurrencySweeperCallback.bind(this),
       },
       shardCount: options.queue?.shardCount,

internal-packages/run-engine/src/engine/types.ts

@@ -46,9 +46,10 @@ export type RunEngineOptions = {
     >;
     dequeueBlockingTimeoutSeconds?: number;
     concurrencySweeper?: {
-      scanIntervalMs?: number;
-      processMarkedIntervalMs?: number;
-      logLevel?: LogLevel;
+      scanSchedule?: string;
+      processMarkedSchedule?: string;
+      scanJitter?: number;
+      processMarkedJitter?: number;
     };
   };
   runLock: {