Handle redis errors more gracefully

Author: nicktrn

apps/webapp/app/v3/marqs/concurrencyMonitor.server.ts

@@ -51,11 +51,24 @@ export class MarqsConcurrencyMonitor {
       processedKeys: 0,
     };
 
-    const { stream, redis } = this.marqs.queueConcurrencyScanStream(10, () => {
-      this._logger.debug("[MarqsConcurrencyMonitor] stream closed", {
-        stats,
-      });
-    });
+    const { stream, redis } = this.marqs.queueConcurrencyScanStream(
+      10,
+      () => {
+        this._logger.debug("[MarqsConcurrencyMonitor] stream closed", {
+          stats,
+        });
+      },
+      (error) => {
+        this._logger.debug("[MarqsConcurrencyMonitor] stream error", {
+          stats,
+          error: {
+            name: error.name,
+            message: error.message,
+            stack: error.stack,
+          },
+        });
+      }
+    );
 
     stream.on("data", async (keys) => {
       stream.pause();
@@ -80,9 +93,11 @@ export class MarqsConcurrencyMonitor {
 
       stats.processedKeys += uniqueKeys.length;
 
-      await Promise.all(uniqueKeys.map((key) => this.#processKey(key, redis))).finally(() => {
-        stream.resume();
-      });
+      await Promise.allSettled(uniqueKeys.map((key) => this.#processKey(key, redis))).finally(
+        () => {
+          stream.resume();
+        }
+      );
     });
   }
 
@@ -91,8 +106,20 @@ export class MarqsConcurrencyMonitor {
     const orgKey = this.keys.orgCurrentConcurrencyKeyFromQueue(key);
     const envKey = this.keys.envCurrentConcurrencyKeyFromQueue(key);
 
-    // Next, we need to get all the items from the key, and any parent keys (org, env, queue) using sunion.
-    const runIds = await redis.sunion(orgKey, envKey, key);
+    let runIds: string[] = [];
+
+    try {
+      // Next, we need to get all the items from the key, and any parent keys (org, env, queue) using sunion.
+      runIds = await redis.sunion(orgKey, envKey, key);
+    } catch (e) {
+      this._logger.error("[MarqsConcurrencyMonitor] error during sunion", {
+        key,
+        orgKey,
+        envKey,
+        runIds,
+        error: e,
+      });
+    }
 
     if (runIds.length === 0) {
       return;

apps/webapp/app/v3/marqs/index.server.ts

@@ -788,7 +788,11 @@ export class MarQS {
     }
   }
 
-  queueConcurrencyScanStream(count: number = 100, onEndCallback?: () => void) {
+  queueConcurrencyScanStream(
+    count: number = 100,
+    onEndCallback?: () => void,
+    onErrorCallback?: (error: Error) => void
+  ) {
     const pattern = this.keys.queueCurrentConcurrencyScanPattern();
 
     logger.debug("Starting queue concurrency scan stream", {
@@ -812,6 +816,11 @@ export class MarQS {
       redis.quit();
     });
 
+    stream.on("error", (error) => {
+      onErrorCallback?.(error);
+      redis.quit();
+    });
+
     return { stream, redis };
   }