Ensure master queue consumers cannot stop from a processing error, and make the consumer interval configurable via an env var

Author: ericallam

apps/webapp/app/env.server.ts

@@ -469,6 +469,7 @@ const EnvironmentSchema = z.object({
   RUN_ENGINE_RETRY_WARM_START_THRESHOLD_MS: z.coerce.number().int().default(30_000),
   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_WORKER_REDIS_HOST: z
     .string()

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

@@ -62,6 +62,7 @@ function createRunEngine() {
       shardCount: env.RUN_ENGINE_RUN_QUEUE_SHARD_COUNT,
       processWorkerQueueDebounceMs: env.RUN_ENGINE_PROCESS_WORKER_QUEUE_DEBOUNCE_MS,
       dequeueBlockingTimeoutSeconds: env.RUN_ENGINE_DEQUEUE_BLOCKING_TIMEOUT_SECONDS,
+      masterQueueConsumersIntervalMs: env.RUN_ENGINE_MASTER_QUEUE_CONSUMERS_INTERVAL_MS,
     },
     runLock: {
       redis: {

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

@@ -124,6 +124,7 @@ export class RunEngine {
       },
       shardCount: options.queue?.shardCount,
       masterQueueConsumersDisabled: options.worker.disabled,
+      masterQueueConsumersIntervalMs: options.queue?.masterQueueConsumersIntervalMs,
       processWorkerQueueDebounceMs: options.queue?.processWorkerQueueDebounceMs,
       dequeueBlockingTimeoutSeconds: options.queue?.dequeueBlockingTimeoutSeconds,
       meter: options.meter,

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

@@ -32,6 +32,7 @@ export type RunEngineOptions = {
     redis: RedisOptions;
     shardCount?: number;
     processWorkerQueueDebounceMs?: number;
+    masterQueueConsumersIntervalMs?: number;
     workerOptions?: WorkerConcurrencyOptions;
     retryOptions?: RetryOptions;
     defaultEnvConcurrency?: number;

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

@@ -67,6 +67,7 @@ export type RunQueueOptions = {
   retryOptions?: RetryOptions;
   shardCount?: number;
   masterQueueConsumersDisabled?: boolean;
+  masterQueueConsumersIntervalMs?: number;
   processWorkerQueueDebounceMs?: number;
   workerOptions?: {
     pollIntervalMs?: number;
@@ -1001,7 +1002,9 @@ export class RunQueue {
     const consumerId = nanoid();
 
     try {
-      for await (const _ of setInterval(500, null, { signal: this.abortController.signal })) {
+      for await (const _ of setInterval(this.options.masterQueueConsumersIntervalMs ?? 500, null, {
+        signal: this.abortController.signal,
+      })) {
         this.logger.verbose(`Processing master queue shard ${shard}`, {
           processedCount,
           lastProcessedAt,
@@ -1012,7 +1015,18 @@ export class RunQueue {
 
         const now = performance.now();
 
-        const results = await this.#processMasterQueueShard(shard, consumerId);
+        const [error, results] = await tryCatch(this.#processMasterQueueShard(shard, consumerId));
+
+        if (error) {
+          this.logger.error(`Failed to process master queue shard ${shard}`, {
+            error,
+            service: this.name,
+            shard,
+            consumerId,
+          });
+
+          continue;
+        }
 
         const duration = performance.now() - now;